From b6451e171f796120da267d7181632c66bef18407 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sat, 25 Jul 2026 11:06:33 +0000 Subject: [PATCH] fix(sidecar): resume truncated Gmail history via page token (#595) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When listNewMessageIds hits MAX_HISTORY_PAGES (3) and pages 1-3 are already ingested, every subsequent poll re-listed the same pages and never reached page 4+ — silently skipping new mail for busy mailboxes catching up after downtime. Persist history_page_token in sidecar_state (migration 32) when a truncated listing is fully worked, and resume from that token on the next poll. Clear the token when the listing completes or when a batch cap leaves unprocessed ids in the current slice. Co-authored-by: schmug --- test/durableObject/sidecar-state.test.ts | 3 +- tests/providers/gmail-client.test.ts | 2 +- tests/providers/workspace-poll.test.ts | 59 ++++++++++++++++++++++++ workers/durableObject/migrations.ts | 10 ++++ workers/durableObject/sidecar-state.ts | 3 ++ workers/providers/gmail-client.ts | 24 ++++++++-- workers/providers/workspace.ts | 19 +++++++- 7 files changed, 111 insertions(+), 9 deletions(-) diff --git a/test/durableObject/sidecar-state.test.ts b/test/durableObject/sidecar-state.test.ts index 3ce1d46a..c7e74d65 100644 --- a/test/durableObject/sidecar-state.test.ts +++ b/test/durableObject/sidecar-state.test.ts @@ -47,7 +47,8 @@ function makeSqlLike(): SqlLike { consecutive_failures INTEGER NOT NULL DEFAULT 0, poll_lease_until INTEGER, label_error TEXT, - label_failure_count INTEGER NOT NULL DEFAULT 0 + label_failure_count INTEGER NOT NULL DEFAULT 0, + history_page_token TEXT ); CREATE TABLE sidecar_audit ( diff --git a/tests/providers/gmail-client.test.ts b/tests/providers/gmail-client.test.ts index 6a9de290..d415f135 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 1e325829..11ed237a 100644 --- a/tests/providers/workspace-poll.test.ts +++ b/tests/providers/workspace-poll.test.ts @@ -76,6 +76,7 @@ function freshState(cursor: string | null) { poll_lease_until: null, label_error: null, label_failure_count: 0, + history_page_token: null, }; } @@ -666,4 +667,62 @@ describe("pollWorkspaceMailbox", () => { expect(r.processed).toBeGreaterThan(0); expect(stub.putSidecarState.mock.calls.at(-1)![0].history_cursor).toBeUndefined(); }); + + it("truncated history deadlock (#595): all deduped pages save page token, next poll resumes and advances cursor", async () => { + // Poll 1: pages 1–3 are already ingested (deduped); page 4 has fresh mail. + // Without a continuation token the poller would re-list pages 1–3 forever. + const stub1 = makeStub(freshState("100")); + stub1.findEmailIdByMessageId.mockImplementation(async (mid: string) => + (["g1@x", "g2@x", "g3@x"].includes(mid) ? "already" : null), + ); + let pages1 = 0; + gmailFetch({ + "/history": () => { + pages1 += 1; + return new Response(JSON.stringify({ + historyId: "999", + nextPageToken: `p${pages1}`, + history: [{ messagesAdded: [{ message: { id: `g${pages1}`, labelIds: ["INBOX"] } }] }], + }), { status: 200 }); + }, + "/messages/": (u) => { + const id = u.pathname.split("/").pop()!; + return new Response(JSON.stringify({ id, raw: rawMessage(`${id}@x`, "s") }), { status: 200 }); + }, + }); + const r1 = await pollWorkspaceMailbox(makeEnv(stub1), ctx, "user@tenant.example", CFG); + expect(r1).toMatchObject({ processed: 0, deduped: 3, error: null }); + const patch1 = stub1.putSidecarState.mock.calls.at(-1)![0]; + expect(patch1.history_cursor).toBeUndefined(); + expect(patch1.history_page_token).toBe("p3"); + + // Poll 2: resume from p3 — only g4 is fresh; listing completes. + const stub2 = makeStub({ ...freshState("100"), history_page_token: "p3" }); + stub2.findEmailIdByMessageId.mockImplementation(async (mid: string) => + (["g1@x", "g2@x", "g3@x"].includes(mid) ? "already" : null), + ); + mockedReceive.mockResolvedValue({ messageId: "local-g4", verdict: null }); + let pages2 = 0; + gmailFetch({ + "/history": (u) => { + pages2 += 1; + const token = u.searchParams.get("pageToken"); + expect(token).toBe(pages2 === 1 ? "p3" : `tail${pages2}`); + if (pages2 === 1) { + return new Response(JSON.stringify({ + historyId: "999", + nextPageToken: "tail2", + history: [{ messagesAdded: [{ message: { id: "g4", labelIds: ["INBOX"] } }] }], + }), { status: 200 }); + } + return new Response(JSON.stringify({ historyId: "1000", history: [] }), { status: 200 }); + }, + "/messages/g4": () => new Response(JSON.stringify({ id: "g4", raw: rawMessage("g4@x", "fresh") }), { status: 200 }), + }); + const r2 = await pollWorkspaceMailbox(makeEnv(stub2), ctx, "user@tenant.example", CFG); + expect(r2).toMatchObject({ processed: 1, deduped: 0, error: null }); + const patch2 = stub2.putSidecarState.mock.calls.at(-1)![0]; + expect(patch2.history_page_token).toBeNull(); + expect(patch2.history_cursor).toBe("1000"); + }); }); diff --git a/workers/durableObject/migrations.ts b/workers/durableObject/migrations.ts index 8103295d..43da47c1 100644 --- a/workers/durableObject/migrations.ts +++ b/workers/durableObject/migrations.ts @@ -656,6 +656,16 @@ export const mailboxMigrations: Migration[] = [ CREATE INDEX IF NOT EXISTS idx_sidecar_audit_gmail_message_id ON sidecar_audit(gmail_message_id); `, }, + { + // Gmail history page continuation (issue #595). `listNewMessageIds` caps + // at MAX_HISTORY_PAGES; when the listing is truncated the cursor must NOT + // advance, but if pages 1–3 are already deduped the poller would otherwise + // re-fetch them forever and never reach page 4+. `history_page_token` + // stores the dangling `nextPageToken` once the current slice is fully + // worked; the next poll resumes from that token with the same cursor. + 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 92566d97..0177f6e9 100644 --- a/workers/durableObject/sidecar-state.ts +++ b/workers/durableObject/sidecar-state.ts @@ -34,12 +34,15 @@ export interface SidecarStateRow { label_error: string | null; /** Failed label writes accumulated since the last successful one (#590). */ label_failure_count: number; + /** Gmail history `nextPageToken` when a listing hit the page cap (#595). */ + history_page_token: string | null; } const STATE_COLUMNS = [ "history_cursor", "access_token", "token_expires_at", "label_ids", "last_poll_at", "last_error", "consecutive_failures", "poll_lease_until", "label_error", "label_failure_count", + "history_page_token", ] as const; export function _getSidecarStateImpl(sql: SqlLike): SidecarStateRow | null { diff --git a/workers/providers/gmail-client.ts b/workers/providers/gmail-client.ts index e440003f..1f555647 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,20 @@ 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, + opts?: { startPageToken?: string }, +): Promise { const ids: string[] = []; const seen = new Set(); let latestHistoryId = startHistoryId; - let pageToken: string | undefined; + let pageToken: string | undefined = opts?.startPageToken; // 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. let truncated = false; + let danglingPageToken: string | undefined; for (let page = 0; page < MAX_HISTORY_PAGES; page++) { const qs = new URLSearchParams({ startHistoryId, @@ -190,9 +195,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; + danglingPageToken = data.nextPageToken; + } } - return { ok: true, messageIds: ids, historyId: latestHistoryId, truncated }; + return { + ok: true, + messageIds: ids, + historyId: latestHistoryId, + truncated, + ...(truncated && danglingPageToken ? { nextPageToken: danglingPageToken } : {}), + }; } export async function getRawMessage(token: string, id: string): Promise { diff --git a/workers/providers/workspace.ts b/workers/providers/workspace.ts index 7fff9216..7b918f8c 100644 --- a/workers/providers/workspace.ts +++ b/workers/providers/workspace.ts @@ -152,6 +152,7 @@ interface SidecarStub { history_cursor: 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; + history_page_token: string | null; } | null>; acquirePollLease(nowMs: number, ttlMs: number): Promise; putSidecarState(patch: Record): Promise; @@ -195,6 +196,7 @@ export async function pollWorkspaceMailbox( history_cursor: 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, + history_page_token: null, }; // Backoff gate (rule 2). Sits BEFORE the lease so a backed-off mailbox @@ -253,7 +255,9 @@ 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, { + startPageToken: state.history_page_token ?? undefined, + }); 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). @@ -378,7 +382,18 @@ 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 the dangling page token once the current slice + // is fully worked so the next poll can resume past the page cap (#595). + // A batch cap mid-slice clears the token so the next poll restarts from + // page 1 of the same cursor and can finish unprocessed ids. + if (!history.truncated) { + patch.history_page_token = null; + if (!hitCap) patch.history_cursor = history.historyId; + } else if (!hitCap && history.nextPageToken) { + patch.history_page_token = history.nextPageToken; + } else if (hitCap) { + patch.history_page_token = null; + } 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