From a833cd45653911ca792f6f5051fd86aab87993b1 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Mon, 3 Aug 2026 11:03:39 +0000 Subject: [PATCH] fix(sidecar): resume Gmail history past page cap when pages are deduped (#667) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When Gmail history spans more than MAX_HISTORY_PAGES (3) and pages 1–3 are fully deduped, truncated=true froze the cursor forever and page 4+ messages were never ingested — silent data loss during mail bursts. Persist history_page_token in sidecar_state (migration 32) and pass it to listNewMessageIds so the next poll resumes from the dangling pageToken instead of re-fetching already-deduped pages. Co-authored-by: schmug --- test/durableObject/sidecar-state.test.ts | 1 + tests/providers/gmail-client.test.ts | 2 +- tests/providers/workspace-poll.test.ts | 67 +++++++++++++++++++++++- workers/durableObject/migrations.ts | 8 +++ workers/durableObject/sidecar-state.ts | 4 +- workers/providers/gmail-client.ts | 25 +++++++-- workers/providers/workspace.ts | 28 ++++++++-- 7 files changed, 123 insertions(+), 12 deletions(-) diff --git a/test/durableObject/sidecar-state.test.ts b/test/durableObject/sidecar-state.test.ts index 3ce1d46af..6a25f0df5 100644 --- a/test/durableObject/sidecar-state.test.ts +++ b/test/durableObject/sidecar-state.test.ts @@ -39,6 +39,7 @@ function makeSqlLike(): SqlLike { CREATE TABLE sidecar_state ( id INTEGER PRIMARY KEY CHECK (id = 1), history_cursor TEXT, + history_page_token TEXT, access_token TEXT, token_expires_at INTEGER, label_ids TEXT, diff --git a/tests/providers/gmail-client.test.ts b/tests/providers/gmail-client.test.ts index 6a9de2901..d415f1359 100644 --- a/tests/providers/gmail-client.test.ts +++ b/tests/providers/gmail-client.test.ts @@ -153,7 +153,7 @@ describe("listNewMessageIds", () => { })); const { listNewMessageIds } = await import("../../workers/providers/gmail-client"); const r = await listNewMessageIds("tok", "100"); - expect(r).toEqual({ ok: true, messageIds: ["m1", "m2", "m3"], historyId: "300", truncated: true }); + expect(r).toEqual({ ok: true, messageIds: ["m1", "m2", "m3"], historyId: "300", truncated: true, nextPageToken: "p3" }); expect(pages).toBe(3); // MAX_HISTORY_PAGES — the 4th page is never fetched }); diff --git a/tests/providers/workspace-poll.test.ts b/tests/providers/workspace-poll.test.ts index 1e3258292..da2547067 100644 --- a/tests/providers/workspace-poll.test.ts +++ b/tests/providers/workspace-poll.test.ts @@ -64,9 +64,10 @@ function makeEnv(stub: ReturnType) { const ctx = { waitUntil: vi.fn() } as never; /** Cached-token state so no token minting happens in most tests. */ -function freshState(cursor: string | null) { +function freshState(cursor: string | null, pageToken: string | null = null) { return { history_cursor: cursor, + history_page_token: pageToken, access_token: "cached-tok", token_expires_at: Date.now() + 3600_000, label_ids: null, @@ -666,4 +667,68 @@ describe("pollWorkspaceMailbox", () => { expect(r.processed).toBeGreaterThan(0); expect(stub.putSidecarState.mock.calls.at(-1)![0].history_cursor).toBeUndefined(); }); + + it("truncated history with all pages deduped saves history_page_token to resume past the cap (#667)", async () => { + // Deadlock scenario: pages 1–3 fully deduped, page 4+ never fetched. + // Poll 1: truncated listing, all ids deduped → save resume token. + // Poll 2: resume from saved token, tail id ingested, cursor advances. + const stub = makeStub(freshState("100")); + mockedReceive.mockResolvedValue({ messageId: "local-tail", verdict: null }); + let historyCalls = 0; + gmailFetch({ + "/history": (u) => { + historyCalls += 1; + const pageToken = u.searchParams.get("pageToken"); + if (!pageToken) { + return new Response(JSON.stringify({ + historyId: "200", + nextPageToken: "p2", + history: [{ messagesAdded: [{ message: { id: "g1", labelIds: ["INBOX"] } }] }], + }), { status: 200 }); + } + if (pageToken === "p2") { + return new Response(JSON.stringify({ + historyId: "250", + nextPageToken: "p3", + history: [{ messagesAdded: [{ message: { id: "g2", labelIds: ["INBOX"] } }] }], + }), { status: 200 }); + } + if (pageToken === "p3") { + return new Response(JSON.stringify({ + historyId: "280", + nextPageToken: "p4", + history: [{ messagesAdded: [{ message: { id: "g3", labelIds: ["INBOX"] } }] }], + }), { status: 200 }); + } + if (pageToken === "p4") { + return new Response(JSON.stringify({ + historyId: "300", + history: [{ messagesAdded: [{ message: { id: "g-tail", labelIds: ["INBOX"] } }] }], + }), { status: 200 }); + } + throw new Error(`unexpected pageToken ${pageToken}`); + }, + "/messages/": (u) => { + const id = u.pathname.split("/").pop()!; + return new Response(JSON.stringify({ id, raw: rawMessage(`${id}@x`, "s") }), { status: 200 }); + }, + }); + stub.findEmailIdByMessageId.mockImplementation(async (msgId: string) => + ["g1@x", "g2@x", "g3@x"].includes(msgId) ? `existing-${msgId}` : null, + ); + + const r1 = await pollWorkspaceMailbox(makeEnv(stub), ctx, "user@tenant.example", CFG); + expect(r1).toEqual({ processed: 0, deduped: 3, error: null }); + const patch1 = stub.putSidecarState.mock.calls.at(-1)![0]; + expect(patch1.history_cursor).toBeUndefined(); + expect(patch1.history_page_token).toBe("p4"); + + stub.getSidecarState.mockResolvedValue(freshState("100", "p4")); + const r2 = await pollWorkspaceMailbox(makeEnv(stub), ctx, "user@tenant.example", CFG); + expect(r2).toEqual({ processed: 1, deduped: 0, error: null }); + expect(historyCalls).toBe(4); // 3 pages on poll 1 + 1 resumed page on poll 2 + const patch2 = stub.putSidecarState.mock.calls.at(-1)![0]; + expect(patch2.history_cursor).toBe("300"); + expect(patch2.history_page_token).toBeNull(); + }); }); diff --git a/workers/durableObject/migrations.ts b/workers/durableObject/migrations.ts index 8103295db..9443f7803 100644 --- a/workers/durableObject/migrations.ts +++ b/workers/durableObject/migrations.ts @@ -656,6 +656,14 @@ export const mailboxMigrations: Migration[] = [ CREATE INDEX IF NOT EXISTS idx_sidecar_audit_gmail_message_id ON sidecar_audit(gmail_message_id); `, }, + { + // History page-cap resume (#667): when Gmail history spans more than + // MAX_HISTORY_PAGES and pages 1–3 are fully deduped, the cursor stays + // frozen (truncated=true) and the poller never reaches page 4+. + // Persist the dangling pageToken so the next tick resumes past the cap. + name: "32_history_page_token", + sql: `ALTER TABLE sidecar_state ADD COLUMN history_page_token TEXT;`, + }, ]; /** diff --git a/workers/durableObject/sidecar-state.ts b/workers/durableObject/sidecar-state.ts index 92566d972..c6b8543af 100644 --- a/workers/durableObject/sidecar-state.ts +++ b/workers/durableObject/sidecar-state.ts @@ -21,6 +21,8 @@ import type { SqlLike } from "./catchall-intel"; export interface SidecarStateRow { history_cursor: string | null; + /** Gmail history.list pageToken resume point when the page cap truncates a listing. */ + history_page_token: string | null; access_token: string | null; token_expires_at: number | null; label_ids: string | null; @@ -37,7 +39,7 @@ export interface SidecarStateRow { } const STATE_COLUMNS = [ - "history_cursor", "access_token", "token_expires_at", + "history_cursor", "history_page_token", "access_token", "token_expires_at", "label_ids", "last_poll_at", "last_error", "consecutive_failures", "poll_lease_until", "label_error", "label_failure_count", ] as const; diff --git a/workers/providers/gmail-client.ts b/workers/providers/gmail-client.ts index e440003f7..437b6d37b 100644 --- a/workers/providers/gmail-client.ts +++ b/workers/providers/gmail-client.ts @@ -140,7 +140,7 @@ export async function getProfile(token: string): Promise<{ emailAddress: string; } export type HistoryResult = - | { ok: true; messageIds: string[]; historyId: string; truncated: boolean } + | { ok: true; messageIds: string[]; historyId: string; truncated: boolean; nextPageToken?: string } | { ok: false; expired: true }; /** Gmail-internal labels that mark non-inbound messages we must never score. */ @@ -152,15 +152,21 @@ const MAX_HISTORY_PAGES = 3; * cursor is older than Gmail's history retention — the caller must * re-initialize from getProfile() and accept the gap. */ -export async function listNewMessageIds(token: string, startHistoryId: string): Promise { +export async function listNewMessageIds( + token: string, + startHistoryId: string, + resumePageToken?: string | null, +): Promise { const ids: string[] = []; const seen = new Set(); let latestHistoryId = startHistoryId; - let pageToken: string | undefined; + let pageToken: string | undefined = resumePageToken ?? undefined; // truncated = we stopped on the page cap with more pages still pending, so // the listing is INCOMPLETE. The caller must NOT advance the cursor on a // truncated result — otherwise the un-fetched tail is skipped forever. + // When truncated, `nextPageToken` is the resume point for the next fetch. let truncated = false; + let nextPageToken: string | undefined; for (let page = 0; page < MAX_HISTORY_PAGES; page++) { const qs = new URLSearchParams({ startHistoryId, @@ -190,9 +196,18 @@ export async function listNewMessageIds(token: string, startHistoryId: string): pageToken = data.nextPageToken; // If this was the last iteration the loop allows but a page still // dangles, the listing is truncated by the page cap. - if (page === MAX_HISTORY_PAGES - 1) truncated = true; + if (page === MAX_HISTORY_PAGES - 1) { + truncated = true; + nextPageToken = data.nextPageToken; + } } - return { ok: true, messageIds: ids, historyId: latestHistoryId, truncated }; + return { + ok: true, + messageIds: ids, + historyId: latestHistoryId, + truncated, + ...(nextPageToken ? { nextPageToken } : {}), + }; } export async function getRawMessage(token: string, id: string): Promise { diff --git a/workers/providers/workspace.ts b/workers/providers/workspace.ts index 7fff92162..3e1d533b7 100644 --- a/workers/providers/workspace.ts +++ b/workers/providers/workspace.ts @@ -149,7 +149,8 @@ export interface PollResult { processed: number; deduped: number; error: string interface SidecarStub { getSidecarState(): Promise<{ - history_cursor: string | null; access_token: string | null; token_expires_at: number | null; + history_cursor: string | null; history_page_token: string | null; + access_token: string | null; token_expires_at: number | null; label_ids: string | null; last_poll_at: number | null; last_error: string | null; consecutive_failures: number; poll_lease_until: number | null; label_error: string | null; label_failure_count: number; } | null>; @@ -192,7 +193,8 @@ export async function pollWorkspaceMailbox( ): Promise { const stub = env.MAILBOX.get(env.MAILBOX.idFromName(mailboxId)) as unknown as SidecarStub; const state = (await stub.getSidecarState()) ?? { - history_cursor: null, access_token: null, token_expires_at: null, + history_cursor: null, history_page_token: null, + access_token: null, token_expires_at: null, label_ids: null, last_poll_at: null, last_error: null, consecutive_failures: 0, poll_lease_until: null, label_error: null, label_failure_count: 0, }; @@ -253,7 +255,11 @@ export async function pollWorkspaceMailbox( return { processed: 0, deduped: 0, error: null }; } - const history = await listNewMessageIds(token, state.history_cursor); + const history = await listNewMessageIds( + token, + state.history_cursor, + state.history_page_token, + ); if (!history.ok) { // Rule 5: cursor older than Gmail's history retention. Re-anchor and // record the gap — informational, NOT a failure (failures gate backoff). @@ -272,6 +278,7 @@ export async function pollWorkspaceMailbox( detail: "cursor expired past Gmail history retention; mail arriving in the gap was never scored", }); patch.history_cursor = profile.historyId; + patch.history_page_token = null; patch.last_error = "history gap: cursor expired; monitoring reinitialized from current historyId"; await stub.putSidecarState(patch); return { processed: 0, deduped: 0, error: null }; @@ -378,7 +385,20 @@ export async function pollWorkspaceMailbox( // Rule 6: advance the cursor ONLY when the listing was complete (not // page-cap truncated) AND we worked through all of it (not batch-capped). - if (!hitCap && !history.truncated) patch.history_cursor = history.historyId; + // When truncated, persist `history_page_token` once the current window + // is exhausted so the next poll resumes past the page cap instead of + // wedging when pages 1–3 are fully deduped (#667). + if (!hitCap && !history.truncated) { + patch.history_cursor = history.historyId; + patch.history_page_token = null; + } else if ( + !hitCap && + history.truncated && + history.nextPageToken && + (processed + deduped > 0 || history.messageIds.length === 0) + ) { + patch.history_page_token = history.nextPageToken; + } if (labelIds) patch.label_ids = JSON.stringify(labelIds); // Label failures don't freeze the cursor (ingest succeeded — the cursor // rules above stand), but they DO surface in health so a wrong-scope