Do not advance the source checkpoint when a batch is truncated (#1212)

crates/utopia-server/src/ingest_sources.rs

@@ -38,6 +38,13 @@ pub struct SyncStats {
     pub discovered: usize,
     pub queued_for_content: usize,
     pub content_terminal: usize,
+    /// 本轮因为触顶而被丢在外面的条目数（`discovered > MAX_NEW_PER_SYNC`）。
+    /// `sync_source` 看到这个标志就把这次跑标为「未完成」：`finish_run` 写
+    /// `failed` 而不是 `ok`，让 `last_successful_sync_start` 沿用上一条成功的。
+    /// 不这么做的话下一轮用 `since = 这一次的 started_at` 重新发，端点按
+    /// `since` 排除已经把 `last_change` 写在那条 started_at 之前的剩余条目，
+    /// 增量游标就悄悄把那一段漏过去（#1212）
+    pub truncated: bool,
 }
 
 #[derive(Debug)]
@@ -101,6 +108,45 @@ pub async fn sync_source(state: &AppState, source_id: Uuid) -> anyhow::Result<()
     };
 
     match outcome {
+        Ok(stats) if stats.truncated => {
+            // #1212：触顶的那一段没读完。把这次跑标为 failed（不是 ok），
+            // `last_successful_sync_start` 不会前进，下一轮用同一条 since 重发
+            // ——剩余条目还在端点里。但这一轮已经 ingest 进来的 created/updated
+            // 仍然在 created_docs/updated_docs 里留下记录，便于审计
+            let err_msg = format!(
+                "items beyond per-sync cap ({} > {}); \
+                 checkpoint not advanced, the next sync will re-fetch this window",
+                stats.discovered, MAX_NEW_PER_SYNC
+            );
+            utopia_store::sources::finish_run(
+                &state.pool,
+                run_id,
+                source_id,
+                Some(&err_msg),
+                stats.created as i32,
+                stats.updated as i32,
+            )
+            .await?;
+            utopia_store::sources::finish_sync(
+                &state.pool,
+                source_id,
+                Some(&err_msg),
+                stats.total() as i32,
+            )
+            .await?;
+            state.emit_source(source.kb_id);
+            tracing::warn!(
+                %source_id,
+                kind = %source.kind,
+                created = stats.created,
+                updated = stats.updated,
+                discovered = stats.discovered,
+                queued_for_content = stats.queued_for_content,
+                content_terminal = stats.content_terminal,
+                "来源同步触顶,checkpoint 未前进,下一轮重试同一窗口"
+            );
+            Ok(())
+        }
         Ok(stats) => {
             utopia_store::sources::finish_run(
                 &state.pool,
@@ -983,6 +1029,15 @@ async fn sync_custom(
         .ok_or_else(|| anyhow::anyhow!("Response is missing the items[] array"))?;
 
     let mut stats = SyncStats::default();
+    // #1212：端点返回的条目数 ≥ MAX_NEW_PER_SYNC 时，触顶的那条没法判断
+    // 是「正好 N」还是「N+ 还没到」。截断而不告诉调用方的话，下一轮的
+    // `since` 会把 `last_change` 早于本轮 started_at 的剩余条目过滤掉，
+    // 增量游标就悄悄把那一段漏过去。把 truncated 标上，让 `sync_source`
+    // 把这次跑标为 failed（`last_successful_sync_start` 不前进），下一轮
+    // 用同一 since 重新拉——剩余的还在
+    if items.len() > MAX_NEW_PER_SYNC {
+        stats.truncated = true;
+    }
     let mut seen_keys: Vec<String> = Vec::new();
     for item in items.iter().take(MAX_NEW_PER_SYNC) {
         let Some(id) = item["id"].as_str().map(str::trim).filter(|s| !s.is_empty()) else {

crates/utopia-server/src/source_checkpoint_tests.rs

@@ -135,3 +135,129 @@ async fn retry_keeps_updates_from_the_previous_successful_run_window() -> anyhow
 async fn updates_during_a_successful_sync_remain_in_the_next_window() -> anyhow::Result<()> {
     retry_reads_the_uncovered_window(true, false).await
 }
+
+/// #1212：端点一次返回的条目数 ≥ `MAX_NEW_PER_SYNC` 时，本轮要标为 `failed` 而不是
+/// `ok` —— `last_successful_sync_start` 不能前进，下一轮用同一 since 重发，剩下
+/// 的条目还在端点里。把这一段钉下来：触顶就该让调度看到、让运维看到，而不是悄悄
+/// 把窗口漏过去
+#[tokio::test]
+async fn a_truncated_batch_marks_the_run_failed_and_holds_the_checkpoint() -> anyhow::Result<()> {
+    let Some(url) = utopia_store::test_db::url() else {
+        return Ok(());
+    };
+    let pool = sqlx::PgPool::connect(&url).await?;
+    let (org, ws, kb, source) = (
+        Uuid::now_v7(),
+        Uuid::now_v7(),
+        Uuid::now_v7(),
+        Uuid::now_v7(),
+    );
+    sqlx::query("INSERT INTO organizations(id,name) VALUES($1,'truncation-test')")
+        .bind(org)
+        .execute(&pool)
+        .await?;
+    sqlx::query("INSERT INTO workspaces(id,org_id,name) VALUES($1,$2,'truncation-test')")
+        .bind(ws)
+        .bind(org)
+        .execute(&pool)
+        .await?;
+    sqlx::query(
+        "INSERT INTO knowledge_bases(id,workspace_id,name) VALUES($1,$2,'truncation-test')",
+    )
+    .bind(kb)
+    .bind(ws)
+    .execute(&pool)
+    .await?;
+
+    // 上一轮成功——它把 last_successful_sync_start 钉在 started，触顶那一轮必须不动它
+    let prior_started = chrono::Utc::now() - chrono::Duration::hours(2);
+    let prior_run = utopia_store::sources::start_run(&pool, source).await?;
+    utopia_store::sources::finish_run(&pool, prior_run, source, None, 0, 0).await?;
+    sqlx::query("UPDATE source_sync_runs SET started_at=$2 WHERE id=$1")
+        .bind(prior_run)
+        .bind(prior_started)
+        .execute(&pool)
+        .await?;
+
+    let server = MockServer::start().await;
+    sqlx::query(
+        "INSERT INTO sources(id,kb_id,kind,name,config) VALUES($1,$2,'custom','fixture',$3)",
+    )
+    .bind(source)
+    .bind(kb)
+    .bind(serde_json::json!({"endpoint":server.uri()}))
+    .execute(&pool)
+    .await?;
+
+    // 触顶：返回 MAX_NEW_PER_SYNC + 1 条，distinct id / distinct content
+    let items: Vec<serde_json::Value> = (0..201u32)
+        .map(|i| {
+            serde_json::json!({
+                "id": format!("item-{i}"),
+                "title": format!("Item {i}"),
+                "content": format!("Body {i}"),
+            })
+        })
+        .collect();
+    wiremock::Mock::given(method("GET"))
+        .respond_with(
+            wiremock::ResponseTemplate::new(200)
+                .set_body_json(serde_json::json!({"items": items})),
+        )
+        .mount(&server)
+        .await;
+
+    let dir = std::env::temp_dir().join(format!("utopia-truncation-{source}"));
+    let cfg = utopia_core::config::AppConfig {
+        data_dir: dir.to_string_lossy().into_owned(),
+        ..Default::default()
+    };
+    let search = Arc::new(utopia_search::SearchIndex::open(&dir.join("search"))?);
+    let state = crate::state::AppState::new(pool.clone(), &cfg, search, "test-only".into());
+
+    let result = async {
+        // 跑 sync_source：返回 Ok，但 `finish_run` 应当以 'failed' 写
+        sync_source(&state, source).await?;
+        // 这一轮 ingest 进来的 created/updated 应该正常记在 sources.last_sync_added
+        // 与 source_sync_runs.created_docs 里——partial work 仍可见
+        let src = utopia_store::sources::get(&pool, source).await?;
+        assert_eq!(
+            src.last_sync_status, "failed",
+            "触顶的那一轮不能记为 'ok'，否则 last_successful_sync_start 会前进、剩余条目被增量游标漏过去"
+        );
+        assert!(
+            src.last_sync_error
+                .as_deref()
+                .unwrap_or("")
+                .contains("beyond per-sync cap"),
+            "last_sync_error 应当说明触顶，operator 看了才知道为什么 failed: {:?}",
+            src.last_sync_error
+        );
+        // 关键：last_successful_sync_start 仍然指 prior_started，没被触顶这一轮
+        // 推进。下一轮用同一 since 重新拉，剩余的 200+ 条还在
+        let after: Option<chrono::DateTime<chrono::Utc>> = sqlx::query_scalar(
+            "SELECT max(started_at) FROM source_sync_runs WHERE source_id = $1 AND status = 'ok'",
+        )
+        .bind(source)
+        .fetch_one(&pool)
+        .await?;
+        let after = after.expect("应该至少有一行 status='ok' 的 prior_started 留下的记录");
+        assert!(
+            (after - prior_started).num_seconds().abs() <= 1,
+            "触顶那一轮不该把 last_successful_sync_start 推过 prior_started: prior={prior_started} after={after}"
+        );
+        Ok::<_, anyhow::Error>(())
+    }
+    .await;
+    sqlx::query("DELETE FROM knowledge_bases WHERE id=$1")
+        .bind(kb)
+        .execute(&pool)
+        .await?;
+    sqlx::query("DELETE FROM organizations WHERE id=$1")
+        .bind(org)
+        .execute(&pool)
+        .await?;
+    drop(state);
+    let _ = std::fs::remove_dir_all(dir);
+    result
+}