From f82a860f7e436c963fc491d70e8d7b05778f63fc Mon Sep 17 00:00:00 2001 From: Royce Lam Date: Wed, 28 Oct 2026 16:13:08 +0800 Subject: [PATCH] The source API says where a ticket source's reading stopped, so an operator sees the checkpoint itself and not only that the round succeeded (#1212, ADR 0075) Signed-off-by: Royce Lam --- crates/utopia-core/src/models.rs | 11 ++ .../src/api/sources_checkpoint_tests.rs | 187 ++++++++++++++++++ .../utopia-server/src/api/sources_routes.rs | 5 + crates/utopia-store/src/sources.rs | 1 + ...t-source-says-where-its-reading-stopped.md | 79 ++++++++ docs/decisions/README.md | 1 + docs/design/sources.md | 6 +- web/src/api.ts | 4 + 8 files changed, 293 insertions(+), 1 deletion(-) create mode 100644 crates/utopia-server/src/api/sources_checkpoint_tests.rs create mode 100644 docs/decisions/0075-a-ticket-source-says-where-its-reading-stopped.md diff --git a/crates/utopia-core/src/models.rs b/crates/utopia-core/src/models.rs index 14e8164eb..257fa8f2b 100644 --- a/crates/utopia-core/src/models.rs +++ b/crates/utopia-core/src/models.rs @@ -158,6 +158,14 @@ pub struct Source { pub last_sync_status: String, pub last_sync_error: Option, pub last_sync_added: i32, + /// 工单来源读到了哪里（ADR 0072）。NULL = 没有游标：首轮同步，或者这一类 + /// 来源不用游标。 + /// + /// **说的是「读到哪」，不是「这一轮什么时候跑」**——后者是 `last_sync_at`。 + /// 一轮读不完是常事（同一个更新时刻上可能挂着一千条工单），游标说的是 + /// 「消费到了哪个更新时刻，且那一刻的整组都读完了」；下一轮从那儿接着读， + /// 而不是把这一轮没读完的跳过去（#1212） + pub sync_checkpoint: Option>, /// api 来源的推送密钥（明文；查看走 Editor 权限的专用端点，列表响应不带） #[serde(skip_serializing)] pub ingest_token: Option, @@ -208,6 +216,7 @@ mod source_extracts_tests { last_sync_status: "never".into(), last_sync_error: None, last_sync_added: 0, + sync_checkpoint: None, ingest_token: None, created_at: chrono::Utc::now(), } @@ -389,6 +398,8 @@ pub struct SourceView { pub last_sync_status: String, pub last_sync_error: Option, pub last_sync_added: i32, + /// 工单来源读到了哪里（ADR 0072）。见 [`Source::sync_checkpoint`] + pub sync_checkpoint: Option>, pub doc_count: i64, /// 已标记"不在来源中"的文档数（url 全集对账 / custom 墓碑产生） pub missing_count: i64, diff --git a/crates/utopia-server/src/api/sources_checkpoint_tests.rs b/crates/utopia-server/src/api/sources_checkpoint_tests.rs new file mode 100644 index 000000000..b68d07222 --- /dev/null +++ b/crates/utopia-server/src/api/sources_checkpoint_tests.rs @@ -0,0 +1,187 @@ +//! 来源的增量游标在 API 里看得见（ADR 0072）。 +//! +//! **说的是「读到哪」，不是「这一轮什么时候跑」。** 一轮读不完是常事——同一个更新 +//! 时刻上可能挂着一千条工单——游标记的是「消费到了哪个更新时刻，且那一刻的整组都 +//! 读完了」。没有它，界面上只看得见「同步成功」，看不见这一轮停在了哪里；要排 +//! 「这一轮到底覆盖了多少」，只能去翻日志。 +//! +//! 用不到的来源那一格是 NULL，而不是 0 也不是别的时刻：没有游标与「游标在零点」 +//! 是两回事，前者是「还没读到哪」，后者是「读到 1970 年」。 +//! +//! 连库的测试，没有 `UTOPIA_DATABASE_URL` 就跳过（同 documents_routes_tests）。 + +use axum::body::{to_bytes, Body}; +use axum::http::{Request, StatusCode}; +use serde_json::{json, Value}; +use std::sync::Arc; +use tower::ServiceExt; +use uuid::Uuid; + +/// 游标落在这一刻：秒精度，正是工单更新时刻的粒度。 +/// +/// **响应里写的是 `Z` 结尾的那一种**（chrono 对 `DateTime` 的写法）， +/// 不是 `+00:00`——断言照着实际发出去的形式写，变了就是变了。 +const CHECKPOINT: &str = "2026-01-02T00:00:00Z"; + +fn checkpoint() -> chrono::DateTime { + chrono::DateTime::parse_from_rfc3339(CHECKPOINT) + .unwrap() + .with_timezone(&chrono::Utc) +} + +struct Fixture { + app: axum::Router, + kb: Uuid, + /// 工单来源：有游标 + ticket: Uuid, + /// 订阅来源：没有游标 + feed: Uuid, + token: String, + _dir: tempfile::TempDir, +} + +impl Fixture { + async fn new() -> anyhow::Result> { + let Some(url) = utopia_store::test_db::url() else { + return Ok(None); + }; + let pool = sqlx::PgPool::connect(&url).await?; + utopia_store::db::migrate(&pool).await?; + let (org, ws, kb, user, ticket, feed) = ( + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + ); + // Only locally generated UUIDs are interpolated into fixture SQL. + sqlx::raw_sql(&format!( + "INSERT INTO organizations(id,name) VALUES ('{org}','checkpoint-test'); + INSERT INTO workspaces(id,org_id,name) VALUES ('{ws}','{org}','checkpoint-test'); + INSERT INTO users(id,org_id,email,display_name,password_hash) + VALUES ('{user}','{org}','{user}@checkpoint.test','checkpoint-test','unused'); + INSERT INTO knowledge_bases(id,workspace_id,name) VALUES ('{kb}','{ws}','observations'); + INSERT INTO kb_members(kb_id,user_id,role) VALUES ('{kb}','{user}','editor'); + INSERT INTO sources(id,kb_id,kind,name,config) + VALUES ('{ticket}','{kb}','github_issues','tickets','{{\"repo\":\"o/r\"}}'); + INSERT INTO sources(id,kb_id,kind,name,config) + VALUES ('{feed}','{kb}','rss','feed','{{\"feed_url\":\"https://example.test/f\"}}');" + )) + .execute(&pool) + .await?; + // 工单来源读到了那一刻；订阅来源不按时刻分组，没有这一格 + sqlx::query("UPDATE sources SET sync_checkpoint = $2 WHERE id = $1") + .bind(ticket) + .bind(checkpoint()) + .execute(&pool) + .await?; + let dir = tempfile::tempdir()?; + let cfg = utopia_core::config::AppConfig { + data_dir: dir.path().to_string_lossy().into_owned(), + ..Default::default() + }; + let search = Arc::new(utopia_search::SearchIndex::open( + &dir.path().join("search"), + )?); + let state = crate::state::AppState::new(pool.clone(), &cfg, search, "test-only".into()); + let token = crate::auth::issue_token(&state, user)?; + let app = super::super::router(state.clone(), &cfg); + Ok(Some(Self { + app, + kb, + ticket, + feed, + token, + _dir: dir, + })) + } + + async fn call( + &self, + method: &str, + uri: &str, + body: Option, + ) -> anyhow::Result<(StatusCode, Value)> { + let request = Request::builder() + .method(method) + .uri(uri) + .header("Authorization", format!("Bearer {}", self.token)); + let request = match body { + Some(body) => request + .header("Content-Type", "application/json") + .body(Body::from(body.to_string()))?, + None => request.body(Body::empty())?, + }; + let response = self.app.clone().oneshot(request).await?; + let status = response.status(); + let bytes = to_bytes(response.into_body(), 1 << 20).await?; + Ok(( + status, + serde_json::from_slice(&bytes).unwrap_or(Value::Null), + )) + } +} + +/// **列表里有这一格。** 工单来源带着游标，用不到游标的来源是 NULL。 +#[tokio::test] +async fn the_source_list_carries_the_checkpoint() -> anyhow::Result<()> { + let Some(fx) = Fixture::new().await? else { + return Ok(()); + }; + let (status, body) = fx + .call("GET", &format!("/api/v1/kbs/{}/sources", fx.kb), None) + .await?; + assert_eq!(status, StatusCode::OK, "{body}"); + let sources = body["sources"].as_array().expect("sources[]"); + let ticket = sources + .iter() + .find(|s| s["id"] == fx.ticket.to_string()) + .expect("工单来源在列表里"); + let feed = sources + .iter() + .find(|s| s["id"] == fx.feed.to_string()) + .expect("订阅来源在列表里"); + + // **时刻本身要给出来**，不是"有几个"：那一格说的是读到哪 + assert_eq!( + ticket["sync_checkpoint"], + json!(CHECKPOINT), + "工单来源的游标要原样给出" + ); + // 用不到的来源是 NULL——没有游标与"游标在零点"是两回事 + assert_eq!( + feed["sync_checkpoint"], + Value::Null, + "用不到游标的来源不该编一个时刻出来" + ); + // 游标与"这一轮什么时候跑"是两件事，两个都在 + assert!( + ticket["last_sync_at"].is_null(), + "还没同步过就没有这一轮的时刻" + ); + Ok(()) +} + +/// **改一个来源，回来的那一格也带着游标。** 列表与详情说的是同一件事， +/// 不能一处有一处没有。 +#[tokio::test] +async fn a_source_update_carries_the_checkpoint_too() -> anyhow::Result<()> { + let Some(fx) = Fixture::new().await? else { + return Ok(()); + }; + let (status, body) = fx + .call( + "PATCH", + &format!("/api/v1/kbs/{}/sources/{}", fx.kb, fx.ticket), + Some(json!({ "name": "renamed" })), + ) + .await?; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!( + body["source"]["sync_checkpoint"], + json!(CHECKPOINT), + "更新回来的那一格也要带游标" + ); + Ok(()) +} diff --git a/crates/utopia-server/src/api/sources_routes.rs b/crates/utopia-server/src/api/sources_routes.rs index bcb62b728..9b220ec56 100644 --- a/crates/utopia-server/src/api/sources_routes.rs +++ b/crates/utopia-server/src/api/sources_routes.rs @@ -335,6 +335,10 @@ pub async fn cleanup_missing( #[path = "sources_cleanup_tests.rs"] mod cleanup_tests; +#[cfg(test)] +#[path = "sources_checkpoint_tests.rs"] +mod checkpoint_tests; + pub async fn delete( State(state): State, AuthUser(user): AuthUser, @@ -965,6 +969,7 @@ mod tests { last_sync_status: "never".into(), last_sync_error: None, last_sync_added: 0, + sync_checkpoint: None, ingest_token: Some("utp_x".into()), created_at: chrono::Utc::now(), }; diff --git a/crates/utopia-store/src/sources.rs b/crates/utopia-store/src/sources.rs index 29e4d7d9a..14c1fe9c4 100644 --- a/crates/utopia-store/src/sources.rs +++ b/crates/utopia-store/src/sources.rs @@ -92,6 +92,7 @@ pub async fn list(pool: &PgPool, kb_id: Uuid) -> AppResult> { listed_source.sync_interval_minutes, listed_source.sync_cron, listed_source.last_sync_at, listed_source.last_sync_status, listed_source.last_sync_error, listed_source.last_sync_added, + listed_source.sync_checkpoint, (SELECT count(*) FROM documents d WHERE d.source_id = listed_source.id AND d.deleted_at IS NULL) AS doc_count, (SELECT count(*) FROM documents d diff --git a/docs/decisions/0075-a-ticket-source-says-where-its-reading-stopped.md b/docs/decisions/0075-a-ticket-source-says-where-its-reading-stopped.md new file mode 100644 index 000000000..6366c1ce7 --- /dev/null +++ b/docs/decisions/0075-a-ticket-source-says-where-its-reading-stopped.md @@ -0,0 +1,79 @@ +# 0075 · A ticket source says in the API where its reading stopped + +- **Status**: Implemented · 2026-10-28 · no migration (the column came with 0119) · open: the interface does not read it yet +- **Written**: 2026-10-28 +- **Related**: [0072](0072-a-source-checkpoint-ends-at-a-timestamp-boundary.md) (what the checkpoint is and how it moves); [#1212](https://github.com/deeplethe/utopia/issues/1212) (the tickets a round used to skip); [0033](0033-rss-source-summaries-are-source-scoped.md) (the RSS summary — a different shape for a different question) + +## Problem + +A ticket source stops at its per-round budget all the time: 200 tickets, or ten +listing pages, whichever comes first. A round that stops there is a **success** — +it consumed a prefix, and the next round continues from where it stopped (0072). +The source list then says "synced" and says how many documents the round added. +It does not say *where* the reading stopped. + +The one thing that says where is the checkpoint, and it lived on the source row +with nothing carrying it out of the API. So an operator asking "did this source +get everything?" had nothing to read: the run history says how many documents +each round added, not how far into the provider's timeline the reading got. The +answer was in a log line. + +## Decisions + +1. **The source API carries the checkpoint.** `Source.sync_checkpoint` and + `SourceView.sync_checkpoint` hold the update time a ticket source has consumed + through — the time whose whole group was read (0072). The list response carries + it, and so do the create/update responses, so the two shapes of a source never + disagree about the same row. + +2. **It is the raw timestamp, not a state the server computes.** The RSS summary + (0033) is a computed block: the server works out which of the five counts + applies, so the client does not re-derive it from `kind` and `content_mode`. + The checkpoint is not like that. Whether the source has more to give is a + question only the provider can answer, and the server does not ask it between + rounds — a state word would have to say "caught up" or "behind", and the server + cannot know which. The timestamp is the fact; what it means is the reader's. + +3. **NULL means there is no checkpoint.** Not zero, not the epoch, not "now". + NULL says the source has not finished a timestamp group yet (first round), or + that this kind does not read by timestamp at all. It doubles as the + applicability signal: a kind that does not read by timestamp has NULL there, so + the client does not need to know the kinds to know whether the field means + anything. + +4. **It says where the reading stopped, not when the round ran.** `last_sync_at` is + the round's time and exists for every kind. The checkpoint is a position in the + provider's timeline and exists only for the kinds that have one. The two answer + different questions, and both are carried. + +5. **The interface does not read it yet.** Where to put it and what to call it is a + change to the source row, and that row is already dense — status, time, added, + error, endpoint, schedule. A diagnostic that only appears in a tooltip is not + obviously better than one that is in the API. This record settles the shape; the + interface follows when someone wants it. + +## Dead ends + +- **A `state` word beside the timestamp** ("caught up", "behind", "partly read"). + The server cannot know it: the provider is not asked between rounds, so + "caught up" would be a guess. A guess in a field that looks authoritative is how + a wrong answer gets believed — the same reason the RSS summary computes what it + can actually compute instead of guessing the rest. +- **A count of tickets still to read.** The server does not know the provider's + total for a ticket source, and counting one would need a listing the round never + made. +- **Putting the checkpoint in the run history instead.** The runs are a tail, not a + ledger: they are capped and pruned. The position a source is at would disappear + with them. +- **Only the list, and not the source responses.** Then the two shapes of a source + disagree about the same row, and a client that read one and wrote the other + would see the field vanish. +- **A separate endpoint for it.** One more route for one nullable column, and the + list already carries the rest of the source's sync state. + +## Open questions + +- **Where the interface shows it.** The source row is dense; a second timestamp + there needs a place and a wording, and neither is settled. +- **Whether a ticket source should say how much of the provider's window it has + read.** That is a count the server does not have. diff --git a/docs/decisions/README.md b/docs/decisions/README.md index 3ac9cafb3..d0eb83c0d 100644 --- a/docs/decisions/README.md +++ b/docs/decisions/README.md @@ -117,6 +117,7 @@ By domain; the domains are the files of [../design/](../design/README.md). **Sta | 0065 | [The OCR reader chooses its provider](0065-readers-choose-their-provider.md) | Implemented | | | 0072 | [A source checkpoint ends at a timestamp boundary](0072-a-source-checkpoint-ends-at-a-timestamp-boundary.md) | In progress | | | 0074 | [A Google Drive source is connected by a person](0074-a-google-drive-source-is-connected-by-a-person.md) | Implemented | | +| 0075 | [A ticket source says in the API where its reading stopped](0075-a-ticket-source-says-where-its-reading-stopped.md) | Implemented | | ### [lakehouse-and-actions](../design/lakehouse-and-actions.md) diff --git a/docs/design/sources.md b/docs/design/sources.md index 5cd6fca9b..fa97c817a 100644 --- a/docs/design/sources.md +++ b/docs/design/sources.md @@ -43,7 +43,11 @@ request at most three times with waits of at most 30 seconds each; longer hints fail immediately. Later ordinary job retries and source schedules do not preserve the provider's `Retry-After` deadline. Jira requires the search `total` to detect premature pagination endings. Custom batch progression is separate work in #1397 -[0072, revision 2026-10-23]. +[0072, revision 2026-10-23]. The source API carries the checkpoint on the +list and on the create/update responses, so an operator can see where a round +stopped rather than only that it succeeded; a kind that does not read by +timestamp has NULL there, which says "no checkpoint" rather than a time of zero +[0075]. **Statements.** A `statements` source takes the open extraction contract itself (`e`, `s`, `n`) on `POST /sources/{id}/statements`, with the `api` push's identity, versions, tombstones and run history; the payload is the document and its single chunk, and extraction parses the chunk instead of prompting a model, then runs the unchanged path, so a pushed statement is an open statement with the item as its evidence (null offsets) and reaches the typed graph only through alignment. The contract has no slot for a property or a class and unknown keys are refused; a second push under the same identity marks the earlier statements stale rather than closing them; tables belong on a mount, not here [0054]. diff --git a/web/src/api.ts b/web/src/api.ts index be54834d8..a9440025f 100644 --- a/web/src/api.ts +++ b/web/src/api.ts @@ -265,6 +265,10 @@ export interface SourceView { last_sync_status: "never" | "queued" | "running" | "ok" | "failed"; last_sync_error: string | null; last_sync_added: number; + /** 工单来源读到了哪里（ADR 0072）：消费到的更新时刻，且那一刻的整组都读完了。 + * null = 没有游标（首轮同步，或这一类来源不用游标）。说的是「读到哪」， + * 不是 `last_sync_at` 的「这一轮什么时候跑」 */ + sync_checkpoint: string | null; doc_count: number; missing_count: number; /** 全文补全那一块；**不是 RSS 全文来源的就是 null**，不用再拿 kind 与