fix: stop Shared Drive-scoped change entries from poisoning the Google content-sync queue

infra/lambdas/google-content-sync/changes.ts

@@ -0,0 +1,57 @@
+/**
+ * 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;
+  removed?: boolean | 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<T extends DriveChangeIdentity>(
+  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;
+}
+
+/**
+ * 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;
+}

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 { isDriveRemovalChange, isFileScopedDriveChange } from "./changes";
 import {
   assertGoogleSourceMetadataSize,
   assertGoogleSourceResponseSize,
@@ -1332,12 +1333,62 @@ type GoogleDriveChange = Awaited<
   ReturnType<GoogleDriveClient["listChanges"]>
 >["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<void> {
+  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<boolean> {
+  // Shared Drive-scoped entries (rename, membership, restriction changes) have
+  // no fileId, so nothing on the file path below can act on them.
+  if (!isFileScopedDriveChange(change)) {
+    // 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)) {
       counters.missing += 1;
@@ -1363,21 +1414,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;
   }
 }

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);

tests/unit/google-content-sync-change-scope.test.ts

@@ -0,0 +1,168 @@
+/** @jest-environment node */
+
+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";
+
+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("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(
+      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("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("),
+    );
+    // 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);
+  });
+});