fix(gmail): prevent duplicate watch renewals during stalls (#108974)

src/hooks/gmail-ops.test.ts

@@ -55,7 +55,7 @@ vi.mock("../infra/executable-path.js", () => ({
 
 const { runGmailService } = await import("./gmail-ops.js");
 
-function createGmailConfig(account = "me@example.com") {
+function createGmailConfig(account = "me@example.com", renewEveryMinutes?: number) {
   return {
     hooks: {
       enabled: true,
@@ -65,6 +65,7 @@ function createGmailConfig(account = "me@example.com") {
         topic: "projects/demo/topics/gmail",
         pushToken: "push-token",
         tailscale: { mode: "off" as const },
+        renewEveryMinutes,
       },
     },
   };
@@ -115,4 +116,43 @@ describe("runGmailService", () => {
       shutdown?.();
     }
   });
+
+  it("keeps a stalled foreground renewal single-flight", async () => {
+    vi.useFakeTimers();
+    let resolveRenewal!: (value: { code: number; stdout: string; stderr: string }) => void;
+    const renewal = new Promise<{ code: number; stdout: string; stderr: string }>((resolve) => {
+      resolveRenewal = resolve;
+    });
+    mocks.getRuntimeConfig.mockReturnValue(createGmailConfig("me@example.com", 1));
+    mocks.runCommandWithTimeout
+      .mockResolvedValueOnce({ code: 0, stdout: "", stderr: "" })
+      .mockImplementation(async () => await renewal);
+
+    const child = new EventEmitter();
+    const kill = vi.fn(() => {
+      child.emit("exit", null, "SIGTERM");
+      return true;
+    });
+    mocks.spawn.mockReturnValue(Object.assign(child, { kill, killed: false }));
+
+    const existingSigintListeners = new Set(process.rawListeners("SIGINT"));
+    let shutdown: (() => void) | undefined;
+    try {
+      await runGmailService({});
+      shutdown = process
+        .rawListeners("SIGINT")
+        .find((listener) => !existingSigintListeners.has(listener)) as (() => void) | undefined;
+
+      await vi.advanceTimersByTimeAsync(60_000);
+      expect(mocks.runCommandWithTimeout).toHaveBeenCalledTimes(2);
+      await vi.advanceTimersByTimeAsync(60_000);
+      const callsWhileStalled = mocks.runCommandWithTimeout.mock.calls.length;
+      resolveRenewal({ code: 0, stdout: "", stderr: "" });
+      await Promise.resolve();
+
+      expect(callsWhileStalled).toBe(2);
+    } finally {
+      shutdown?.();
+    }
+  });
 });

src/hooks/gmail-ops.ts

@@ -321,10 +321,21 @@ export async function runGmailService(opts: GmailRunOptions) {
   let child = spawnGogServe(runtimeConfig);
 
   const renewMs = runtimeConfig.renewEveryMinutes * 60_000;
+  let renewalInFlight: Promise<void> | null = null;
   const renewTimer = setInterval(() => {
-    void startGmailWatch(runtimeConfig).catch((err: unknown) => {
-      defaultRuntime.error(`gmail watch renew failed: ${String(err)}`);
-    });
+    if (shuttingDown || renewalInFlight) {
+      return;
+    }
+    const renewal = startGmailWatch(runtimeConfig)
+      .catch((err: unknown) => {
+        defaultRuntime.error(`gmail watch renew failed: ${String(err)}`);
+      })
+      .finally(() => {
+        if (renewalInFlight === renewal) {
+          renewalInFlight = null;
+        }
+      });
+    renewalInFlight = renewal;
   }, renewMs);
 
   const detachSignals = () => {

src/hooks/gmail-watcher.test.ts

@@ -32,7 +32,7 @@ vi.mock("../process/exec.js", () => ({
 
 const { startGmailWatcher, stopGmailWatcher } = await import("./gmail-watcher.js");
 
-function createGmailConfig(account = "me@example.com") {
+function createGmailConfig(account = "me@example.com", renewEveryMinutes?: number) {
   return {
     hooks: {
       enabled: true,
@@ -41,6 +41,7 @@ function createGmailConfig(account = "me@example.com") {
         account,
         topic: "projects/demo/topics/gmail",
         pushToken: "push-token",
+        renewEveryMinutes,
       },
     },
   } as never;
@@ -282,6 +283,65 @@ describe("startGmailWatcher", () => {
     }
   });
 
+  it("keeps a stalled periodic renewal single-flight", async () => {
+    vi.useFakeTimers();
+    try {
+      const renewal = deferredCommandResult();
+      mocks.runCommandWithTimeout
+        .mockResolvedValueOnce({ code: 0, stdout: "", stderr: "" })
+        .mockImplementation(async () => await renewal.promise);
+
+      await startGmailWatcher(createGmailConfig("me@example.com", 1));
+      await vi.advanceTimersByTimeAsync(60_000);
+      expect(mocks.runCommandWithTimeout).toHaveBeenCalledTimes(2);
+
+      await vi.advanceTimersByTimeAsync(60_000);
+      const callsWhileStalled = mocks.runCommandWithTimeout.mock.calls.length;
+      renewal.resolve({ code: 0, stdout: "", stderr: "" });
+      await Promise.resolve();
+
+      expect(callsWhileStalled).toBe(2);
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
+  it("does not let a stalled renewal survive stop and suppress a replacement watcher", async () => {
+    vi.useFakeTimers();
+    try {
+      let stalledSignal: AbortSignal | undefined;
+      mocks.runCommandWithTimeout
+        .mockResolvedValueOnce({ code: 0, stdout: "", stderr: "" })
+        .mockImplementationOnce(
+          async (_args, options: { signal?: AbortSignal }) =>
+            await new Promise<{ code: number; stdout: string; stderr: string }>((resolve) => {
+              stalledSignal = options.signal;
+              options.signal?.addEventListener(
+                "abort",
+                () => resolve({ code: 1, stdout: "", stderr: "aborted" }),
+                { once: true },
+              );
+            }),
+        )
+        .mockResolvedValue({ code: 0, stdout: "", stderr: "" });
+
+      await startGmailWatcher(createGmailConfig("old@example.com", 1));
+      await vi.advanceTimersByTimeAsync(60_000);
+      expect(stalledSignal?.aborted).toBe(false);
+
+      await stopGmailWatcher();
+      expect(stalledSignal?.aborted).toBe(true);
+
+      await startGmailWatcher(createGmailConfig("new@example.com", 1));
+      await vi.advanceTimersByTimeAsync(60_000);
+
+      expect(mocks.runCommandWithTimeout).toHaveBeenCalledTimes(4);
+      expect(mocks.runCommandWithTimeout.mock.calls[3]?.[0]).toContain("new@example.com");
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
   it("escalates to SIGKILL and resolves on final timeout when process ignores signals", async () => {
     vi.useFakeTimers();
     try {

src/hooks/gmail-watcher.ts

@@ -26,6 +26,8 @@ const log = createSubsystemLogger("gmail-watcher");
 
 let watcherProcess: ChildProcess | null = null;
 let renewInterval: ReturnType<typeof setInterval> | null = null;
+let renewalInFlight: Promise<boolean> | null = null;
+let renewalAbortController: AbortController | null = null;
 let shuttingDown = false;
 let currentConfig: GmailHookRuntimeConfig | null = null;
 let respawnTimeout: ReturnType<typeof setTimeout> | null = null;
@@ -177,6 +179,29 @@ function settleProcess(proc: ChildProcess): Promise<void> {
   });
 }
 
+async function stopPeriodicRenewal(): Promise<void> {
+  if (renewInterval) {
+    clearInterval(renewInterval);
+    renewInterval = null;
+  }
+
+  const renewal = renewalInFlight;
+  const controller = renewalAbortController;
+  if (!renewal) {
+    renewalAbortController = null;
+    return;
+  }
+
+  controller?.abort();
+  await renewal;
+  if (renewalInFlight === renewal) {
+    renewalInFlight = null;
+  }
+  if (renewalAbortController === controller) {
+    renewalAbortController = null;
+  }
+}
+
 type GmailWatcherStartResult = {
   started: boolean;
   reason?: string;
@@ -291,16 +316,13 @@ export async function startGmailWatcher(
   // does not orphan the old serve process or leave a dangling timer.
   // This must run before Tailscale/watch-start to prevent the old
   // process from exiting and queuing a respawn during async work.
-  if (watcherProcess || renewInterval || respawnTimeout) {
+  if (watcherProcess || renewInterval || renewalInFlight || respawnTimeout) {
     shuttingDown = true;
     if (respawnTimeout) {
       clearTimeout(respawnTimeout);
       respawnTimeout = null;
     }
-    if (renewInterval) {
-      clearInterval(renewInterval);
-      renewInterval = null;
-    }
+    await stopPeriodicRenewal();
     if (watcherProcess) {
       const oldProcess = watcherProcess;
       watcherProcess = null;
@@ -363,10 +385,20 @@ export async function startGmailWatcher(
   watcherProcess = spawnGogServe(runtimeConfig);
   const renewMs = runtimeConfig.renewEveryMinutes * 60_000;
   renewInterval = setInterval(() => {
-    if (shuttingDown) {
+    if (shuttingDown || renewalInFlight) {
       return;
     }
-    void startGmailWatch(runtimeConfig);
+    const controller = new AbortController();
+    renewalAbortController = controller;
+    const renewal = startGmailWatch(runtimeConfig, { signal: controller.signal }).finally(() => {
+      if (renewalInFlight === renewal) {
+        renewalInFlight = null;
+      }
+      if (renewalAbortController === controller) {
+        renewalAbortController = null;
+      }
+    });
+    renewalInFlight = renewal;
   }, renewMs);
 
   log.info(
@@ -386,10 +418,7 @@ export async function stopGmailWatcher(): Promise<void> {
     clearTimeout(respawnTimeout);
     respawnTimeout = null;
   }
-  if (renewInterval) {
-    clearInterval(renewInterval);
-    renewInterval = null;
-  }
+  await stopPeriodicRenewal();
 
   if (watcherProcess) {
     log.info("stopping gmail watcher");