diff --git a/specs/mail/gmail-push.md b/specs/mail/gmail-push.md index 160504c..b0b1040 100644 --- a/specs/mail/gmail-push.md +++ b/specs/mail/gmail-push.md @@ -160,16 +160,59 @@ here the cursor itself is unrecoverable.) never doubled (inbound-ingestion.md §4). (Cadence is a tuning knob: daily bounds worst-case staleness to ~24h for a dropped tail notification; a tighter interval trades quota for freshness and can be revisited without changing the design.) -- **Serialize reconciliation per mailbox — deferred to HT-48.** Push-triggered - reconciliation (§2–§3) and this sweep both advance the same mailbox's cursor, so a - mailbox's reconciliation runs *should* be serialized by a **reconciliation lease** (the - inbound analogue of the outbound delivery lease, sending.md §3a); different mailboxes - still reconcile concurrently. This is an efficiency guard, **not** a correctness one — - §4 already makes each run's cursor advance independently safe, so a push landing - mid-sweep is deduped, never doubled — it only avoids redundant - `history.list`/`messages.get` work. Because it is pure optimization and carries a - migration, it is **split out of HT-42 into HT-48**: HT-42 ships the renewal cron and the - sweep (which are correct without the lease); HT-48 adds the lease. +- **Reconciliation is serialized per mailbox by a reconciliation lease (HT-48, + implemented).** Push-triggered reconciliation (§2–§3) and the daily sweep both advance + the same mailbox's cursor, so a mailbox's reconciliation runs are serialized by a + **reconciliation lease** — the inbound analogue of the outbound delivery lease + (sending.md §3a) — held on `gmail_watch_state.claimed_until` (migration 016, + `src/store/gmail-watch-state.ts`'s `claimReconcileLease`/`releaseReconcileLease`); + different mailboxes still reconcile concurrently, since the lease is keyed by + `mailboxId`. This is an efficiency guard, **not** a correctness one — §4 already makes + each run's cursor advance independently safe, so a push landing mid-sweep is deduped, + never doubled — it only avoids redundant `history.list`/`messages.get` work. + + The lease lives entirely in the reconcile job's **consumer** (`src/mail/gmail- + reconcile.ts`), not in either producer (the push webhook or this sweep): a run claims + the lease once it has a confirmed stored cursor and before calling `history.list`; a run + that cannot claim it (another holder's lease is still live) does **not** ack — it returns + `retry` with a short `backoffSeconds` hint + (`DEFAULT_RECONCILE_LEASE_RETRY_BACKOFF_SECONDS`, `src/mail/gmail-reconcile.ts`) and does + no Gmail work of its own that attempt. Acking on a failed claim (an earlier version of + this handler's behavior) is unsafe: the holder's `history.list` snapshot is fixed the + moment it runs, so a message that arrives in Gmail's history *after* that snapshot is + invisible to the holder's own cursor advance — acking the notification for it would drop + it on the floor until the next trigger (a further push, or the next daily sweep), up to + ~24h of added latency on an otherwise-quiet mailbox. Retrying instead means the same job + is redelivered shortly after the holder has very likely released, at which point its own + `history.list` (from the cursor the holder just advanced to) picks up anything the holder + missed — trivially and cheaply in the common case where nothing new arrived. The backoff + is sized so that, combined with the queue's own exponential backoff and `maxAttempts` + dead-letter ceiling (`src/providers/adapters/postgres-queue/index.ts`), a claim that keeps + losing the race still gets an attempt after the holder is *guaranteed* to have released + (its lease cannot outlive `reconcileLeaseMs`) before the job is given up on — see that + constant's own doc comment for the arithmetic. Even in the pathological case where the job + is eventually dead-lettered, no message is lost: cursor-advance and ingest dedup mean the + next trigger reconciles the mailbox from wherever the holder left the cursor, exactly as + it would have before this lease existed. The lease is released in a `finally` around the + `history.list`/fetch/ingest/cursor-advance block, so it is released on every exit — the + happy-path ack, the expired-cursor pause, the blocked-retry, and an unexpected thrown error + alike — *before* that error propagates to the handler's own top-level catch. This was a + deliberate choice: because the lease is a pure efficiency guard, the one failure mode it + must never produce is a mailbox permanently (or even needlessly long) locked out of + reconciliation after a crash; releasing on every path, including a throw, means the next + trigger can reconcile the mailbox immediately rather than waiting out the lease's duration. + The lease's own expiry remains as a backstop for the one case a `finally` cannot reach — + the process being killed outright before it runs. + + The release itself is scoped to the exact lease this run was granted: `claimReconcileLease` + returns an opaque token (the `claimed_until` value it just wrote) that must be passed back + to `releaseReconcileLease`, which clears the lease only if that token still matches the + row's current `claimed_until` — otherwise it is a silent no-op (`src/store/gmail-watch- + state.ts`). This guards against a stale holder (one that overran `reconcileLeaseMs`, e.g. a + large post-downtime backlog) releasing a legitimate successor's live lease out from under + it, which would otherwise let a third trigger claim and duplicate the successor's in-flight + `history.list`/`messages.get` work — precisely the case an unconditional release fails in, + and precisely the load under which that redundant work is most expensive. - **Failure handling — the token layer owns `needs_reconnect`.** A dead grant (revoked/expired, admin change) surfaces as an `invalid_grant` when the OAuth token service refreshes, and *that* is what marks the mailbox **needs-reconnect** (HT-38, diff --git a/src/db/migrate.test.ts b/src/db/migrate.test.ts index d0b873a..6f75be2 100644 --- a/src/db/migrate.test.ts +++ b/src/db/migrate.test.ts @@ -55,6 +55,7 @@ describe('migrate', () => { { id: 13, name: 'queue_jobs' }, { id: 14, name: 'inbound_delivery_lease' }, { id: 15, name: 'thread_attachments' }, + { id: 16, name: 'gmail_reconcile_lease' }, ]) }) @@ -80,6 +81,7 @@ describe('migrate', () => { { id: 13 }, { id: 14 }, { id: 15 }, + { id: 16 }, ]) }) diff --git a/src/db/migrate.ts b/src/db/migrate.ts index c8cf3ff..70778d8 100644 --- a/src/db/migrate.ts +++ b/src/db/migrate.ts @@ -754,6 +754,39 @@ CREATE TABLE thread_attachments ( CREATE INDEX thread_attachments_thread_id_idx ON thread_attachments (thread_id); ` +/** + * Migration 016 — the per-mailbox Gmail reconciliation lease (HT-48; + * specs/mail/gmail-push.md §6, "reconciliation lease → HT-48"). Adds + * `claimed_until` to `gmail_watch_state` (migration 011) — the inbound + * analogue of migration 003's `threads.claimed_until` outbound delivery + * lease, same column name and same `UPDATE ... WHERE claimed_until IS NULL + * OR claimed_until < now()` claim shape (`GmailWatchStateStore + * .claimReconcileLease`/`.releaseReconcileLease`, `src/store/gmail-watch- + * state.ts`). + * + * Unlike the outbound lease, there is no accompanying "status" to record on + * release — this lease guards nothing but redundant Gmail API work + * (`history.list`/`messages.get`) between a push-triggered reconcile + * (HT-41) and the daily sweep (HT-42) landing on the SAME mailbox at + * overlapping times. It is a pure efficiency guard, not a correctness one: + * `src/mail/gmail-reconcile.ts`'s own cursor-advance rule (step 6) and the + * ingest pipeline's dedup on `(mailboxId, providerMessageId)` + * (inbound-ingestion.md §4) already make either ordering safe with no lease + * at all. A run that cannot claim it retries shortly (a short + * `backoffSeconds` hint, not an ack) rather than skipping outright — see + * `src/mail/gmail-reconcile.ts`'s module doc ("Why a failed claim retries + * instead of acking") for why an unconditional skip can silently drop a + * message that arrives after the holder's own `history.list` snapshot. + * + * No `NOT NULL`/CHECK: `NULL` is "unclaimed," matching `threads.claimed_ + * until`'s own nullability. No index: this column is only ever read via an + * equality match on the `mailbox_id` PRIMARY KEY (migration 011), which + * already has its own index. + */ +const MIGRATION_016_GMAIL_RECONCILE_LEASE = ` +ALTER TABLE gmail_watch_state ADD COLUMN claimed_until timestamptz; +` + /** * Every migration, in the order they must apply. `id` is the sole ordering * key (ascending) — array position is not relied upon, so re-sorting this @@ -831,6 +864,11 @@ const MIGRATIONS: Migration[] = [ name: 'thread_attachments', sql: MIGRATION_015_THREAD_ATTACHMENTS, }, + { + id: 16, + name: 'gmail_reconcile_lease', + sql: MIGRATION_016_GMAIL_RECONCILE_LEASE, + }, ] /** diff --git a/src/mail/gmail-connect.test.ts b/src/mail/gmail-connect.test.ts index 209a121..dc16764 100644 --- a/src/mail/gmail-connect.test.ts +++ b/src/mail/gmail-connect.test.ts @@ -490,6 +490,12 @@ describe('createGmailConnectService', () => { setWatchExpiration: async () => { throw new Error('setWatchExpiration: not used by the connect flow') }, + claimReconcileLease: async () => { + throw new Error('claimReconcileLease: not used by the connect flow') + }, + releaseReconcileLease: async () => { + throw new Error('releaseReconcileLease: not used by the connect flow') + }, } const { fetchImpl } = fakeTokenEndpoint(200, DEFAULT_TOKEN_RESPONSE) const service = createGmailConnectService({ diff --git a/src/mail/gmail-reconcile.test.ts b/src/mail/gmail-reconcile.test.ts index f79d892..88da457 100644 --- a/src/mail/gmail-reconcile.test.ts +++ b/src/mail/gmail-reconcile.test.ts @@ -15,7 +15,11 @@ import type { QueueMessage } from '../providers/queue.js' import type { GmailWatchStateStore } from '../store/gmail-watch-state.js' import type { MailboxRecord, MailboxStore } from '../store/mailboxes.js' import type { GmailOAuthTokenService } from './gmail-oauth.js' -import { createGmailReconcileHandler, type GmailReconcileHandlerDeps } from './gmail-reconcile.js' +import { + createGmailReconcileHandler, + DEFAULT_RECONCILE_LEASE_RETRY_BACKOFF_SECONDS, + type GmailReconcileHandlerDeps, +} from './gmail-reconcile.js' import type { IngestOutcome } from './ingest.js' const MAILBOX_ID = '11111111-1111-4111-8111-111111111111' @@ -68,13 +72,31 @@ function fakeMailboxStore(initial: MailboxRecord): { } } +/** + * A `GmailWatchStateStore` fake backed by in-memory maps. `leases` tracks + * each mailbox's held lease as `{ until, token }` (absent = unclaimed) and + * is exposed directly so a test can rewind `until` into the past — + * mirroring `conversations.test.ts`'s `expireLease` helper for the outbound + * lease — to exercise lease expiry without a real sleep. `token` mirrors + * the real store's opaque lease-token contract (`src/store/gmail-watch- + * state.ts`): `claimReconcileLease` returns it on success, + * `releaseReconcileLease` clears the lease ONLY if the token passed back + * still matches, exactly like the real `claimed_until`-scoped `UPDATE`. + * `rows` mirrors the real store's "no `gmail_watch_state` row" case: + * `claimReconcileLease` requires a row to exist, exactly like the real + * `UPDATE`-only (non-upserting) SQL. + */ function fakeWatchStateStore(initial: Record = {}): { store: GmailWatchStateStore cursors: Map setCalls: Array<{ mailboxId: string; historyId: string }> + leases: Map } { const cursors = new Map(Object.entries(initial)) + const rows = new Set(Object.keys(initial)) const setCalls: Array<{ mailboxId: string; historyId: string }> = [] + const leases = new Map() + let leaseTokenCounter = 0 return { store: { async getCursor(mailboxId) { @@ -82,6 +104,7 @@ function fakeWatchStateStore(initial: Record = {}): { }, async setCursor(mailboxId, historyId) { cursors.set(mailboxId, historyId) + rows.add(mailboxId) setCalls.push({ mailboxId, historyId }) }, async seedBaseline() { @@ -90,9 +113,27 @@ function fakeWatchStateStore(initial: Record = {}): { async setWatchExpiration() { throw new Error('setWatchExpiration: not used by the reconcile handler') }, + async claimReconcileLease(mailboxId, leaseMs) { + if (!rows.has(mailboxId)) return null + const current = leases.get(mailboxId) + if (current !== undefined && current.until > Date.now()) return null + const token = `lease-token-${++leaseTokenCounter}` + leases.set(mailboxId, { until: Date.now() + leaseMs, token }) + return token + }, + async releaseReconcileLease(mailboxId, leaseToken) { + // Mirrors the real store's `WHERE claimed_until = $2` scoping: a + // release whose token no longer matches the CURRENT holder (already + // superseded, or the row is gone) is a silent no-op, never a throw. + const current = leases.get(mailboxId) + if (current !== undefined && current.token === leaseToken) { + leases.delete(mailboxId) + } + }, }, cursors, setCalls, + leases, } } @@ -642,4 +683,297 @@ describe('createGmailReconcileHandler', () => { expect(setCalls).toEqual([]) consoleErrorSpy.mockRestore() }) + + // --- The reconciliation lease (HT-48; gmail-push.md §6) --------------------- + + describe('reconciliation lease', () => { + it('concurrent reconcile of the same mailbox does the Gmail work once — the loser retries instead of acking', async () => { + const { store: watchStateStore, setCalls } = fakeWatchStateStore({ [MAILBOX_ID]: 'cursor-1' }) + const listAddedMessageIds = vi.fn(async () => ({ + kind: 'ok' as const, + messageIds: [], + newHistoryId: 'cursor-2', + })) + const historyClient: GmailHistoryClient = { + listAddedMessageIds, + async getRawMessage() { + return null + }, + } + const ingest = vi.fn() + + const handler = createGmailReconcileHandler( + baseDeps({ watchStateStore, ingest, createHistoryClient: () => historyClient }), + ) + + // Two triggers landing on the SAME mailbox at once — e.g. a push + // notification and the daily sweep both enqueuing/consuming a + // reconcile job for it around the same moment. + const [first, second] = await Promise.all([handler(job()), handler(job())]) + + // Which of the two wins the claim race is nondeterministic — assert + // on the pair, not on `first`/`second` individually. Exactly one acks + // (the holder); the other retries with the lease-held backoff hint + // rather than acking (module doc's "Why a failed claim retries + // instead of acking") — it did not do the Gmail work, but it also did + // not silently discard whatever notification it was carrying. + const results = [first, second] + expect(results).toContainEqual({ kind: 'ack' }) + expect(results).toContainEqual({ + kind: 'retry', + backoffSeconds: DEFAULT_RECONCILE_LEASE_RETRY_BACKOFF_SECONDS, + }) + // Only the lease-holder actually called history.list; the other run + // skipped its own Gmail work entirely (module doc's "The + // reconciliation lease" section). + expect(listAddedMessageIds).toHaveBeenCalledTimes(1) + expect(setCalls).toEqual([{ mailboxId: MAILBOX_ID, historyId: 'cursor-2' }]) + }) + + it('different mailboxes reconcile concurrently — one mailbox holding the lease never blocks another', async () => { + const MAILBOX_A = MAILBOX_ID + const MAILBOX_B = '22222222-2222-4222-8222-222222222222' + const recordsByAddress = new Map([ + ['support@example.test', activeMailbox({ id: MAILBOX_A })], + [ + 'support-b@example.test', + activeMailbox({ id: MAILBOX_B, address: 'support-b@example.test' }), + ], + ]) + const recordsById = new Map( + [...recordsByAddress.values()].map((r) => [r.id, r]), + ) + const mailboxStore: MailboxStore = { + async getMailboxByAddress(address) { + return recordsByAddress.get(address) ?? null + }, + async getMailboxById(id) { + return recordsById.get(id) ?? null + }, + async markNeedsReconnect() { + throw new Error('markNeedsReconnect: not used by this test') + }, + async markPaused() { + throw new Error('markPaused: not used by this test') + }, + async upsertConnectedMailbox() { + throw new Error('upsertConnectedMailbox: not used by this test') + }, + async listActiveMailboxes() { + throw new Error('listActiveMailboxes: not used by this test') + }, + } + const { store: watchStateStore, setCalls } = fakeWatchStateStore({ + [MAILBOX_A]: 'cursor-a-1', + [MAILBOX_B]: 'cursor-b-1', + }) + const listAddedMessageIds = vi.fn(async (cursor: string) => ({ + kind: 'ok' as const, + messageIds: [], + newHistoryId: cursor === 'cursor-a-1' ? 'cursor-a-2' : 'cursor-b-2', + })) + const historyClient: GmailHistoryClient = { + listAddedMessageIds, + async getRawMessage() { + return null + }, + } + + const handler = createGmailReconcileHandler( + baseDeps({ + mailboxStore, + watchStateStore, + ingest: vi.fn(), + createHistoryClient: () => historyClient, + }), + ) + + const [a, b] = await Promise.all([ + handler(job({ mailboxId: MAILBOX_A })), + handler(job({ mailboxId: MAILBOX_B })), + ]) + + expect(a).toEqual({ kind: 'ack' }) + expect(b).toEqual({ kind: 'ack' }) + // Both mailboxes did their OWN history.list — a lease is strictly + // per-mailbox, so mailbox A holding its lease never blocks mailbox B. + expect(listAddedMessageIds).toHaveBeenCalledTimes(2) + expect(setCalls).toEqual( + expect.arrayContaining([ + { mailboxId: MAILBOX_A, historyId: 'cursor-a-2' }, + { mailboxId: MAILBOX_B, historyId: 'cursor-b-2' }, + ]), + ) + }) + + it('a crashed holder lease expires and reconciliation resumes', async () => { + const { + store: watchStateStore, + setCalls, + leases, + } = fakeWatchStateStore({ + [MAILBOX_ID]: 'cursor-1', + }) + // Simulate a PRIOR run that claimed the lease and then crashed before + // ever reaching its own release (the one case the `finally` in + // gmail-reconcile.ts's module doc cannot help) — its claimed_until is + // already in the past, exactly as it would be once reconcileLeaseMs + // has elapsed with no release call ever having run. + leases.set(MAILBOX_ID, { until: Date.now() - 1000, token: 'stale-crashed-holder-token' }) + const listAddedMessageIds = vi.fn(async () => ({ + kind: 'ok' as const, + messageIds: [], + newHistoryId: 'cursor-2', + })) + const historyClient: GmailHistoryClient = { + listAddedMessageIds, + async getRawMessage() { + return null + }, + } + + const handler = createGmailReconcileHandler( + baseDeps({ watchStateStore, ingest: vi.fn(), createHistoryClient: () => historyClient }), + ) + + const result = await handler(job()) + + expect(result).toEqual({ kind: 'ack' }) + expect(listAddedMessageIds).toHaveBeenCalledTimes(1) + expect(setCalls).toEqual([{ mailboxId: MAILBOX_ID, historyId: 'cursor-2' }]) + }) + + it('a run that cannot claim the lease retries with a backoff hint, without touching the Gmail client at all', async () => { + const { + store: watchStateStore, + setCalls, + leases, + } = fakeWatchStateStore({ + [MAILBOX_ID]: 'cursor-1', + }) + // Held by someone else, unexpired. + leases.set(MAILBOX_ID, { until: Date.now() + 60_000, token: 'live-holder-token' }) + const createHistoryClient = vi.fn() + const ingest = vi.fn() + + const handler = createGmailReconcileHandler( + baseDeps({ watchStateStore, ingest, createHistoryClient }), + ) + + const result = await handler(job()) + + // NOT an ack: acking here would silently drop a message that arrives + // in Gmail's history after the holder's own history.list snapshot — + // see gmail-reconcile.ts's module doc ("Why a failed claim retries + // instead of acking"). Retrying with a backoff hint gives a later + // attempt (after the holder has very likely released) a chance to + // pick that message up. + expect(result).toEqual({ + kind: 'retry', + backoffSeconds: DEFAULT_RECONCILE_LEASE_RETRY_BACKOFF_SECONDS, + }) + expect(createHistoryClient).not.toHaveBeenCalled() + expect(ingest).not.toHaveBeenCalled() + expect(setCalls).toEqual([]) + }) + + it("the arrives-after-snapshot case: a message that lands after the holder's history.list is picked up on the retried attempt, not dropped", async () => { + const { + store: watchStateStore, + setCalls, + cursors, + leases, + } = fakeWatchStateStore({ + [MAILBOX_ID]: 'cursor-1', + }) + // Simulate holder A already in flight: it claimed the lease and — in + // the concrete scenario this guards against — has already called + // history.list (fixing its own snapshot) and is mid-fetch/ingest. + leases.set(MAILBOX_ID, { until: Date.now() + 60_000, token: 'holder-a-token' }) + const createHistoryClient = vi.fn() + const ingest = vi.fn() + const handler = createGmailReconcileHandler( + baseDeps({ watchStateStore, ingest, createHistoryClient }), + ) + + // Worker B's job — enqueued by a push notification for message M, + // which arrived AFTER holder A's history.list snapshot — cannot claim + // the lease and gets a retry, not an ack. + const firstAttempt = await handler(job()) + expect(firstAttempt).toEqual({ + kind: 'retry', + backoffSeconds: DEFAULT_RECONCILE_LEASE_RETRY_BACKOFF_SECONDS, + }) + expect(createHistoryClient).not.toHaveBeenCalled() + + // Holder A finishes its own run and releases with ITS token. The + // cursor is now at A's snapshot watermark, which does NOT yet cover + // message M (that is exactly the scenario: M arrived after A's list + // call, so A's own advance cannot have reached it). + await watchStateStore.releaseReconcileLease(MAILBOX_ID, 'holder-a-token') + cursors.set(MAILBOX_ID, 'cursor-after-a') + + // The retried delivery of B's job now claims the free lease and runs + // its OWN history.list from the advanced cursor, picking up M. + const historyClient: GmailHistoryClient = { + listAddedMessageIds: vi.fn(async (cursor: string) => ({ + kind: 'ok' as const, + messageIds: cursor === 'cursor-after-a' ? ['m-arrived-after-a'] : [], + newHistoryId: 'cursor-after-b', + })), + getRawMessage: vi.fn(async () => ({ + rawBytes: textBytes('raw-M'), + receivedAt: new Date('2026-01-03T00:00:00Z'), + })), + } + const ingestedAfterRetry = vi.fn(async (raw: RawInboundMessage) => storedOutcome(raw)) + const retriedHandler = createGmailReconcileHandler( + baseDeps({ + watchStateStore, + ingest: ingestedAfterRetry, + createHistoryClient: () => historyClient, + }), + ) + + const secondAttempt = await retriedHandler(job()) + + expect(secondAttempt).toEqual({ kind: 'ack' }) + expect(ingestedAfterRetry).toHaveBeenCalledTimes(1) + expect((ingestedAfterRetry.mock.calls[0][0] as RawInboundMessage).providerMessageId).toBe( + 'm-arrived-after-a', + ) + expect(setCalls).toEqual([{ mailboxId: MAILBOX_ID, historyId: 'cursor-after-b' }]) + }) + + it('an unexpected throw releases the lease immediately — a next run need not wait out reconcileLeaseMs', async () => { + const { store: watchStateStore, leases } = fakeWatchStateStore({ [MAILBOX_ID]: 'cursor-1' }) + const historyClient: GmailHistoryClient = { + async listAddedMessageIds() { + throw new Error('boom: transient network failure') + }, + async getRawMessage() { + return null + }, + } + const consoleErrorSpy = vi.spyOn(console, 'error').mockImplementation(() => undefined) + + const handler = createGmailReconcileHandler( + baseDeps({ + watchStateStore, + createHistoryClient: () => historyClient, + // A long lease — if release-on-throw did NOT happen, the lease + // would still read as held for a very long time. + reconcileLeaseMs: 10 * 60_000, + }), + ) + + const result = await handler(job()) + expect(result).toEqual({ kind: 'retry' }) + + // The lease must already be free — released in the `finally` before + // the throw propagated — not still held for another ~10 minutes. + expect(leases.has(MAILBOX_ID)).toBe(false) + consoleErrorSpy.mockRestore() + }) + }) }) diff --git a/src/mail/gmail-reconcile.ts b/src/mail/gmail-reconcile.ts index ee5919f..2dede5c 100644 --- a/src/mail/gmail-reconcile.ts +++ b/src/mail/gmail-reconcile.ts @@ -56,6 +56,86 @@ * and reported as `{ kind: 'retry' }` — never as `ack`, and never after * advancing the cursor. * + * ## The reconciliation lease (HT-48; gmail-push.md §6) + * + * Between step 3 (a confirmed, non-null stored cursor) and step 4 + * (`history.list`), this run claims `mailboxId`'s reconciliation lease + * (`GmailWatchStateStore.claimReconcileLease`, `claimed_until` on + * `gmail_watch_state`, migration 016) — the inbound analogue of the + * outbound delivery lease (`ConversationStore.claimThreadForDelivery`, + * sending.md §3a). It exists ONLY to stop a push-triggered reconcile + * (HT-41) and the daily sweep (HT-42) from doing the SAME `history.list`/ + * `messages.get` work concurrently when both land on one mailbox at once — + * gmail-push.md §6 is explicit this is an efficiency guard, not a + * correctness one: step 6's cursor-advance rule and the ingest pipeline's + * dedup on `(mailboxId, providerMessageId)` (inbound-ingestion.md §4) + * already make either ordering safe with no lease at all. Different + * mailboxes never contend — the lease is keyed by `mailboxId`. + * + * A run that cannot claim the lease (another holder's `claimed_until` is + * still in the future) does NOT ack — it returns `{ kind: 'retry', + * backoffSeconds: DEFAULT_RECONCILE_LEASE_RETRY_BACKOFF_SECONDS }` and does + * no Gmail work of its own this attempt. + * + * ## Why a failed claim retries instead of acking (correction, flagged in review) + * + * An earlier version of this handler acked on a failed claim, reasoning + * "the holder will advance the cursor — there is nothing this run needs to + * do that the holder won't already do." That reasoning is false for + * anything that arrives AFTER the holder's `history.list` snapshot: the + * holder's `listAddedMessageIds` call fixes its batch and its eventual + * `newHistoryId` the moment it runs; a message that lands in Gmail's + * history a moment later is invisible to that in-flight run and will not + * be swept up by its cursor advance. Concretely — a sweep-triggered run + * claims the lease and lists history up to `H1`, then spends the + * fetch/ingest phase on that batch; a NEW customer message arrives at + * `H2 > H1` and Gmail pushes a notification for it; that push's reconcile + * job is consumed by a second run WHILE the first still holds the lease, + * so the second run's claim fails. Acking there — as this handler used + * to — discards that notification outright: the holder's `setCursor` only + * advances to `H1`, so the message at `H2` is not reconciled until the + * NEXT trigger (a further push, or the daily sweep, gmail-push.md §6) — + * up to ~24h of silent added latency on an otherwise-quiet mailbox. This + * is a correctness-adjacent latency regression, not covered by the + * `(mailboxId, providerMessageId)` ingest dedup (inbound-ingestion.md §4), + * which guards against DOUBLING work, not against a run's snapshot simply + * predating the message. Returning `retry` with a short backoff instead + * means the SAME job is redelivered after the holder has very likely + * released (see {@link DEFAULT_RECONCILE_LEASE_RETRY_BACKOFF_SECONDS} for + * how the backoff is sized against `reconcileLeaseMs` and the queue's own + * `maxAttempts` dead-letter ceiling); that retried attempt claims the + * now-free lease and runs its OWN `history.list` from the cursor the + * holder just advanced to, which trivially and cheaply picks up `H2`. The + * lease therefore remains a pure efficiency guard in the COMMON case (no + * new mail mid-run: the retry's `history.list` comes back empty, `newHistoryId` + * unchanged) while no longer silently dropping the promptness of the RARE + * arrives-mid-run case. + * + * The lease is released in a `finally` wrapped around steps 4-6, so it is + * released on every path out of that block: the happy-path ack, the + * expired-cursor pause, the blocked-retry (non-terminal ingest outcome), + * AND an unexpected thrown error (network, Gmail client, ingest, store) — + * release happens BEFORE the throw propagates to this handler's own + * top-level catch. This is a deliberate choice: because the lease is purely + * an efficiency guard (never a correctness one), the failure mode it must + * never produce is "a mailbox that just threw is locked out of + * reconciliation until the lease naturally expires" — releasing + * immediately on every exit path, including a throw, means the NEXT + * trigger (a fresh push, or tomorrow's sweep) can reconcile this mailbox + * right away rather than waiting out `reconcileLeaseMs`. The release call + * itself is wrapped in its own try/catch that only logs — a release + * failure (a genuine DB error) must not override this run's own outcome + * (`ack`/`retry`) with something else, and IS still covered by the lease's + * own expiry as a backstop for the one case a `finally` block cannot help: + * the process being killed outright before the `finally` ever runs. + * + * The release itself is now scoped to the exact lease this run was granted + * (`GmailWatchStateStore.claimReconcileLease`'s returned token, passed back + * to `releaseReconcileLease`) rather than an unconditional clear — see that + * store module's doc comment for the stale-holder scenario (an overrunning + * run's release clobbering a legitimate successor's live lease) this + * closes. + * * ## Never drop a message (charter §2; gmail-push.md §4) * * The cursor is the only thing that can make a message permanently @@ -112,6 +192,56 @@ import type { IngestOutcome } from './ingest.js' */ export const DEFAULT_MAX_INLINE_RAW_BYTES = 1_000_000 +/** + * How long one mailbox's reconciliation lease (module doc's "The + * reconciliation lease" section, HT-48) is held for. Not pinned to any spec + * number — chosen as a judgment call (flagged in this ticket's report), + * matching `./delivery-worker.ts`'s `DEFAULT_STALE_AFTER_MS` (5 minutes): + * long enough to cover a realistic `history.list` page plus a batch of + * `messages.get`/`ingest` calls, short enough that a crashed holder (the + * one case the `finally`-release in the module doc cannot reach) does not + * lock a mailbox out of reconciliation for long. Because this is a pure + * efficiency guard (never a correctness one — module doc), the exact value + * only trades a little redundant Gmail API work against lock-out latency, + * never data safety. + */ +export const DEFAULT_RECONCILE_LEASE_MS = 5 * 60_000 + +/** + * `backoffSeconds` hint returned with `{ kind: 'retry' }` when a run cannot + * claim the reconciliation lease (module doc's "Why a failed claim retries + * instead of acking"). Not pinned to any spec number — chosen as a judgment + * call (flagged in this ticket's report), sized against TWO other numbers + * this file does not otherwise control: + * + * - `DEFAULT_RECONCILE_LEASE_MS` (5 minutes) — the longest a legitimate + * holder can keep the lease before releasing it. + * - The queue adapter's retry-until-dead-letter ceiling + * (`createPostgresQueue`'s `maxAttempts`, default 5, and its exponential + * backoff growth — `src/providers/adapters/postgres-queue/index.ts`): + * returning `backoffSeconds: b` makes `b` the exponential BASE for this + * job's own subsequent retries (`b, 2b, 4b, 8b` before the 5th and final + * attempt), so the total window this job keeps retrying before the queue + * gives up and dead-letters it is `15b` seconds. + * + * `25` seconds makes that total window `375s` (~6.25 minutes) — comfortably + * longer than `DEFAULT_RECONCILE_LEASE_MS`, so a claim that keeps losing the + * race against a legitimately slow holder still gets one attempt after that + * holder is GUARANTEED to have released (its lease cannot outlive + * `reconcileLeaseMs`). Even in the pathological case where every retry + * still loses the race and the job is eventually dead-lettered, no message + * is dropped: cursor-advance (step 6) and ingest dedup (inbound- + * ingestion.md §4) mean the next trigger — a further push, or the daily + * sweep (gmail-push.md §6) — reconciles this mailbox from wherever the + * holder left the cursor, exactly as before this lease existed at all. This + * constant only trades a little redundant queue churn against how quickly a + * message that arrived mid-holder-run gets reconciled, never data safety. + * If `reconcileLeaseMs` is overridden well above its default at the + * composition root, this constant (or the queue's own `maxAttempts`/backoff + * options) should be reconsidered alongside it. + */ +export const DEFAULT_RECONCILE_LEASE_RETRY_BACKOFF_SECONDS = 25 + /** Dependencies {@link createGmailReconcileHandler} needs. */ export interface GmailReconcileHandlerDeps { /** Resolves a live Gmail API access token for one mailbox at a time (`./gmail-oauth.ts`). */ @@ -160,6 +290,12 @@ export interface GmailReconcileHandlerDeps { /** See {@link DEFAULT_MAX_INLINE_RAW_BYTES}. */ maxInlineRawBytes?: number + + /** See {@link DEFAULT_RECONCILE_LEASE_MS}. */ + reconcileLeaseMs?: number + + /** See {@link DEFAULT_RECONCILE_LEASE_RETRY_BACKOFF_SECONDS}. */ + reconcileLeaseRetryBackoffSeconds?: number } /** Build the `QueueMessageHandler`. See the module doc for the full control flow. */ @@ -174,6 +310,8 @@ export function createGmailReconcileHandler( ingest, createHistoryClient, maxInlineRawBytes = DEFAULT_MAX_INLINE_RAW_BYTES, + reconcileLeaseMs = DEFAULT_RECONCILE_LEASE_MS, + reconcileLeaseRetryBackoffSeconds = DEFAULT_RECONCILE_LEASE_RETRY_BACKOFF_SECONDS, } = deps return async (message: QueueMessage): Promise => { @@ -187,6 +325,8 @@ export function createGmailReconcileHandler( ingest, createHistoryClient, maxInlineRawBytes, + reconcileLeaseMs, + reconcileLeaseRetryBackoffSeconds, }) } catch (err) { // Any unexpected throw (network, timeout, a non-404 non-2xx from the @@ -216,6 +356,8 @@ interface ReconcileDeps { ingest: (raw: RawInboundMessage) => Promise createHistoryClient: (getAccessToken: () => Promise) => GmailHistoryClient maxInlineRawBytes: number + reconcileLeaseMs: number + reconcileLeaseRetryBackoffSeconds: number } /** Run steps 1-6 (module doc) for one mailbox's reconcile job. */ @@ -232,6 +374,8 @@ async function reconcileOneMailbox( ingest, createHistoryClient, maxInlineRawBytes, + reconcileLeaseMs, + reconcileLeaseRetryBackoffSeconds, } = deps // --- Step 1: re-read CURRENT status — never trust the job's snapshot. --- @@ -285,71 +429,116 @@ async function reconcileOneMailbox( return { kind: 'ack' } } - // --- Step 4: history.list from the stored cursor. --- - const client = createHistoryClient(getAccessToken) - const listed = await client.listAddedMessageIds(cursor) - if (listed.kind === 'expired') { - await mailboxStore.markPaused(mailboxId) - logReconcileEvent('warn', { + // --- Step 3a: claim the reconciliation lease (HT-48; module doc's "The + // reconciliation lease" section). A run that cannot claim it retries + // shortly rather than acking — module doc's "Why a failed claim retries + // instead of acking" explains why acking here can silently drop a + // message that arrived after the holder's own history.list snapshot. --- + const leaseToken = await watchStateStore.claimReconcileLease(mailboxId, reconcileLeaseMs) + if (leaseToken === null) { + logReconcileEvent('info', { mailboxId, - outcome: 'ack', - reason: 'cursor-expired', - cursor, - note: 'cursor expired (404); mailbox paused for manual rebaseline per gmail-push.md §5', + outcome: 'retry', + reason: 'reconcile-lease-held', + backoffSeconds: reconcileLeaseRetryBackoffSeconds, + note: "another in-flight reconcile (push or sweep) holds this mailbox lease; retrying shortly rather than acking, so anything past the holder's own history.list snapshot is not silently dropped — gmail-push.md §6, HT-48", }) - return { kind: 'ack' } + return { kind: 'retry', backoffSeconds: reconcileLeaseRetryBackoffSeconds } } - // --- Step 5: fetch + ingest each added message, in order. --- - const outcomes: IngestOutcome[] = [] - for (const messageId of listed.messageIds) { - const fetched = await client.getRawMessage(messageId) - if (fetched === null) { - // Deleted between list and get — nothing to ingest, nothing to - // retry; skip (module doc, step 5). - continue + try { + // --- Step 4: history.list from the stored cursor. --- + const client = createHistoryClient(getAccessToken) + const listed = await client.listAddedMessageIds(cursor) + if (listed.kind === 'expired') { + await mailboxStore.markPaused(mailboxId) + logReconcileEvent('warn', { + mailboxId, + outcome: 'ack', + reason: 'cursor-expired', + cursor, + note: 'cursor expired (404); mailbox paused for manual rebaseline per gmail-push.md §5', + }) + return { kind: 'ack' } } - const content = await buildRawMessageContent(fetched.rawBytes, { - mailboxId, - messageId, - maxInlineRawBytes, - blobStore, - }) + // --- Step 5: fetch + ingest each added message, in order. --- + const outcomes: IngestOutcome[] = [] + for (const messageId of listed.messageIds) { + const fetched = await client.getRawMessage(messageId) + if (fetched === null) { + // Deleted between list and get — nothing to ingest, nothing to + // retry; skip (module doc, step 5). + continue + } - const raw: RawInboundMessage = { - content, - mailboxId, - providerMessageId: messageId, - receivedAt: fetched.receivedAt, + const content = await buildRawMessageContent(fetched.rawBytes, { + mailboxId, + messageId, + maxInlineRawBytes, + blobStore, + }) + + const raw: RawInboundMessage = { + content, + mailboxId, + providerMessageId: messageId, + receivedAt: fetched.receivedAt, + } + outcomes.push(await ingest(raw)) } - outcomes.push(await ingest(raw)) - } - // --- Step 6: advance the cursor iff every outcome is terminal & ledgered. --- - const blocking = outcomes.find((o) => o.kind === 'failed' || o.kind === 'in-progress') - if (blocking !== undefined) { - logReconcileEvent('warn', { + // --- Step 6: advance the cursor iff every outcome is terminal & ledgered. --- + const blocking = outcomes.find((o) => o.kind === 'failed' || o.kind === 'in-progress') + if (blocking !== undefined) { + logReconcileEvent('warn', { + mailboxId, + outcome: 'retry', + reason: 'non-terminal-ingest-outcome', + blockingOutcomeKind: blocking.kind, + blockingProviderMessageId: blocking.providerMessageId, + batchSize: listed.messageIds.length, + }) + return { kind: 'retry' } + } + + await watchStateStore.setCursor(mailboxId, listed.newHistoryId) + logReconcileEvent('info', { mailboxId, - outcome: 'retry', - reason: 'non-terminal-ingest-outcome', - blockingOutcomeKind: blocking.kind, - blockingProviderMessageId: blocking.providerMessageId, - batchSize: listed.messageIds.length, + outcome: 'ack', + reason: 'reconciled', + messageCount: listed.messageIds.length, + previousCursor: cursor, + newHistoryId: listed.newHistoryId, }) - return { kind: 'retry' } + return { kind: 'ack' } + } finally { + // Release on every exit from the try above — success, the + // expired-cursor pause, the blocked-retry, AND an unexpected throw + // (which this `finally` runs BEFORE the exception propagates to + // createGmailReconcileHandler's own top-level catch). See the module + // doc's "The reconciliation lease" section for why this must never be + // conditioned on the outcome: the lease is a pure efficiency guard, so + // a mailbox that just threw must not be locked out of reconciliation + // until reconcileLeaseMs elapses. + try { + // Scoped to `leaseToken` — the exact lease THIS run was granted — so + // an overrunning run's release can never clobber a legitimate + // successor's live lease (`GmailWatchStateStore.releaseReconcileLease`'s + // doc comment for the stale-holder scenario this closes). + await watchStateStore.releaseReconcileLease(mailboxId, leaseToken) + } catch (releaseErr) { + // A release failure must not override this run's own outcome (ack/ + // retry, or the throw already in flight) — logged only. The lease's + // own expiry remains the backstop (module doc). + logReconcileEvent('error', { + mailboxId, + outcome: 'lease-release-failed', + reason: 'unexpected-error', + error: releaseErr instanceof Error ? releaseErr.message : String(releaseErr), + }) + } } - - await watchStateStore.setCursor(mailboxId, listed.newHistoryId) - logReconcileEvent('info', { - mailboxId, - outcome: 'ack', - reason: 'reconciled', - messageCount: listed.messageIds.length, - previousCursor: cursor, - newHistoryId: listed.newHistoryId, - }) - return { kind: 'ack' } } /** diff --git a/src/mail/gmail-watch-maintenance.ts b/src/mail/gmail-watch-maintenance.ts index 7fce68c..b37e1ce 100644 --- a/src/mail/gmail-watch-maintenance.ts +++ b/src/mail/gmail-watch-maintenance.ts @@ -80,18 +80,22 @@ * fresh `historyId` is AHEAD of the stored cursor, and overwriting the * cursor with it would silently skip un-reconciled mail). * - * ## No reconciliation lease here — deferred to HT-48 + * ## The reconciliation lease lives in the CONSUMER, not here (HT-48) * * Push-triggered reconciliation and this sweep both advance the same * mailbox's cursor. gmail-push.md §6 calls serializing them a pure * efficiency guard (avoiding redundant `history.list`/`messages.get` * work), NOT a correctness requirement — the ingest pipeline's own dedup - * (inbound-ingestion.md §4) already makes either ordering safe. The lease - * is out of scope for this ticket (HT-48); this sweep enqueues its - * reconcile job with NO `dedupeKey` on purpose (see - * {@link maintainOneMailbox}) — a daily sweep of an already-current, quiet + * (inbound-ingestion.md §4) already makes either ordering safe. HT-48 + * implements that lease entirely in the reconcile job's CONSUMER + * (`./gmail-reconcile.ts`'s `claimReconcileLease`/`releaseReconcileLease` + * around `history.list`), not in this PRODUCER — this sweep still enqueues + * its reconcile job with NO `dedupeKey` on purpose (see + * {@link maintainOneMailbox}): a daily sweep of an already-current, quiet * mailbox must still run, not be silently suppressed as a duplicate of an - * earlier job. + * earlier job. Whether the resulting job actually does any Gmail work, or + * skips because another in-flight reconcile already holds the lease, is + * decided entirely on the consumer side, once the job is dequeued. */ import { GMAIL_RECONCILE_TOPIC, type GmailReconcileJob } from '../api/gmail-webhook.js' @@ -294,8 +298,10 @@ async function maintainOneMailbox( // only renews the expiration). NO dedupeKey (module doc): a daily sweep // of an already-current, quiet mailbox must still run, never be // suppressed as a duplicate of an earlier job — redundant reconcile work - // here is exactly what HT-48's lease will optimize away, and is safe - // today because ingest dedups on (mailboxId, providerMessageId). --- + // here is exactly what the consumer's lease (HT-48, ./gmail-reconcile.ts) + // now optimizes away when this job lands while another reconcile of the + // same mailbox is already in flight, and is otherwise safe because + // ingest dedups on (mailboxId, providerMessageId). --- const cursor = await watchStateStore.getCursor(mailboxId) if (cursor === null) { logMaintenanceEvent('info', { diff --git a/src/store/gmail-watch-state.test.ts b/src/store/gmail-watch-state.test.ts index 6a151d4..da7b2c2 100644 --- a/src/store/gmail-watch-state.test.ts +++ b/src/store/gmail-watch-state.test.ts @@ -346,4 +346,163 @@ describe('createGmailWatchStateStore', () => { expect(await store.getCursor(mailboxB)).toBe('b-1') }) }) + + // --- claimReconcileLease / releaseReconcileLease (HT-48, gmail-push.md §6) --- + + /** Directly rewinds a mailbox's claimed_until into the past — mirrors conversations.test.ts's expireLease for the outbound lease, exercising expiry without a real sleep. */ + async function expireReconcileLease(db: Db, mailboxId: string) { + await db.query( + "UPDATE gmail_watch_state SET claimed_until = now() - interval '1 second' WHERE mailbox_id = $1", + [mailboxId], + ) + } + + describe('claimReconcileLease / releaseReconcileLease', () => { + it('claims an unclaimed mailbox, setting claimed_until in the future and returning it as the lease token', async () => { + const { db, store } = await freshStore() + const mailboxId = await insertMailbox(db) + await store.setCursor(mailboxId, 'cursor-1') + + const before = new Date() + const token = await store.claimReconcileLease(mailboxId, 30_000) + + expect(token).not.toBeNull() + const rows = await db.query<{ claimed_until: Date | null }>( + 'SELECT claimed_until FROM gmail_watch_state WHERE mailbox_id = $1', + [mailboxId], + ) + expect(rows[0].claimed_until).not.toBeNull() + expect((rows[0].claimed_until as Date).getTime()).toBeGreaterThan(before.getTime()) + // The returned token is Postgres's own textual rendering of the exact + // claimed_until it just wrote (store module doc's "Why the token is + // text, not a Date") — round-tripping it back through ::timestamptz + // must land on the identical instant. + const roundTripped = await db.query<{ matches: boolean }>( + 'SELECT claimed_until = $2::timestamptz AS matches FROM gmail_watch_state WHERE mailbox_id = $1', + [mailboxId, token as string], + ) + expect(roundTripped[0].matches).toBe(true) + }) + + it('a second claim attempt while the lease is held returns null', async () => { + const { db, store } = await freshStore() + const mailboxId = await insertMailbox(db) + await store.setCursor(mailboxId, 'cursor-1') + + const first = await store.claimReconcileLease(mailboxId, 30_000) + expect(first).not.toBeNull() + + const second = await store.claimReconcileLease(mailboxId, 30_000) + expect(second).toBeNull() + }) + + it('claiming succeeds again once the previous lease has expired, returning a NEW token', async () => { + const { db, store } = await freshStore() + const mailboxId = await insertMailbox(db) + await store.setCursor(mailboxId, 'cursor-1') + + const first = await store.claimReconcileLease(mailboxId, 30_000) + expect(first).not.toBeNull() + await expireReconcileLease(db, mailboxId) + + const second = await store.claimReconcileLease(mailboxId, 30_000) + expect(second).not.toBeNull() + expect(second).not.toBe(first) + }) + + it('different mailboxes claim independently — one holding its lease does not block another', async () => { + const { db, store } = await freshStore() + const mailboxA = await insertMailbox(db, 'lease-a@example.test') + const mailboxB = await insertMailbox(db, 'lease-b@example.test') + await store.setCursor(mailboxA, 'a-1') + await store.setCursor(mailboxB, 'b-1') + + const claimedA = await store.claimReconcileLease(mailboxA, 30_000) + const claimedB = await store.claimReconcileLease(mailboxB, 30_000) + + expect(claimedA).not.toBeNull() + expect(claimedB).not.toBeNull() + }) + + it('returns null for a mailbox with no gmail_watch_state row at all', async () => { + const { db, store } = await freshStore() + const mailboxId = await insertMailbox(db) + // No setCursor/seedBaseline call — no gmail_watch_state row exists yet. + + expect(await store.claimReconcileLease(mailboxId, 30_000)).toBeNull() + }) + + it('releaseReconcileLease clears claimed_until when the token matches the current holder', async () => { + const { db, store } = await freshStore() + const mailboxId = await insertMailbox(db) + await store.setCursor(mailboxId, 'cursor-1') + const token = await store.claimReconcileLease(mailboxId, 30_000) + + await store.releaseReconcileLease(mailboxId, token as string) + + const rows = await db.query<{ claimed_until: Date | null }>( + 'SELECT claimed_until FROM gmail_watch_state WHERE mailbox_id = $1', + [mailboxId], + ) + expect(rows[0].claimed_until).toBeNull() + }) + + it('releaseReconcileLease is a silent no-op (never throws) for a mailbox with no gmail_watch_state row', async () => { + const { store } = await freshStore() + // A syntactically valid `timestamptz` text (the shape every real + // token takes — see the store module doc's "Why the token is text, + // not a Date") that simply cannot match any row, since none exists. + await expect( + store.releaseReconcileLease( + '00000000-0000-4000-8000-000000000000', + '2020-01-01 00:00:00+00', + ), + ).resolves.toBeUndefined() + }) + + it("releaseReconcileLease is a silent no-op when the token no longer matches — a stale holder cannot clobber a live successor's lease", async () => { + const { db, store } = await freshStore() + const mailboxId = await insertMailbox(db) + await store.setCursor(mailboxId, 'cursor-1') + + // Holder A claims, then (in the scenario this guards against) overruns + // its lease before ever calling release. + const tokenA = await store.claimReconcileLease(mailboxId, 30_000) + expect(tokenA).not.toBeNull() + await expireReconcileLease(db, mailboxId) + + // Holder B — a legitimate successor — claims the now-expired lease and + // is actively working. + const tokenB = await store.claimReconcileLease(mailboxId, 30_000) + expect(tokenB).not.toBeNull() + expect(tokenB).not.toBe(tokenA) + + // A finally reaches its own (deliberately delayed) release, using ITS + // stale token. + await store.releaseReconcileLease(mailboxId, tokenA as string) + + // B's lease must still be held — A's stale release must not have + // cleared it out from under B. + const rows = await db.query<{ claimed_until: Date | null }>( + 'SELECT claimed_until FROM gmail_watch_state WHERE mailbox_id = $1', + [mailboxId], + ) + expect(rows[0].claimed_until).not.toBeNull() + // And a third claimant must still be blocked by B's live lease. + expect(await store.claimReconcileLease(mailboxId, 30_000)).toBeNull() + }) + + it('after release, the lease is immediately claimable again — no need to wait out leaseMs', async () => { + const { db, store } = await freshStore() + const mailboxId = await insertMailbox(db) + await store.setCursor(mailboxId, 'cursor-1') + const token = await store.claimReconcileLease(mailboxId, 10 * 60_000) // a long lease + + await store.releaseReconcileLease(mailboxId, token as string) + + // Reclaimable right away — release does not wait out the original + // leaseMs, exactly like ConversationStore.releaseThreadLease. + expect(await store.claimReconcileLease(mailboxId, 30_000)).not.toBeNull() + }) + }) }) diff --git a/src/store/gmail-watch-state.ts b/src/store/gmail-watch-state.ts index 379c00d..26507f4 100644 --- a/src/store/gmail-watch-state.ts +++ b/src/store/gmail-watch-state.ts @@ -3,8 +3,11 @@ * (`gmail_watch_state`, migration 011, `src/db/migrate.ts`; gmail-push.md * §4 "the cursor"). One row per mailbox (`mailbox_id` is the PRIMARY KEY — * migration 011's doc comment), holding the `history_id` watermark - * `history.list` resumes from, plus `watch_expiration` (the `watch()` - * renewal deadline, gmail-push.md §6). + * `history.list` resumes from, `watch_expiration` (the `watch()` renewal + * deadline, gmail-push.md §6), and `claimed_until` (the per-mailbox + * reconciliation lease, HT-48, migration 016, gmail-push.md §6) — + * {@link GmailWatchStateStore.claimReconcileLease}/ + * {@link GmailWatchStateStore.releaseReconcileLease} below. * * ## Why {@link GmailWatchStateStore.setCursor} upserts rather than a plain `UPDATE` * @@ -96,6 +99,95 @@ export interface GmailWatchStateStore { * as a safe no-op. */ setWatchExpiration(mailboxId: string, watchExpiration: Date): Promise + + /** + * Claim `mailboxId`'s reconciliation lease for `leaseMs` (HT-48, gmail- + * push.md §6): an atomic `UPDATE ... SET claimed_until = now() + leaseMs + * WHERE mailbox_id = $1 AND (claimed_until IS NULL OR claimed_until < + * now()) RETURNING claimed_until` — the inbound analogue of + * `ConversationStore.claimThreadForDelivery` (sending.md §3a, migration + * 003). Ordinary Postgres row-level locking on the `UPDATE` makes "at + * most one claimant wins" hold under true concurrency exactly as it does + * there — no advisory lock needed. + * + * Returns an opaque **lease token** (the exact `claimed_until` value THIS + * call just wrote, as Postgres's own `::text` rendering of it — see below + * for why text, not a `Date`) iff this call won the claim, or `null` if + * another holder's lease is still live (or no `gmail_watch_state` row + * exists yet for this mailbox — `src/mail/gmail-reconcile.ts` only ever + * calls this after {@link getCursor} has already confirmed a row with a + * non-null cursor exists, so that case is not expected in practice). + * Unlike `claimThreadForDelivery`, there is no accompanying status + * re-check: this lease guards nothing but redundant Gmail API work + * (gmail-push.md §6), so a `null` return means "another run already holds + * this mailbox — try again shortly," never "this work already happened + * elsewhere and must not be repeated" (that correctness property is the + * ingest pipeline's dedup, inbound-ingestion.md §4, not this lease). + * + * The caller MUST pass this token back to {@link releaseReconcileLease} + * to prove it still owns the lease it is releasing — see that method's + * doc for the stale-holder scenario this guards against. + * + * ## Why the token is `claimed_until` rendered as text, not a `Date` + * + * `claimed_until` is `timestamptz`, which Postgres stores with + * microsecond precision; `now()` routinely produces a non-zero + * microsecond remainder. A `pg`-wire-protocol driver parses a `timestamptz` + * column into a JS `Date`, which only carries MILLISECOND precision — the + * sub-millisecond remainder is silently truncated on the way out. If + * {@link releaseReconcileLease} compared `claimed_until = $2` against a + * `Date` round-tripped through JS, the truncated value would almost never + * bit-for-bit equal what is actually stored, and every legitimate release + * would silently fail to match (falling into the "already superseded" + * no-op path below) — reintroducing the exact lock-out this token exists + * to prevent, but permanently, since the lease would then never be + * released until natural expiry. Casting to `::text` in the `RETURNING` + * clause instead hands back Postgres's own full-precision textual + * rendering; passing that same string back and casting it `::timestamptz` + * in the release `WHERE` clause compares against the identical value with + * no lossy JS `Date` round-trip in between. Callers must treat this return + * value as an opaque token — never parse it as a `Date` or do arithmetic + * on it. + */ + claimReconcileLease(mailboxId: string, leaseMs: number): Promise + + /** + * Release `mailboxId`'s reconciliation lease — but ONLY if `leaseToken` + * (the value {@link claimReconcileLease} returned when it granted this + * run's claim) still matches the row's CURRENT `claimed_until`: `UPDATE + * gmail_watch_state SET claimed_until = NULL WHERE mailbox_id = $1 AND + * claimed_until = $2::timestamptz`. + * + * ## Why this must be conditioned on the token, not unconditional + * + * An unconditional release (`WHERE mailbox_id = $1` alone) has a + * stale-holder hole: if THIS run overran `leaseMs` (e.g. a large backlog's + * `history.list`/`messages.get` batch), the lease already expired and a + * SUCCESSOR run may have claimed it and be actively working. This run's + * release would then clear the successor's LIVE lease out from under it, + * letting a third trigger claim and duplicate the successor's in-flight + * `history.list`/`messages.get` work — exactly the redundant-work case + * the lease exists to prevent, and worst under the fat-batch load that + * makes overrunning `leaseMs` most likely. Scoping the release to the + * token this call was granted makes it a no-op once that token no longer + * matches — this run's lease was already superseded, so there is nothing + * of ITS to release. + * + * Zero rows matched (the token doesn't match the current `claimed_until`, + * because it was already released and reclaimed by a successor, or the + * `gmail_watch_state` row no longer exists) is therefore a SILENT no-op, + * not an error — unlike `ConversationStore.releaseThreadLease`'s + * throw-on-zero-rows contract, which this deliberately does NOT mirror: + * that method's zero-rows case signals a genuine anomaly (the outbound + * lease is unconditional, so zero rows there can only mean the row + * vanished), whereas here zero rows is the routine, expected outcome of + * "our lease was already superseded" and must not be escalated into a + * caller-visible failure — the caller (`src/mail/gmail-reconcile.ts`) + * already wraps this in its own try/catch purely as a backstop for + * genuine, unexpected DB errors (a connection failure, say), not for this + * expected case. + */ + releaseReconcileLease(mailboxId: string, leaseToken: string): Promise } /** Create a {@link GmailWatchStateStore} backed by `db`. */ @@ -141,5 +233,39 @@ export function createGmailWatchStateStore(db: Db): GmailWatchStateStore { [mailboxId, watchExpiration], ) }, + + async claimReconcileLease(mailboxId, leaseMs) { + // A single UPDATE is already atomic with respect to itself under + // Postgres row-level locking — see ConversationStore + // .claimThreadForDelivery's identical reasoning. No status re-check + // here (unlike that method): this lease has no outcome to protect, + // only redundant Gmail API work to avoid (see the interface doc). + // + // `claimed_until::text` hands back the FULL-PRECISION value this call + // just wrote, as a plain string — see the interface doc's "Why the + // token is text, not a Date" for why the release side must compare + // against this exact textual round-trip rather than a JS `Date`. + const rows = await db.query<{ claimed_until: string }>( + `UPDATE gmail_watch_state + SET claimed_until = now() + ($2::double precision * interval '1 millisecond') + WHERE mailbox_id = $1 + AND (claimed_until IS NULL OR claimed_until < now()) + RETURNING claimed_until::text AS claimed_until`, + [mailboxId, leaseMs], + ) + return rows.length > 0 ? rows[0].claimed_until : null + }, + + async releaseReconcileLease(mailboxId, leaseToken) { + // Scoped to the token this call was granted (interface doc's "Why + // this must be conditioned on the token" section) — zero rows matched + // means our lease was already superseded (expired and reclaimed by a + // successor) or the row is gone, and is a silent no-op either way, + // never a throw (unlike ConversationStore.releaseThreadLease). + await db.query( + 'UPDATE gmail_watch_state SET claimed_until = NULL WHERE mailbox_id = $1 AND claimed_until = $2::timestamptz', + [mailboxId, leaseToken], + ) + }, } }