From e159156357b1f46d308d1e8b60f4cb7b8bd31974 Mon Sep 17 00:00:00 2001 From: Kris Hagel Date: Fri, 7 Aug 2026 09:04:55 -0700 Subject: [PATCH 1/9] fix(google-content-sync): stop drive-scoped change entries from poisoning the sync queue The prod alarm aistudio-prod-google-content-sync-oldest-message fired because two connectors (b1429b0b-..., 1f870c92-...) had synced nothing since 2026-08-06 ~12:00 UTC and 904 messages had accumulated in the DLQ. Root cause ---------- Google's Drive Changes feed (GET /drive/v3/changes) is not exclusively file-scoped. Entries with changeType "drive" describe the Shared Drive itself - rename, membership change, restriction change - and carry NO fileId. driveChangeSchema in lib/repositories/google-drive/drive-client.ts declared fileId as a required z.string(), so changesListSchema.parse() threw a ZodError inside listChanges the moment such an entry appeared in a page. The throw happened before reconcileChanges could persist the cursor, so every SQS redelivery re-fetched the identical page and failed identically. The message never drained, its age breached the 30-minute alarm threshold, and after maxReceiveCount it landed in the DLQ - repeatedly. Changes ------- lib/repositories/google-drive/drive-client.ts - driveChangeSchema.fileId is now z.string().optional(). - Added changeType: z.string().optional(). Deliberately a plain string and NOT a z.enum: an enum would re-poison the queue the first time Google ships a changeType value we did not anticipate. - listChanges now requests changeType in its fields projection, so the API actually returns the discriminator the guard reads. Without this the projection would silently strip it and every entry would look untyped. infra/lambdas/google-content-sync/changes.ts (new) - isFileScopedDriveChange(): the single place that decides whether a change entry names a file the sync path can act on. Returns false for changeType === "drive" and for a missing/empty fileId; a type predicate so the caller gets fileId narrowed to string. Extracted into its own module because index.ts is not importable from the root Jest gate. infra/lambdas/google-content-sync/index.ts - processGoogleDriveChange() now guards FIRST: a non-file-scoped entry is logged and skipped with return false, before markSourceMissing() or any other side effect can run. Returning false leaves requiresSelectionSnapshot untouched, so reconcileChanges still persists the cursor for that page and the loop advances. - Drive deletion and access loss remain covered by the existing 403/404 -> markConnectorAccessLost path; nothing about that behaviour changes. - Extracted the catch-block outcome recording into recordDriveChangeFailure() - behaviour-identical, and it brings the function back under the complexity-15 lint ceiling that the new guard pushed it over. Tests ----- - tests/unit/lib/repositories/google-drive-changes-feed.test.ts: a mixed page containing a changeType "drive" entry with no fileId plus a normal file entry now parses and returns both values; an unknown future changeType parses; and the fields projection is asserted to request changeType. - tests/unit/google-content-sync-change-scope.test.ts: full truth table for isFileScopedDriveChange (drive-scope wins even if a fileId is present, empty/missing fileId skipped, unknown changeType with a fileId processed), the fileId type narrowing, plus source-level assertions that the guard precedes the first side effect in processGoogleDriveChange and that reconcileChanges still persists the cursor on the skip path. Operational notes ----------------- No alarm thresholds changed. Existing DLQ messages are intentionally left to expire under the 14-day retention rather than being redriven or purged. The affected connectors self-heal on their next webhook or scheduled tick once the Lambda is deployed. --- infra/lambdas/google-content-sync/changes.ts | 35 +++++++ infra/lambdas/google-content-sync/index.ts | 54 ++++++++--- lib/repositories/google-drive/drive-client.ts | 15 ++- .../google-content-sync-change-scope.test.ts | 94 +++++++++++++++++++ .../google-drive-changes-feed.test.ts | 89 ++++++++++++++++++ 5 files changed, 270 insertions(+), 17 deletions(-) create mode 100644 infra/lambdas/google-content-sync/changes.ts create mode 100644 tests/unit/google-content-sync-change-scope.test.ts create mode 100644 tests/unit/lib/repositories/google-drive-changes-feed.test.ts diff --git a/infra/lambdas/google-content-sync/changes.ts b/infra/lambdas/google-content-sync/changes.ts new file mode 100644 index 000000000..81377cd64 --- /dev/null +++ b/infra/lambdas/google-content-sync/changes.ts @@ -0,0 +1,35 @@ +/** + * Classification helpers for entries returned by the Google Drive Changes + * feed (`GET /drive/v3/changes`). + * + * The feed is not exclusively file-scoped. Google emits `changeType: "drive"` + * entries for Shared Drive-level events — rename, membership change, + * restriction change — and those entries carry NO `fileId`. There is no file + * to import, resolve, or mark missing for them. + * + * Every downstream step of the reconcile loop is keyed on a file id, so a + * `fileId`-less entry must be skipped before that work begins. `changeType` is + * matched as a plain string rather than an enum so a future value Google adds + * cannot turn the queue into poison messages again. + */ + +/** Shared Drive-scoped change entries never carry a `fileId`. */ +export const DRIVE_SCOPED_CHANGE_TYPE = "drive"; + +/** The minimum shape needed to decide whether a change entry is actionable. */ +export interface DriveChangeIdentity { + changeType?: string | undefined; + fileId?: string | undefined; +} + +/** + * True when the entry names a specific Drive file, i.e. it is safe to hand to + * the file-scoped import/removal path. Drive-scoped entries and entries with a + * missing or empty `fileId` return false and must be skipped. + */ +export function isFileScopedDriveChange( + change: T, +): change is T & { fileId: string } { + if (change.changeType === DRIVE_SCOPED_CHANGE_TYPE) return false; + return typeof change.fileId === "string" && change.fileId.length > 0; +} diff --git a/infra/lambdas/google-content-sync/index.ts b/infra/lambdas/google-content-sync/index.ts index b45cfff5d..b4e7ce5a3 100644 --- a/infra/lambdas/google-content-sync/index.ts +++ b/infra/lambdas/google-content-sync/index.ts @@ -76,6 +76,7 @@ import { } from "../../../lib/repositories/google-drive/formats"; import { refreshGoogleAccessToken } from "../../../lib/repositories/google-drive/oauth"; import { getGoogleContentWifAccessToken } from "../../../lib/repositories/google-drive/wif"; +import { isFileScopedDriveChange } from "./changes"; import { assertGoogleSourceMetadataSize, assertGoogleSourceResponseSize, @@ -1332,12 +1333,49 @@ type GoogleDriveChange = Awaited< ReturnType >["values"][number]; +/** + * Records the outcome of a failed file-scoped change. Connector lifecycle + * errors are fatal to the whole run and are rethrown; a Drive 403/404 means the + * file was deleted or access was lost, which is the markSourceMissing path. + * Anything else is rethrown so the caller can fail the record. + */ +async function recordDriveChangeFailure( + context: ConnectorContext, + fileId: string, + error: unknown, + counters: SyncCounters, +): Promise { + if ( + error instanceof ConnectorRevokedError || + error instanceof ConnectorPausedError || + error instanceof ConnectorSelectionChangedError + ) { + throw error; + } + if (!isGoogleDriveMissingError(error)) throw error; + if (await markSourceMissing(context, fileId)) { + counters.missing += 1; + } else { + counters.failed += 1; + } +} + async function processGoogleDriveChange( context: ConnectorContext, client: GoogleDriveClient, change: GoogleDriveChange, counters: SyncCounters, ): Promise { + // Shared Drive-scoped entries (rename, membership, restriction changes) have + // no fileId. Nothing below can act on them, and they must not stall the + // cursor: skip without side effects so the loop keeps advancing. + if (!isFileScopedDriveChange(change)) { + log.info("Skipped non-file Google Drive change entry", { + connectorId: context.connector.id, + changeType: change.changeType ?? null, + }); + return false; + } if (change.removed || !change.file || change.file.trashed) { if (await markSourceMissing(context, change.fileId)) { counters.missing += 1; @@ -1363,21 +1401,7 @@ async function processGoogleDriveChange( ); return false; } catch (error) { - if ( - error instanceof ConnectorRevokedError || - error instanceof ConnectorPausedError || - error instanceof ConnectorSelectionChangedError - ) { - throw error; - } - if (!isGoogleDriveMissingError(error)) throw error; - if ( - await markSourceMissing(context, change.fileId) - ) { - counters.missing += 1; - } else { - counters.failed += 1; - } + await recordDriveChangeFailure(context, change.fileId, error, counters); return false; } } diff --git a/lib/repositories/google-drive/drive-client.ts b/lib/repositories/google-drive/drive-client.ts index b6ad170b2..473a1a781 100644 --- a/lib/repositories/google-drive/drive-client.ts +++ b/lib/repositories/google-drive/drive-client.ts @@ -56,8 +56,19 @@ const filesListSchema = z.object({ files: z.array(googleDriveFileSchema).default([]), }); +/** + * Google's Changes feed is not exclusively file-scoped. Entries with + * `changeType: "drive"` describe the Shared Drive itself (rename, membership, + * restriction changes) and carry NO `fileId`. Requiring `fileId` here made + * every such entry a poison message: `listChanges` threw before the cursor + * advanced, so the Lambda retried the same page forever. + * + * `changeType` is deliberately a plain string rather than an enum — a new + * value Google adds later must not re-poison the queue. + */ const driveChangeSchema = z.object({ - fileId: z.string(), + changeType: z.string().optional(), + fileId: z.string().optional(), removed: z.boolean().optional().default(false), time: z.string().datetime({ offset: true }).optional(), driveId: z.string().optional(), @@ -267,7 +278,7 @@ export class GoogleDriveClient { url.searchParams.set("includeItemsFromAllDrives", "true"); url.searchParams.set( "fields", - `nextPageToken,newStartPageToken,changes(fileId,removed,time,driveId,file(${FILE_FIELDS}))`, + `nextPageToken,newStartPageToken,changes(changeType,fileId,removed,time,driveId,file(${FILE_FIELDS}))`, ); if (driveId) { url.searchParams.set("driveId", driveId); diff --git a/tests/unit/google-content-sync-change-scope.test.ts b/tests/unit/google-content-sync-change-scope.test.ts new file mode 100644 index 000000000..6770b07ad --- /dev/null +++ b/tests/unit/google-content-sync-change-scope.test.ts @@ -0,0 +1,94 @@ +/** @jest-environment node */ + +import fs from "node:fs"; +import path from "node:path"; +import { + DRIVE_SCOPED_CHANGE_TYPE, + isFileScopedDriveChange, +} from "../../infra/lambdas/google-content-sync/changes"; +import { stripComments } from "../helpers/strip-ts-comments"; + +describe("Google Drive change scope classification", () => { + test("skips Shared Drive-scoped entries even when a fileId is present", () => { + expect( + isFileScopedDriveChange({ changeType: DRIVE_SCOPED_CHANGE_TYPE }), + ).toBe(false); + expect( + isFileScopedDriveChange({ + changeType: DRIVE_SCOPED_CHANGE_TYPE, + fileId: "file-1", + }), + ).toBe(false); + }); + + test("skips entries with a missing or empty fileId", () => { + expect(isFileScopedDriveChange({})).toBe(false); + expect(isFileScopedDriveChange({ changeType: "file" })).toBe(false); + expect(isFileScopedDriveChange({ fileId: "" })).toBe(false); + expect(isFileScopedDriveChange({ fileId: undefined })).toBe(false); + }); + + test("processes file-scoped entries, including ones with no changeType", () => { + expect(isFileScopedDriveChange({ changeType: "file", fileId: "f1" })).toBe( + true, + ); + expect(isFileScopedDriveChange({ fileId: "f1" })).toBe(true); + }); + + test("treats unknown change types as file-scoped when a fileId is present", () => { + // A future changeType value must not become a poison message: if Google + // gave us a fileId, the file path can still act on it. + expect( + isFileScopedDriveChange({ changeType: "someFutureScope", fileId: "f1" }), + ).toBe(true); + expect(isFileScopedDriveChange({ changeType: "someFutureScope" })).toBe( + false, + ); + }); + + test("narrows fileId to a string for the file-scoped branch", () => { + const change: { changeType?: string; fileId?: string } = { + changeType: "file", + fileId: "file-1", + }; + if (isFileScopedDriveChange(change)) { + const fileId: string = change.fileId; + expect(fileId).toBe("file-1"); + } else { + throw new Error("expected the entry to be file-scoped"); + } + }); +}); + +describe("processGoogleDriveChange guard wiring", () => { + const source = stripComments( + fs.readFileSync( + path.join(process.cwd(), "infra/lambdas/google-content-sync/index.ts"), + "utf8", + ), + ); + + test("guards the change handler before any file-scoped work", () => { + const body = source.slice( + source.indexOf("async function processGoogleDriveChange("), + ); + const guardIndex = body.indexOf("isFileScopedDriveChange(change)"); + const firstSideEffect = body.indexOf("markSourceMissing("); + + expect(guardIndex).toBeGreaterThan(-1); + expect(firstSideEffect).toBeGreaterThan(-1); + // The skip must come first: nothing downstream can act on a fileId-less + // entry, and the reconcile loop still advances the cursor afterwards. + expect(guardIndex).toBeLessThan(firstSideEffect); + }); + + test("the reconcile loop persists the cursor for skipped entries", () => { + const loop = source.slice( + source.indexOf("async function reconcileChanges("), + ); + // processGoogleDriveChange returns false for a skipped entry, which leaves + // requiresSelectionSnapshot false, so persistSyncCursor runs each page. + expect(loop).toContain("if (!requiresSelectionSnapshot) {"); + expect(loop).toContain("await persistSyncCursor("); + }); +}); diff --git a/tests/unit/lib/repositories/google-drive-changes-feed.test.ts b/tests/unit/lib/repositories/google-drive-changes-feed.test.ts new file mode 100644 index 000000000..4c74a5fe3 --- /dev/null +++ b/tests/unit/lib/repositories/google-drive-changes-feed.test.ts @@ -0,0 +1,89 @@ +/** @jest-environment node */ + +import { GoogleDriveClient } from "@/lib/repositories/google-drive/drive-client"; + +function jsonResponse(body: unknown, status = 200): Response { + return { + ok: status >= 200 && status < 300, + status, + headers: new Headers({ "content-type": "application/json" }), + json: async () => body, + } as unknown as Response; +} + +describe("GoogleDriveClient changes feed", () => { + test("accepts Shared Drive-scoped change entries that carry no fileId", async () => { + // Regression: Google emits changeType "drive" entries (Shared Drive + // rename/membership/restriction events) with no fileId. Requiring fileId + // made listChanges throw before the cursor advanced, so the sync Lambda + // retried the same page forever and the queue filled with poison messages. + const fetchMock = jest + .fn, Parameters>() + .mockResolvedValue( + jsonResponse({ + newStartPageToken: "start-2", + changes: [ + { + changeType: "drive", + time: "2026-08-06T12:00:00.000Z", + driveId: "drive-1", + }, + { + changeType: "file", + fileId: "file-1", + time: "2026-08-06T12:00:01.000Z", + }, + ], + }), + ); + + const page = await new GoogleDriveClient("token", { + fetch: fetchMock, + }).listChanges("cursor-1", "drive-1"); + + expect(page.values).toHaveLength(2); + expect(page.values[0]).toEqual( + expect.objectContaining({ changeType: "drive", driveId: "drive-1" }), + ); + expect(page.values[0]?.fileId).toBeUndefined(); + expect(page.values[1]).toEqual( + expect.objectContaining({ changeType: "file", fileId: "file-1" }), + ); + expect(page.newStartPageToken).toBe("start-2"); + }); + + test("tolerates change types Google has not shipped yet", async () => { + const fetchMock = jest + .fn, Parameters>() + .mockResolvedValue( + jsonResponse({ + newStartPageToken: "start-3", + changes: [{ changeType: "someFutureScope" }], + }), + ); + + const page = await new GoogleDriveClient("token", { + fetch: fetchMock, + }).listChanges("cursor-1"); + + expect(page.values).toEqual([ + expect.objectContaining({ changeType: "someFutureScope" }), + ]); + }); + + test("requests changeType in the changes fields projection", async () => { + const fetchMock = jest + .fn, Parameters>() + .mockResolvedValue(jsonResponse({ newStartPageToken: "start-1" })); + + await new GoogleDriveClient("token", { fetch: fetchMock }).listChanges( + "cursor-1", + ); + + const fields = new URL( + String(fetchMock.mock.calls[0]?.[0]), + ).searchParams.get("fields"); + expect(fields).toContain("changes(changeType,fileId,"); + }); + +}); From bf5378cf9f009bb7dd50f8aaa1795f4489d15585 Mon Sep 17 00:00:00 2001 From: Kris Hagel Date: Fri, 7 Aug 2026 09:29:45 -0700 Subject: [PATCH 2/9] fix(google-content-sync): rebuild the selection when a Shared Drive is removed Review follow-up (Codex P1 on PR #1617). The first commit skipped every non-file-scoped change entry unconditionally. That is correct for the metadata noise the outage was actually about - Shared Drive rename, membership change, restriction change - but it also silently consumed the one drive-scoped entry that carries real meaning: removed: true, which Google emits when the Shared Drive is deleted or the account loses access to it. Consuming that is a data-correctness bug. Sources already imported from that drive become unreachable, and the connector gets no other signal: - A connector scoped to a shared drive would eventually surface the loss, because its next listChanges/watch call against that driveId returns 403/404 and the existing markConnectorAccessLost path fires. - A connector reading the global changes feed would NOT. Its subsequent listChanges and watch calls keep succeeding, so the stale content stays active indefinitely with nothing to retire it. Fix --- infra/lambdas/google-content-sync/changes.ts - New isDriveRemovalChange(): true when an entry is not file-scoped and reports removed. Deliberately NOT keyed on changeType === "drive" - the legacy "teamDrive" value and any future scope Google introduces must take this path too. What identifies the case is "removed, and not about a specific file", which the caller has already established. - DriveChangeIdentity gained the optional removed flag. infra/lambdas/google-content-sync/index.ts - processGoogleDriveChange() now returns isDriveRemovalChange(change) from the non-file-scoped branch rather than a bare false. Returning true sets requiresSelectionSnapshot, so reconcileChanges runs reconcileSelectionSnapshot -> markUnseenSourcesMissing and retires whatever is no longer reachable; a hard access loss still surfaces through the existing 403/404 -> markConnectorAccessLost path. - The cursor advances on both branches. Skips take the per-page persistSyncCursor; removals take the post-snapshot persistSyncCursor. The invariant the outage broke stays intact either way. - The skip/removal log line now also carries driveId and time, so a spike in either can be correlated against Drive audit logs. Tests ----- tests/unit/google-content-sync-change-scope.test.ts - Truth table for isDriveRemovalChange: drive-scoped removal flagged; "teamDrive", an unknown future scope, and a changeType-less removal all covered; rename/membership noise (removed false or absent) ignored; a file-scoped removal explicitly NOT claimed, since that is the file path's job via markSourceMissing; a drive-scoped entry that also carries a fileId still treated as drive-scoped. - Wiring assertions that the removal signal is returned rather than swallowed, and that reconcileChanges persists the cursor on BOTH branches (asserted as exactly two persistSyncCursor call sites in the loop). Verification: eslint --max-warnings 0 clean; 573 suites / 5434 tests passing (the single failing suite, tests/unit/agent-workspace-identity.test.ts, fails identically on the clean tree - @aws-sdk/client-ecs is absent from package.json and bun.lock entirely). --- infra/lambdas/google-content-sync/changes.ts | 22 +++++ infra/lambdas/google-content-sync/index.ts | 29 +++++-- .../google-content-sync-change-scope.test.ts | 82 ++++++++++++++++++- 3 files changed, 121 insertions(+), 12 deletions(-) diff --git a/infra/lambdas/google-content-sync/changes.ts b/infra/lambdas/google-content-sync/changes.ts index 81377cd64..3ffcbdb02 100644 --- a/infra/lambdas/google-content-sync/changes.ts +++ b/infra/lambdas/google-content-sync/changes.ts @@ -20,6 +20,7 @@ export const DRIVE_SCOPED_CHANGE_TYPE = "drive"; export interface DriveChangeIdentity { changeType?: string | undefined; fileId?: string | undefined; + removed?: boolean | undefined; } /** @@ -33,3 +34,24 @@ export function isFileScopedDriveChange( if (change.changeType === DRIVE_SCOPED_CHANGE_TYPE) return false; return typeof change.fileId === "string" && change.fileId.length > 0; } + +/** + * True when a non-file-scoped entry reports that the Shared Drive itself went + * away — deleted, or this account lost access to it. + * + * These entries must NOT be skipped like the other drive-scoped noise. Sources + * already imported from that drive are now unreachable, and a connector + * reading the global changes feed gets no other signal: its subsequent + * `listChanges` and watch calls keep succeeding, so the stale content would + * stay active indefinitely. The caller answers this by requesting a selection + * snapshot, which retires whatever is no longer reachable. + * + * Deliberately not keyed on `changeType === "drive"`: the legacy "teamDrive" + * value and any future scope Google introduces must take this path too. What + * identifies the case is "removed, and not about a specific file" — which is + * exactly the state the caller has already established. + */ +export function isDriveRemovalChange(change: DriveChangeIdentity): boolean { + if (isFileScopedDriveChange(change)) return false; + return change.removed === true; +} diff --git a/infra/lambdas/google-content-sync/index.ts b/infra/lambdas/google-content-sync/index.ts index b4e7ce5a3..123d656ce 100644 --- a/infra/lambdas/google-content-sync/index.ts +++ b/infra/lambdas/google-content-sync/index.ts @@ -76,7 +76,7 @@ import { } from "../../../lib/repositories/google-drive/formats"; import { refreshGoogleAccessToken } from "../../../lib/repositories/google-drive/oauth"; import { getGoogleContentWifAccessToken } from "../../../lib/repositories/google-drive/wif"; -import { isFileScopedDriveChange } from "./changes"; +import { isDriveRemovalChange, isFileScopedDriveChange } from "./changes"; import { assertGoogleSourceMetadataSize, assertGoogleSourceResponseSize, @@ -1367,14 +1367,27 @@ async function processGoogleDriveChange( counters: SyncCounters, ): Promise { // Shared Drive-scoped entries (rename, membership, restriction changes) have - // no fileId. Nothing below can act on them, and they must not stall the - // cursor: skip without side effects so the loop keeps advancing. + // no fileId, so nothing on the file path below can act on them. if (!isFileScopedDriveChange(change)) { - log.info("Skipped non-file Google Drive change entry", { - connectorId: context.connector.id, - changeType: change.changeType ?? null, - }); - return false; + // A removal, though, is not noise: the Shared Drive was deleted or this + // account lost access, so sources imported from it are now unreachable. + // Requesting a selection snapshot lets markUnseenSourcesMissing retire + // them; a hard access loss still surfaces through the existing 403/404 + // path. Anything else is metadata noise and is skipped without side + // effects so the cursor keeps advancing. + const driveRemoved = isDriveRemovalChange(change); + log.info( + driveRemoved + ? "Shared Drive removed from the change feed" + : "Skipped non-file Google Drive change entry", + { + connectorId: context.connector.id, + changeType: change.changeType ?? null, + driveId: change.driveId ?? null, + time: change.time ?? null, + }, + ); + return driveRemoved; } if (change.removed || !change.file || change.file.trashed) { if (await markSourceMissing(context, change.fileId)) { diff --git a/tests/unit/google-content-sync-change-scope.test.ts b/tests/unit/google-content-sync-change-scope.test.ts index 6770b07ad..d8bb58db9 100644 --- a/tests/unit/google-content-sync-change-scope.test.ts +++ b/tests/unit/google-content-sync-change-scope.test.ts @@ -4,6 +4,7 @@ import fs from "node:fs"; import path from "node:path"; import { DRIVE_SCOPED_CHANGE_TYPE, + isDriveRemovalChange, isFileScopedDriveChange, } from "../../infra/lambdas/google-content-sync/changes"; import { stripComments } from "../helpers/strip-ts-comments"; @@ -60,6 +61,63 @@ describe("Google Drive change scope classification", () => { }); }); +describe("Shared Drive removal detection", () => { + test("flags a drive-scoped removal so the caller rebuilds the selection", () => { + // The Shared Drive was deleted or access was lost. Sources imported from + // it are unreachable, and a connector reading the global changes feed gets + // no other signal — its later listChanges calls keep succeeding. + expect( + isDriveRemovalChange({ + changeType: DRIVE_SCOPED_CHANGE_TYPE, + removed: true, + }), + ).toBe(true); + }); + + test("covers legacy and future drive-like scopes, not just \"drive\"", () => { + expect( + isDriveRemovalChange({ changeType: "teamDrive", removed: true }), + ).toBe(true); + expect( + isDriveRemovalChange({ changeType: "someFutureScope", removed: true }), + ).toBe(true); + expect(isDriveRemovalChange({ removed: true })).toBe(true); + }); + + test("ignores drive-scoped metadata noise", () => { + // Rename, membership and restriction changes carry removed: false. + expect( + isDriveRemovalChange({ + changeType: DRIVE_SCOPED_CHANGE_TYPE, + removed: false, + }), + ).toBe(false); + expect( + isDriveRemovalChange({ changeType: DRIVE_SCOPED_CHANGE_TYPE }), + ).toBe(false); + }); + + test("never claims a file-scoped removal — that is the file path's job", () => { + expect( + isDriveRemovalChange({ + changeType: "file", + fileId: "file-1", + removed: true, + }), + ).toBe(false); + }); + + test("treats a drive-scoped entry that also carries a fileId as drive-scoped", () => { + expect( + isDriveRemovalChange({ + changeType: DRIVE_SCOPED_CHANGE_TYPE, + fileId: "file-1", + removed: true, + }), + ).toBe(true); + }); +}); + describe("processGoogleDriveChange guard wiring", () => { const source = stripComments( fs.readFileSync( @@ -82,13 +140,29 @@ describe("processGoogleDriveChange guard wiring", () => { expect(guardIndex).toBeLessThan(firstSideEffect); }); - test("the reconcile loop persists the cursor for skipped entries", () => { + test("propagates the removal signal instead of swallowing it", () => { + const body = source.slice( + source.indexOf("async function processGoogleDriveChange("), + ); + // A drive removal must return true so reconcileChanges rebuilds the + // selection snapshot; returning a bare false here would strand every + // source imported from the drive that went away. + expect(body).toContain("isDriveRemovalChange(change)"); + expect(body).toContain("return driveRemoved;"); + }); + + test("the reconcile loop persists the cursor on both skip and removal", () => { const loop = source.slice( source.indexOf("async function reconcileChanges("), + source.indexOf("async function markConnectorAccessLost("), ); - // processGoogleDriveChange returns false for a skipped entry, which leaves - // requiresSelectionSnapshot false, so persistSyncCursor runs each page. + // Skipped entry: returns false, requiresSelectionSnapshot stays false, so + // the per-page persist runs. expect(loop).toContain("if (!requiresSelectionSnapshot) {"); - expect(loop).toContain("await persistSyncCursor("); + // Removal: returns true, so the cursor is persisted after the snapshot. + expect(loop).toContain("if (requiresSelectionSnapshot) {"); + expect(loop).toContain("await reconcileSelectionSnapshot("); + // Either way the cursor advances — that is the invariant the outage broke. + expect(loop.match(/await persistSyncCursor\(/g)).toHaveLength(2); }); }); From 4ff6a690fde141829136b150077db25d20bd5444 Mon Sep 17 00:00:00 2001 From: Kris Hagel Date: Fri, 7 Aug 2026 14:43:28 -0700 Subject: [PATCH 3/9] fix(google-content-sync): harden the connector against the poison-page class Follow-up to the drive-scoped change-entry hotfix. That fix removed one specific poison entry; these three changes remove the failure class it belonged to, plus the blind spot that let it ship. 1. Per-entry parsing in the Drive client (lib/repositories/google-drive) changesListSchema and filesListSchema parsed their arrays as z.array(entrySchema) and .parse()d the whole payload, so ONE malformed entry among a thousand rejected the entire page. listChanges then threw before the cursor advanced and the connector retried the same page forever - the exact shape of the fileId incident. Two-pronged: (a) Validators that were stricter than Google's actual contract were loosened, each after checking its downstream reader: - shortcutDetails.targetId / targetMimeType -> optional. Only targetId is dereferenced (resolveShortcut, and the selections route), and both already raise a graceful "no target" error. targetMimeType has no reader anywhere, so requiring it could only ever poison a page. - modifiedTime / change.time: .datetime({offset:true}) -> plain string. Consumers are new Date(...) and a revision string. - size: the ^\d+$ regex is gone. assertGoogleSourceMetadataSize re-checks the numeric shape and the streaming byte bound is authoritative regardless. - webViewLink / iconLink: .url() -> plain string, but a non-http(s) scheme is DROPPED rather than persisted, so loosening the validator cannot open an href injection path. - file, driveId, parents, owners, trashed and the optional strings accept an explicit null as well as an omission. (b) Defense in depth: the list envelopes now parse entries as z.unknown() and safeParse each one individually. A failing entry is reported on the page as a GoogleDriveSkippedEntry (index, any extractable id, the zod issues) and DROPPED. It is never reinterpreted - in particular markSourceMissing is never called for it, because a parse failure says nothing about whether the file still exists and reclassifying it would trade a stalled queue for silent data loss. The accepted cost (a missed update until that file changes again) is stated in the log line the Lambda emits. Because a dropped entry makes an enumeration incomplete, the snapshot path now gates markUnseenSourcesMissing behind shouldRetireUnseenSources(): retiring unseen sources is only sound after a COMPLETE pass, so the sweep is skipped (loudly) and left to the next clean snapshot. 2. Durable selection-snapshot obligation (infra/lambdas/google-content-sync) reconcileChanges deferred persistSyncCursor for the whole run once any page demanded a selection snapshot. A 900s Lambda timeout or a snapshot budget failure therefore threw away every already-processed change page and replayed them from the original cursor on the next attempt - indefinitely if the snapshot kept failing. The obligation is now durable instead of implied by cursor position: a selectionSnapshotPendingAt timestamp on repository_connectors.metadata (existing jsonb - no migration), written before the cursor moves past the page that raised it, with the cursor advancing per page exactly as the non-snapshot path already did. The flag is cleared only after a COMPLETE snapshot, and every sync start discharges an outstanding obligation before consuming new changes. The flag is written with a jsonb merge/delete rather than a read-modify-write so it cannot clobber the webhook path's concurrent metadata writes. The loop itself moved to reconcile.ts behind injected collaborators, so the cursor/obligation ordering is provable without a database, Google credentials, or the Lambda's module-level environment requirements. Mid-enumeration checkpointing remains out of scope; partial enumeration still never feeds markUnseenSourcesMissing. 3. The lambda is now type-checked infra/lambdas/google-content-sync had no tsconfig.json: excluded from the root tsc ("exclude": ["infra"]) and from the infra tsc ("exclude": ["lambdas/**/*"]), with esbuild only transpiling. Adds a tsconfig.json modeled on unified-content-processor (extends the root config so the lambda's repo-root lib/ imports resolve) and a typecheck:google-content-sync script chained into infra's build, which is what the cdk-validate CI job runs - the same route typecheck:unified-content takes. Fixes the P2 that motivated it: GoogleDriveChange was re-derived locally via Awaited>["values"][number] and now imports the exported type. Tests: new google-content-sync-reconcile.test.ts covers cursor-advances- per-page-while-pending, flag-written-before-the-cursor-moves, marked once, cleared only on a complete snapshot, retained on snapshot throw, and the unseen-sweep gate. google-drive-changes-feed.test.ts gains real-payload coverage for every loosened field (including the javascript:/data: drop) and for one-malformed-entry-among-many across listChanges, listChildren and listSharedDriveFiles. The reconcile source-string assertion in google-content-sync-change-scope.test.ts is replaced by those behavioral tests and now only pins that index.ts still delegates to the shared loop. --- infra/lambdas/google-content-sync/index.ts | 213 ++++++++++++--- .../lambdas/google-content-sync/reconcile.ts | 130 ++++++++++ .../lambdas/google-content-sync/tsconfig.json | 9 + infra/package.json | 3 +- lib/db/schema/tables/repository-connectors.ts | 10 + lib/repositories/google-drive/drive-client.ts | 188 +++++++++++--- .../google-content-sync-change-scope.test.ts | 17 +- .../google-content-sync-reconcile.test.ts | 242 ++++++++++++++++++ .../google-drive-changes-feed.test.ts | 223 ++++++++++++++++ 9 files changed, 953 insertions(+), 82 deletions(-) create mode 100644 infra/lambdas/google-content-sync/reconcile.ts create mode 100644 infra/lambdas/google-content-sync/tsconfig.json create mode 100644 tests/unit/google-content-sync-reconcile.test.ts diff --git a/infra/lambdas/google-content-sync/index.ts b/infra/lambdas/google-content-sync/index.ts index 123d656ce..799c689f2 100644 --- a/infra/lambdas/google-content-sync/index.ts +++ b/infra/lambdas/google-content-sync/index.ts @@ -66,7 +66,9 @@ import { GoogleDriveApiError, GoogleDriveClient, GoogleDriveDownloadPendingError, + type GoogleDriveChange, type GoogleDriveFile, + type GoogleDriveSkippedEntry, } from "../../../lib/repositories/google-drive/drive-client"; import { exportedGoogleDriveFileName, @@ -77,6 +79,13 @@ import { import { refreshGoogleAccessToken } from "../../../lib/repositories/google-drive/oauth"; import { getGoogleContentWifAccessToken } from "../../../lib/repositories/google-drive/wif"; import { isDriveRemovalChange, isFileScopedDriveChange } from "./changes"; +import { + isSelectionSnapshotPending, + reconcileChangePages, + resumePendingSelectionSnapshot, + SELECTION_SNAPSHOT_PENDING_KEY, + shouldRetireUnseenSources, +} from "./reconcile"; import { assertGoogleSourceMetadataSize, assertGoogleSourceResponseSize, @@ -865,10 +874,18 @@ type AddDiscoveredFile = ( selectedVia: string, ) => Promise; +/** Reports entries Google returned that failed per-entry validation. */ +type RecordSkippedEntries = ( + operation: string, + scopeId: string, + skipped: GoogleDriveSkippedEntry[], +) => void; + async function enumerateSharedDriveSelection( client: GoogleDriveClient, selection: RepositoryConnectorSelectionRow, add: AddDiscoveredFile, + recordSkipped: RecordSkippedEntries, ): Promise { let pageToken: string | null = null; do { @@ -876,6 +893,11 @@ async function enumerateSharedDriveSelection( selection.externalId, pageToken, ); + recordSkipped( + "listSharedDriveFiles", + selection.externalId, + page.skippedEntries, + ); for (const file of page.values) { await add(file, selection.externalId); } @@ -903,6 +925,7 @@ async function enumerateFileSelection( selection: RepositoryConnectorSelectionRow, budget: GoogleDriveSnapshotBudget, add: AddDiscoveredFile, + recordSkipped: RecordSkippedEntries, ): Promise { const selected = await client.getFile(selection.externalId); if ( @@ -923,6 +946,7 @@ async function enumerateFileSelection( let pageToken: string | null = null; do { const page = await client.listChildren(folderId, pageToken); + recordSkipped("listChildren", folderId, page.skippedEntries); await collectFolderChildren( page.values, selection.externalId, @@ -940,6 +964,7 @@ async function enumerateInitialFiles( ): Promise<{ files: Array<{ file: GoogleDriveFile; selectedVia: string[] }>; inaccessibleSelectionCount: number; + skippedEntryCount: number; }> { const discovered = new Map< string, @@ -947,6 +972,25 @@ async function enumerateInitialFiles( >(); const budget = new GoogleDriveSnapshotBudget(); let inaccessibleSelectionCount = 0; + let skippedEntryCount = 0; + const recordSkipped: RecordSkippedEntries = ( + operation, + scopeId, + skipped, + ) => { + if (skipped.length === 0) return; + skippedEntryCount += skipped.length; + log.error("Google Drive returned unparseable list entries", { + connectorId: context.connector.id, + operation, + scopeId, + skippedCount: skipped.length, + // The entries are dropped, never reinterpreted as removals. The + // enumeration is therefore incomplete, which suppresses + // markUnseenSourcesMissing for this snapshot. + entries: skipped, + }); + }; const add = async (candidate: GoogleDriveFile, selectedVia: string) => { const file = await resolveShortcut(client, candidate); if (file.trashed || file.mimeType === GOOGLE_FOLDER_MIME_TYPE) return; @@ -965,10 +1009,21 @@ async function enumerateInitialFiles( for (const selection of context.selections) { try { if (selection.selectionKind === "drive") { - await enumerateSharedDriveSelection(client, selection, add); + await enumerateSharedDriveSelection( + client, + selection, + add, + recordSkipped, + ); continue; } - await enumerateFileSelection(client, selection, budget, add); + await enumerateFileSelection( + client, + selection, + budget, + add, + recordSkipped, + ); } catch (error) { if ( selection.selectionKind !== "drive" && @@ -992,6 +1047,7 @@ async function enumerateInitialFiles( selectedVia: [...selectedVia], })), inaccessibleSelectionCount, + skippedEntryCount, }; } @@ -1254,21 +1310,38 @@ async function importFileIsolated( } } +/** + * Re-enumerate every selection. Resolves false when the enumeration was + * incomplete — the caller must not treat a snapshot obligation as discharged. + */ async function reconcileSelectionSnapshot( context: ConnectorContext, client: GoogleDriveClient, counters: SyncCounters, -): Promise { +): Promise { const snapshot = await enumerateInitialFiles(context, client); counters.discovered += snapshot.files.length; counters.failed += snapshot.inaccessibleSelectionCount; for (const { file, selectedVia } of snapshot.files) { await importFileIsolated(context, client, file, selectedVia, counters); } + if (!shouldRetireUnseenSources(snapshot)) { + counters.failed += snapshot.skippedEntryCount; + log.error( + "Skipped the unseen-source sweep after an incomplete Google Drive enumeration", + { + connectorId: context.connector.id, + skippedEntryCount: snapshot.skippedEntryCount, + discovered: snapshot.files.length, + }, + ); + return false; + } counters.missing += await markUnseenSourcesMissing( context, snapshot.files.map(({ file }) => file.id), ); + return true; } async function reconcileInitial( @@ -1279,7 +1352,11 @@ async function reconcileInitial( const startPageToken = await client.getStartPageToken( context.connector.sharedDriveId, ); - await reconcileSelectionSnapshot(context, client, counters); + const complete = await reconcileSelectionSnapshot(context, client, counters); + // A full rebuild discharges any obligation an earlier run left behind. + if (complete && isSelectionSnapshotPending(context.connector.metadata)) { + await setSelectionSnapshotPending(context, false); + } return startPageToken; } @@ -1329,10 +1406,6 @@ async function retryDeferredDownloads( } } -type GoogleDriveChange = Awaited< - ReturnType ->["values"][number]; - /** * Records the outcome of a failed file-scoped change. Connector lifecycle * errors are fatal to the whole run and are rethrown; a Drive 403/404 means the @@ -1447,48 +1520,87 @@ async function persistSyncCursor( } } +/** + * Write (or clear) the durable "a selection snapshot is owed" flag on the + * connector. + * + * The write is a jsonb merge rather than a read-modify-write of the loaded + * row: the webhook path also writes connector metadata, and replaying a stale + * in-memory copy would clobber it. + */ +async function setSelectionSnapshotPending( + context: ConnectorContext, + pending: boolean, +): Promise { + const persisted = await executeQuery( + (db) => + db + .update(repositoryConnectors) + .set({ + metadata: pending + ? sql`${repositoryConnectors.metadata} || ${JSON.stringify({ + [SELECTION_SNAPSHOT_PENDING_KEY]: new Date().toISOString(), + })}::jsonb` + : sql`${repositoryConnectors.metadata} - ${SELECTION_SNAPSHOT_PENDING_KEY}`, + updatedAt: new Date(), + }) + .where( + and( + eq(repositoryConnectors.id, context.connector.id), + eq( + repositoryConnectors.selectionRevision, + context.connector.selectionRevision, + ), + ne(repositoryConnectors.status, "revoked"), + ), + ) + .returning({ id: repositoryConnectors.id }), + pending + ? "googleContent.markSnapshotPending" + : "googleContent.clearSnapshotPending", + ); + if (persisted.length === 0) { + throw new ConnectorSelectionChangedError(); + } +} + async function reconcileChanges( context: ConnectorContext, client: GoogleDriveClient, initialCursor: string, counters: SyncCounters, ): Promise { - let cursor = initialCursor; - let requiresSelectionSnapshot = false; - for (;;) { - const page = await client.listChanges( - cursor, - context.connector.sharedDriveId, - ); - for (const change of page.values) { - counters.discovered += 1; - requiresSelectionSnapshot = - (await processGoogleDriveChange( - context, - client, - change, - counters, - )) || requiresSelectionSnapshot; - } - cursor = page.nextPageToken ?? page.newStartPageToken ?? cursor; - if (!requiresSelectionSnapshot) { - await persistSyncCursor( - context, + return reconcileChangePages(initialCursor, { + listChanges: async (cursor) => { + const page = await client.listChanges( cursor, - "googleContent.persistCursor", + context.connector.sharedDriveId, ); - } - if (!page.nextPageToken) break; - } - if (requiresSelectionSnapshot) { - await reconcileSelectionSnapshot(context, client, counters); - await persistSyncCursor( - context, - cursor, - "googleContent.persistSnapshotCursor", - ); - } - return cursor; + if (page.skippedEntries.length > 0) { + // Dropped, never reinterpreted: a change entry that fails validation + // says nothing about whether its file still exists, so no source is + // marked missing here. The cost is a missed update until that file + // changes again — visible in this line. + log.error("Skipped unparseable Google Drive change entries", { + connectorId: context.connector.id, + skippedCount: page.skippedEntries.length, + entries: page.skippedEntries, + }); + counters.failed += page.skippedEntries.length; + } + return page; + }, + processChange: async (change) => { + counters.discovered += 1; + return processGoogleDriveChange(context, client, change, counters); + }, + persistCursor: (cursor) => + persistSyncCursor(context, cursor, "googleContent.persistCursor"), + markSnapshotPending: () => setSelectionSnapshotPending(context, true), + runSelectionSnapshot: () => + reconcileSelectionSnapshot(context, client, counters), + clearSnapshotPending: () => setSelectionSnapshotPending(context, false), + }); } async function markConnectorAccessLost(connectorId: string): Promise { @@ -1740,6 +1852,23 @@ async function resolveSyncCursor( if (!context.connector.cursor) { return reconcileInitial(context, client, counters); } + // An earlier run advanced its cursor past a page that obligated a selection + // snapshot but did not finish (or even reach) that snapshot. Discharge the + // obligation before consuming anything new — the cursor no longer carries + // that information. + if (isSelectionSnapshotPending(context.connector.metadata)) { + log.info("Resuming a pending Google Drive selection snapshot", { + connectorId: context.connector.id, + runId, + pendingSince: + context.connector.metadata.selectionSnapshotPendingAt ?? null, + }); + await resumePendingSelectionSnapshot({ + runSelectionSnapshot: () => + reconcileSelectionSnapshot(context, client, counters), + clearSnapshotPending: () => setSelectionSnapshotPending(context, false), + }); + } try { return await reconcileChanges( context, diff --git a/infra/lambdas/google-content-sync/reconcile.ts b/infra/lambdas/google-content-sync/reconcile.ts new file mode 100644 index 000000000..e02f8a024 --- /dev/null +++ b/infra/lambdas/google-content-sync/reconcile.ts @@ -0,0 +1,130 @@ +/** + * The changes-feed reconcile loop, expressed against injected collaborators + * so it can be exercised without a database, Google credentials, or the + * Lambda's module-level environment requirements. + * + * The invariant this module exists to protect: the cursor advances page by + * page, ALWAYS, and the obligation to rebuild the selection snapshot is + * recorded durably instead of being implied by a cursor that was left behind. + * + * Before this split, a page that demanded a snapshot suppressed every + * per-page cursor write until the snapshot finished. A 900s Lambda timeout or + * a snapshot budget failure therefore threw away all the change-page work and + * replayed it from the original cursor on the next attempt — indefinitely, if + * the snapshot kept failing. + */ + +/** The connector-metadata slice that carries the snapshot obligation. */ +export interface SnapshotObligationMetadata { + selectionSnapshotPendingAt?: string; +} + +export function isSelectionSnapshotPending( + metadata: SnapshotObligationMetadata | null | undefined, +): boolean { + return typeof metadata?.selectionSnapshotPendingAt === "string"; +} + +/** The jsonb key the durable flag is stored under. */ +export const SELECTION_SNAPSHOT_PENDING_KEY = "selectionSnapshotPendingAt"; + +/** + * Retiring sources that a snapshot did not see is only sound after a COMPLETE + * enumeration. + * + * Per-entry validation drops entries Google returned in a shape this client + * cannot read. A dropped entry says nothing about whether its file still + * exists, so a file could be absent from the seen set purely because one + * record was unreadable — and marking it missing would be silent data loss. + * When anything was dropped, the sweep is skipped and left to the next clean + * snapshot. + */ +export function shouldRetireUnseenSources(enumeration: { + skippedEntryCount: number; +}): boolean { + return enumeration.skippedEntryCount === 0; +} + +export interface ChangesPage { + values: TChange[]; + nextPageToken: string | null; + newStartPageToken: string | null; +} + +export interface ReconcileChangesDeps { + listChanges(cursor: string): Promise>; + /** + * Handles one change entry. Returns true when the entry obligates a full + * selection snapshot (a Shared Drive went away). + */ + processChange(change: TChange): Promise; + /** Advance the durable cursor. */ + persistCursor(cursor: string): Promise; + /** Record the snapshot obligation durably. Called at most once per run. */ + markSnapshotPending(): Promise; + /** + * Re-enumerate every selection and retire whatever is unreachable. Resolves + * false when the enumeration was incomplete, in which case the unseen-source + * sweep did not run and the obligation is NOT discharged. + */ + runSelectionSnapshot(): Promise; + /** Clear the durable obligation. Only ever called after a complete snapshot. */ + clearSnapshotPending(): Promise; +} + +export type ResumeSnapshotDeps = Pick< + ReconcileChangesDeps, + "runSelectionSnapshot" | "clearSnapshotPending" +>; + +/** + * Discharge a snapshot obligation left behind by an earlier run, before any + * new change page is consumed. The flag is cleared only once the snapshot has + * completed, so a repeated failure repeats the snapshot rather than silently + * dropping the obligation. + */ +export async function resumePendingSelectionSnapshot( + deps: ResumeSnapshotDeps, +): Promise { + const complete = await deps.runSelectionSnapshot(); + if (complete) await deps.clearSnapshotPending(); + return complete; +} + +/** + * Consume the changes feed from `initialCursor` and return the cursor to + * persist as the run's result. + * + * Ordering rules, in the order they matter: + * 1. A snapshot obligation is written BEFORE the cursor moves past the page + * that raised it. A crash between the two can only replay work, never + * lose the obligation. + * 2. The cursor is persisted after every page, unconditionally. + * 3. The snapshot runs after the feed is drained, and the flag is cleared + * only when it succeeded. + */ +export async function reconcileChangePages( + initialCursor: string, + deps: ReconcileChangesDeps, +): Promise { + let cursor = initialCursor; + let requiresSelectionSnapshot = false; + for (;;) { + const page = await deps.listChanges(cursor); + for (const change of page.values) { + const demandsSnapshot = await deps.processChange(change); + if (demandsSnapshot && !requiresSelectionSnapshot) { + requiresSelectionSnapshot = true; + // Rule 1: durable before the cursor moves past this page. + await deps.markSnapshotPending(); + } + } + cursor = page.nextPageToken ?? page.newStartPageToken ?? cursor; + await deps.persistCursor(cursor); + if (!page.nextPageToken) break; + } + if (requiresSelectionSnapshot) { + await resumePendingSelectionSnapshot(deps); + } + return cursor; +} diff --git a/infra/lambdas/google-content-sync/tsconfig.json b/infra/lambdas/google-content-sync/tsconfig.json new file mode 100644 index 000000000..6de1924a6 --- /dev/null +++ b/infra/lambdas/google-content-sync/tsconfig.json @@ -0,0 +1,9 @@ +{ + "extends": "../../../tsconfig.json", + "compilerOptions": { + "incremental": false, + "types": ["node", "aws-lambda"] + }, + "include": ["./*.ts", "../../../next-env.d.ts"], + "exclude": ["./__tests__"] +} diff --git a/infra/package.json b/infra/package.json index 8f92105d7..6fcd118b7 100644 --- a/infra/package.json +++ b/infra/package.json @@ -5,13 +5,14 @@ "infra": "bin/infra.js" }, "scripts": { - "build": "tsc && bun run typecheck:unified-content && bun run test:lambdas", + "build": "tsc && bun run typecheck:unified-content && bun run typecheck:google-content-sync && bun run test:lambdas", "build:lambdas": "./build-lambdas.sh", "build:all": "bun run build && bun run build:lambdas", "watch": "tsc -w", "test": "jest", "test:lambdas": "bun test lambdas/embedding-generator/__tests__ && cd lambdas/agent-skill-builder && bun install && bun run test", "typecheck:unified-content": "tsc --project lambdas/unified-content-processor/tsconfig.json", + "typecheck:google-content-sync": "tsc --project lambdas/google-content-sync/tsconfig.json", "cdk": "cdk" }, "devDependencies": { diff --git a/lib/db/schema/tables/repository-connectors.ts b/lib/db/schema/tables/repository-connectors.ts index e197b1cad..50ff5d388 100644 --- a/lib/db/schema/tables/repository-connectors.ts +++ b/lib/db/schema/tables/repository-connectors.ts @@ -42,6 +42,16 @@ export interface RepositoryConnectorMetadata { oauthEmail?: string; googleDriveName?: string; lastNotificationState?: string; + /** + * ISO timestamp set when a change page has obligated the connector to + * rebuild its selection snapshot, and cleared once that snapshot completes. + * + * The obligation must outlive the process: without it, the only record that + * a snapshot was owed was the un-advanced cursor, so a Lambda timeout or a + * snapshot budget failure forced every already-processed change page to be + * re-fetched and re-processed on the next attempt. + */ + selectionSnapshotPendingAt?: string; } export interface RepositoryConnectorSourceMetadata { diff --git a/lib/repositories/google-drive/drive-client.ts b/lib/repositories/google-drive/drive-client.ts index 473a1a781..75ce10f0c 100644 --- a/lib/repositories/google-drive/drive-client.ts +++ b/lib/repositories/google-drive/drive-client.ts @@ -24,36 +24,97 @@ const FILE_FIELDS = [ "shortcutDetails(targetId,targetMimeType,targetResourceKey)", ].join(","); -const driveUserSchema = z.object({ displayName: z.string().optional() }); +/** + * Google omits absent fields far more often than it nulls them, but an + * explicit `null` must never fail a whole page. `nullish` + a normalizing + * transform keeps the parsed shape identical to the previous `optional()` + * output while tolerating both encodings. + */ +const optionalString = z + .string() + .nullish() + .transform((value) => value ?? undefined) + .optional(); + +/** + * Link fields are stored as source metadata and are candidates for rendering + * as an href. They are deliberately NOT validated with a strict URL parser — + * a link Google formats unexpectedly must not stall the connector — but a + * non-http(s) scheme (`javascript:`, `data:`) is dropped rather than + * persisted, so loosening the validator cannot open an injection path. + */ +const optionalWebUrl = z + .string() + .nullish() + .transform((value) => + value && /^https?:\/\//i.test(value) ? value : undefined, + ) + .optional(); + +const optionalBoolean = (fallback: boolean) => + z + .boolean() + .nullish() + .transform((value) => value ?? fallback); + +const driveUserSchema = z.object({ displayName: optionalString }); + +/** + * Only `targetId` is consumed (`resolveShortcut` dereferences it and raises a + * graceful "no target" error when it is absent). `targetMimeType` has no + * reader anywhere in the codebase, so requiring it could only ever turn a + * complete page into a poison message — it is optional and unused. + */ const shortcutDetailsSchema = z.object({ - targetId: z.string(), - targetMimeType: z.string(), - targetResourceKey: z.string().optional(), + targetId: optionalString, + targetMimeType: optionalString, + targetResourceKey: optionalString, }); export const googleDriveFileSchema = z.object({ id: z.string(), name: z.string(), mimeType: z.string(), - parents: z.array(z.string()).optional().default([]), - driveId: z.string().optional(), - modifiedTime: z.string().datetime({ offset: true }).optional(), - md5Checksum: z.string().optional(), - version: z.string().optional(), - headRevisionId: z.string().optional(), - size: z.string().regex(/^\d+$/).optional(), - trashed: z.boolean().optional().default(false), - webViewLink: z.string().url().optional(), - iconLink: z.string().url().optional(), - owners: z.array(driveUserSchema).optional().default([]), - shortcutDetails: shortcutDetailsSchema.optional(), + parents: z + .array(z.string()) + .nullish() + .transform((value) => value ?? []), + driveId: optionalString, + // Plain string, not `.datetime()`: the only consumers are `new Date(...)` + // and a revision string. Callers guard invalid dates themselves. + modifiedTime: optionalString, + md5Checksum: optionalString, + version: optionalString, + headRevisionId: optionalString, + // The numeric shape is re-checked by `assertGoogleSourceMetadataSize`, and + // the streaming byte bound is authoritative regardless of what Drive + // reports here. + size: optionalString, + trashed: optionalBoolean(false), + webViewLink: optionalWebUrl, + iconLink: optionalWebUrl, + owners: z + .array(driveUserSchema) + .nullish() + .transform((value) => value ?? []), + shortcutDetails: shortcutDetailsSchema.nullish(), }); export type GoogleDriveFile = z.infer; +/** + * List envelopes are parsed with UNVALIDATED entries so a single malformed + * member cannot reject the page. Each entry is validated on its own by + * {@link parseDriveEntries}. + */ +const rawEntriesSchema = z + .array(z.unknown()) + .nullish() + .transform((value) => value ?? []); + const filesListSchema = z.object({ - nextPageToken: z.string().optional(), - files: z.array(googleDriveFileSchema).default([]), + nextPageToken: optionalString, + files: rawEntriesSchema, }); /** @@ -67,18 +128,18 @@ const filesListSchema = z.object({ * value Google adds later must not re-poison the queue. */ const driveChangeSchema = z.object({ - changeType: z.string().optional(), - fileId: z.string().optional(), - removed: z.boolean().optional().default(false), - time: z.string().datetime({ offset: true }).optional(), - driveId: z.string().optional(), - file: googleDriveFileSchema.optional(), + changeType: optionalString, + fileId: optionalString, + removed: optionalBoolean(false), + time: optionalString, + driveId: optionalString, + file: googleDriveFileSchema.nullish(), }); const changesListSchema = z.object({ - nextPageToken: z.string().optional(), - newStartPageToken: z.string().optional(), - changes: z.array(driveChangeSchema).default([]), + nextPageToken: optionalString, + newStartPageToken: optionalString, + changes: rawEntriesSchema, }); const startPageTokenSchema = z.object({ startPageToken: z.string() }); @@ -106,15 +167,76 @@ const downloadOperationSchema = z.object({ export type GoogleDriveChange = z.infer; +/** + * A single list entry Google returned that this client could not validate. + * + * The entry is DROPPED, never reinterpreted: a parse failure says nothing + * about whether the underlying Drive file still exists, so treating it as a + * removal would trade a stalled connector for silent data loss. The cost of + * dropping it is that the file's latest change is missed until it changes + * again (or a selection snapshot re-enumerates it) — which is why every + * skipped entry is reported to the caller for logging. + */ +export interface GoogleDriveSkippedEntry { + /** Position within the page, so repeat offenders are identifiable. */ + index: number; + /** `id` / `fileId` / `file.id` when one of them was extractable. */ + id: string | null; + issues: Array<{ path: string; message: string }>; +} + export interface GoogleDriveListPage { values: T[]; nextPageToken: string | null; + /** Entries dropped by per-entry validation. Empty on a healthy page. */ + skippedEntries: GoogleDriveSkippedEntry[]; } export interface GoogleDriveChangesPage extends GoogleDriveListPage { newStartPageToken: string | null; } +function extractEntryId(entry: unknown): string | null { + if (typeof entry !== "object" || entry === null) return null; + const record = entry as Record; + if (typeof record.id === "string") return record.id; + if (typeof record.fileId === "string") return record.fileId; + const file = record.file; + if (typeof file === "object" && file !== null) { + const nested = (file as Record).id; + if (typeof nested === "string") return nested; + } + return null; +} + +/** + * Validate list entries one at a time. One malformed member of a 1000-entry + * page costs that one entry, not the page — and therefore not the cursor. + */ +function parseDriveEntries( + schema: z.ZodType, + entries: readonly unknown[], +): { values: T[]; skippedEntries: GoogleDriveSkippedEntry[] } { + const values: T[] = []; + const skippedEntries: GoogleDriveSkippedEntry[] = []; + for (const [index, entry] of entries.entries()) { + const result = schema.safeParse(entry); + if (result.success) { + values.push(result.data); + continue; + } + skippedEntries.push({ + index, + id: extractEntryId(entry), + issues: result.error.issues.map((issue) => ({ + path: issue.path.join("."), + message: issue.message, + })), + }); + } + return { values, skippedEntries }; +} + export interface GoogleDriveWatch { channelId: string; resourceId: string; @@ -232,9 +354,11 @@ export class GoogleDriveClient { if (pageToken) url.searchParams.set("pageToken", pageToken); const response = await this.request(url); const page = filesListSchema.parse(await response.json()); + const parsed = parseDriveEntries(googleDriveFileSchema, page.files); return { - values: page.files, + values: parsed.values, nextPageToken: page.nextPageToken ?? null, + skippedEntries: parsed.skippedEntries, }; } @@ -253,9 +377,11 @@ export class GoogleDriveClient { if (pageToken) url.searchParams.set("pageToken", pageToken); const response = await this.request(url); const page = filesListSchema.parse(await response.json()); + const parsed = parseDriveEntries(googleDriveFileSchema, page.files); return { - values: page.files, + values: parsed.values, nextPageToken: page.nextPageToken ?? null, + skippedEntries: parsed.skippedEntries, }; } @@ -286,10 +412,12 @@ export class GoogleDriveClient { } const response = await this.request(url); const page = changesListSchema.parse(await response.json()); + const parsed = parseDriveEntries(driveChangeSchema, page.changes); return { - values: page.changes, + values: parsed.values, nextPageToken: page.nextPageToken ?? null, newStartPageToken: page.newStartPageToken ?? null, + skippedEntries: parsed.skippedEntries, }; } diff --git a/tests/unit/google-content-sync-change-scope.test.ts b/tests/unit/google-content-sync-change-scope.test.ts index d8bb58db9..df8b435ef 100644 --- a/tests/unit/google-content-sync-change-scope.test.ts +++ b/tests/unit/google-content-sync-change-scope.test.ts @@ -151,18 +151,17 @@ describe("processGoogleDriveChange guard wiring", () => { expect(body).toContain("return driveRemoved;"); }); - test("the reconcile loop persists the cursor on both skip and removal", () => { + test("the reconcile loop is the shared, behaviorally tested one", () => { + // The cursor/obligation ordering itself is proven in + // google-content-sync-reconcile.test.ts against the real loop. All this + // asserts is that index.ts still delegates to it rather than growing a + // second copy. const loop = source.slice( source.indexOf("async function reconcileChanges("), source.indexOf("async function markConnectorAccessLost("), ); - // Skipped entry: returns false, requiresSelectionSnapshot stays false, so - // the per-page persist runs. - expect(loop).toContain("if (!requiresSelectionSnapshot) {"); - // Removal: returns true, so the cursor is persisted after the snapshot. - expect(loop).toContain("if (requiresSelectionSnapshot) {"); - expect(loop).toContain("await reconcileSelectionSnapshot("); - // Either way the cursor advances — that is the invariant the outage broke. - expect(loop.match(/await persistSyncCursor\(/g)).toHaveLength(2); + expect(loop).toContain("return reconcileChangePages("); + expect(loop).toContain("runSelectionSnapshot:"); + expect(loop).toContain("markSnapshotPending:"); }); }); diff --git a/tests/unit/google-content-sync-reconcile.test.ts b/tests/unit/google-content-sync-reconcile.test.ts new file mode 100644 index 000000000..7a96ef10e --- /dev/null +++ b/tests/unit/google-content-sync-reconcile.test.ts @@ -0,0 +1,242 @@ +/** @jest-environment node */ + +import { + isSelectionSnapshotPending, + reconcileChangePages, + resumePendingSelectionSnapshot, + SELECTION_SNAPSHOT_PENDING_KEY, + shouldRetireUnseenSources, + type ChangesPage, + type ReconcileChangesDeps, +} from "../../infra/lambdas/google-content-sync/reconcile"; + +interface Recorder { + events: string[]; + cursors: string[]; + deps: ReconcileChangesDeps; +} + +function page( + values: string[], + nextPageToken: string | null, + newStartPageToken: string | null = null, +): ChangesPage { + return { values, nextPageToken, newStartPageToken }; +} + +function recorder(options: { + pages: Array>; + demandsSnapshot?: (change: string) => boolean; + snapshotComplete?: boolean; + snapshotThrows?: boolean; +}): Recorder { + const events: string[] = []; + const cursors: string[] = []; + let pageIndex = 0; + return { + events, + cursors, + deps: { + listChanges: async (cursor) => { + events.push(`list:${cursor}`); + const next = options.pages[pageIndex]; + pageIndex += 1; + if (!next) throw new Error("listChanges called more times than staged"); + return next; + }, + processChange: async (change) => { + events.push(`process:${change}`); + return options.demandsSnapshot?.(change) ?? false; + }, + persistCursor: async (cursor) => { + events.push(`persist:${cursor}`); + cursors.push(cursor); + }, + markSnapshotPending: async () => { + events.push("mark-pending"); + }, + runSelectionSnapshot: async () => { + events.push("snapshot"); + if (options.snapshotThrows) throw new Error("snapshot failed"); + return options.snapshotComplete ?? true; + }, + clearSnapshotPending: async () => { + events.push("clear-pending"); + }, + }, + }; +} + +describe("selection snapshot obligation flag", () => { + test("is detected only when the connector metadata carries a timestamp", () => { + expect(isSelectionSnapshotPending(undefined)).toBe(false); + expect(isSelectionSnapshotPending(null)).toBe(false); + expect(isSelectionSnapshotPending({})).toBe(false); + expect( + isSelectionSnapshotPending({ + [SELECTION_SNAPSHOT_PENDING_KEY]: "2026-08-07T00:00:00.000Z", + }), + ).toBe(true); + }); +}); + +describe("reconcileChangePages cursor durability", () => { + test("advances the cursor after every page", async () => { + const { deps, cursors } = recorder({ + pages: [page(["a"], "cursor-2"), page(["b"], null, "start-3")], + }); + + const cursor = await reconcileChangePages("cursor-1", deps); + + expect(cursors).toEqual(["cursor-2", "start-3"]); + expect(cursor).toBe("start-3"); + }); + + test("keeps advancing the cursor while a snapshot obligation is outstanding", async () => { + // This is the regression: the old loop suppressed every per-page cursor + // write once a page demanded a snapshot, so a Lambda timeout replayed all + // of the already-processed change pages on the next attempt. + const { deps, cursors, events } = recorder({ + pages: [page(["removal"], "cursor-2"), page(["b"], null, "start-3")], + demandsSnapshot: (change) => change === "removal", + }); + + await reconcileChangePages("cursor-1", deps); + + expect(cursors).toEqual(["cursor-2", "start-3"]); + // The obligation is durable BEFORE the cursor moves past the page that + // raised it — a crash in between can only replay work, never lose it. + expect(events.indexOf("mark-pending")).toBeLessThan( + events.indexOf("persist:cursor-2"), + ); + }); + + test("records the obligation once, no matter how many entries demand it", async () => { + const { deps, events } = recorder({ + pages: [page(["r1", "r2", "r3"], null, "start-2")], + demandsSnapshot: () => true, + }); + + await reconcileChangePages("cursor-1", deps); + + expect(events.filter((event) => event === "mark-pending")).toHaveLength(1); + }); + + test("runs the snapshot after the feed is drained and clears the flag", async () => { + const { deps, events } = recorder({ + pages: [page(["removal"], "cursor-2"), page([], null, "start-3")], + demandsSnapshot: () => true, + }); + + await reconcileChangePages("cursor-1", deps); + + expect(events).toEqual([ + "list:cursor-1", + "process:removal", + "mark-pending", + "persist:cursor-2", + "list:cursor-2", + "persist:start-3", + "snapshot", + "clear-pending", + ]); + }); + + test("leaves the flag set when the snapshot enumeration was incomplete", async () => { + // An incomplete enumeration never feeds the unseen-source sweep, so the + // obligation is not discharged and the next run retries it. + const { deps, events } = recorder({ + pages: [page(["removal"], null, "start-2")], + demandsSnapshot: () => true, + snapshotComplete: false, + }); + + await reconcileChangePages("cursor-1", deps); + + expect(events).toContain("snapshot"); + expect(events).not.toContain("clear-pending"); + }); + + test("leaves the flag set when the snapshot throws, after the cursor advanced", async () => { + const { deps, cursors, events } = recorder({ + pages: [page(["removal"], null, "start-2")], + demandsSnapshot: () => true, + snapshotThrows: true, + }); + + await expect(reconcileChangePages("cursor-1", deps)).rejects.toThrow( + "snapshot failed", + ); + + // The change-page work is banked; only the snapshot is retried. + expect(cursors).toEqual(["start-2"]); + expect(events).not.toContain("clear-pending"); + }); + + test("never marks the obligation when no entry demands a snapshot", async () => { + const { deps, events } = recorder({ + pages: [page(["a", "b"], null, "start-2")], + }); + + await reconcileChangePages("cursor-1", deps); + + expect(events).not.toContain("mark-pending"); + expect(events).not.toContain("snapshot"); + }); + + test("holds the cursor when Google returns neither token", async () => { + const { deps, cursors } = recorder({ pages: [page(["a"], null, null)] }); + + const cursor = await reconcileChangePages("cursor-1", deps); + + expect(cursor).toBe("cursor-1"); + expect(cursors).toEqual(["cursor-1"]); + }); +}); + +describe("resumePendingSelectionSnapshot", () => { + test("discharges the obligation only after a complete snapshot", async () => { + const events: string[] = []; + const complete = await resumePendingSelectionSnapshot({ + runSelectionSnapshot: async () => { + events.push("snapshot"); + return true; + }, + clearSnapshotPending: async () => { + events.push("clear-pending"); + }, + }); + + expect(complete).toBe(true); + expect(events).toEqual(["snapshot", "clear-pending"]); + }); + + test("keeps the obligation when the snapshot was incomplete", async () => { + const events: string[] = []; + const complete = await resumePendingSelectionSnapshot({ + runSelectionSnapshot: async () => { + events.push("snapshot"); + return false; + }, + clearSnapshotPending: async () => { + events.push("clear-pending"); + }, + }); + + expect(complete).toBe(false); + expect(events).toEqual(["snapshot"]); + }); +}); + +describe("shouldRetireUnseenSources", () => { + test("allows the sweep only after a complete enumeration", () => { + expect(shouldRetireUnseenSources({ skippedEntryCount: 0 })).toBe(true); + }); + + test("blocks the sweep when any entry was dropped by validation", () => { + // Marking a still-existing file missing because one record was + // unreadable would be silent data loss. + expect(shouldRetireUnseenSources({ skippedEntryCount: 1 })).toBe(false); + expect(shouldRetireUnseenSources({ skippedEntryCount: 250 })).toBe(false); + }); +}); diff --git a/tests/unit/lib/repositories/google-drive-changes-feed.test.ts b/tests/unit/lib/repositories/google-drive-changes-feed.test.ts index 4c74a5fe3..908dad2ee 100644 --- a/tests/unit/lib/repositories/google-drive-changes-feed.test.ts +++ b/tests/unit/lib/repositories/google-drive-changes-feed.test.ts @@ -11,6 +11,15 @@ function jsonResponse(body: unknown, status = 200): Response { } as unknown as Response; } +function fileEntry(id: string, overrides: Record = {}) { + return { + id, + name: `File ${id}`, + mimeType: "application/vnd.google-apps.document", + ...overrides, + }; +} + describe("GoogleDriveClient changes feed", () => { test("accepts Shared Drive-scoped change entries that carry no fileId", async () => { // Regression: Google emits changeType "drive" entries (Shared Drive @@ -71,6 +80,87 @@ describe("GoogleDriveClient changes feed", () => { ]); }); + test("keeps every well-formed entry when one entry is malformed", async () => { + // Regression class: one poisoned entry among a thousand used to reject the + // whole page, so the cursor never advanced and the connector stalled — + // exactly the original fileId incident with a different field. + const fetchMock = jest + .fn, Parameters>() + .mockResolvedValue( + jsonResponse({ + nextPageToken: "page-2", + changes: [ + { changeType: "file", fileId: "file-1", file: fileEntry("file-1") }, + // `file.name` is a number: not representable, so this entry alone + // is dropped. + { + changeType: "file", + fileId: "file-2", + file: { ...fileEntry("file-2"), name: 42 }, + }, + { changeType: "file", fileId: "file-3", file: fileEntry("file-3") }, + ], + }), + ); + + const page = await new GoogleDriveClient("token", { + fetch: fetchMock, + }).listChanges("cursor-1"); + + expect(page.values.map((change) => change.fileId)).toEqual([ + "file-1", + "file-3", + ]); + // The page still yields its continuation token, so the cursor advances. + expect(page.nextPageToken).toBe("page-2"); + expect(page.skippedEntries).toHaveLength(1); + expect(page.skippedEntries[0]).toEqual( + expect.objectContaining({ index: 1, id: "file-2" }), + ); + expect(page.skippedEntries[0]?.issues[0]?.path).toBe("file.name"); + expect(page.skippedEntries[0]?.issues[0]?.message).toEqual( + expect.any(String), + ); + }); + + test("extracts an identifier from a malformed entry when one is present", async () => { + const fetchMock = jest + .fn, Parameters>() + .mockResolvedValue( + jsonResponse({ + newStartPageToken: "start-9", + changes: [{ fileId: 12, file: { id: "nested-1" } }, "not-an-object"], + }), + ); + + const page = await new GoogleDriveClient("token", { + fetch: fetchMock, + }).listChanges("cursor-1"); + + expect(page.values).toHaveLength(0); + expect(page.skippedEntries.map((entry) => entry.id)).toEqual([ + "nested-1", + null, + ]); + }); + + test("reports no skipped entries for a healthy page", async () => { + const fetchMock = jest + .fn, Parameters>() + .mockResolvedValue( + jsonResponse({ + newStartPageToken: "start-4", + changes: [{ changeType: "file", fileId: "file-1" }], + }), + ); + + const page = await new GoogleDriveClient("token", { + fetch: fetchMock, + }).listChanges("cursor-1"); + + expect(page.skippedEntries).toEqual([]); + }); + test("requests changeType in the changes fields projection", async () => { const fetchMock = jest .fn, Parameters>() @@ -87,3 +177,136 @@ describe("GoogleDriveClient changes feed", () => { }); }); + +describe("GoogleDriveClient file entry tolerance", () => { + async function listOneFile(overrides: Record) { + const fetchMock = jest + .fn, Parameters>() + .mockResolvedValue( + jsonResponse({ files: [fileEntry("file-1", overrides)] }), + ); + const page = await new GoogleDriveClient("token", { + fetch: fetchMock, + }).listChildren("folder-1"); + return page; + } + + test("accepts timestamps Google formats without a UTC offset", async () => { + // The only consumers are `new Date(...)` and a revision string; a strict + // RFC-3339-with-offset validator would have rejected the whole page. + const page = await listOneFile({ modifiedTime: "2026-08-06T12:00:00Z" }); + expect(page.skippedEntries).toEqual([]); + expect(page.values[0]?.modifiedTime).toBe("2026-08-06T12:00:00Z"); + }); + + test("accepts a non-numeric size instead of rejecting the entry", async () => { + // The streaming byte bound is authoritative, so a surprising size string + // must not cost us the file. + const page = await listOneFile({ size: "unknown" }); + expect(page.skippedEntries).toEqual([]); + expect(page.values[0]?.size).toBe("unknown"); + }); + + test("keeps http(s) links and drops non-navigable schemes", async () => { + const kept = await listOneFile({ + webViewLink: "https://drive.example.test/d/file-1", + iconLink: "http://drive.example.test/icon.png", + }); + expect(kept.values[0]?.webViewLink).toBe( + "https://drive.example.test/d/file-1", + ); + expect(kept.values[0]?.iconLink).toBe("http://drive.example.test/icon.png"); + + // Loosening the URL validator must not make the field an injection + // vector: a non-http(s) scheme is dropped, not persisted. + const dropped = await listOneFile({ + webViewLink: "javascript:alert(1)", + iconLink: "data:text/html,