From b4a206fc41f1dd31968bd1f68e501aa174ce2bdb Mon Sep 17 00:00:00 2001 From: xingzhou Date: Sun, 19 Jul 2026 22:05:04 +0800 Subject: [PATCH] fix(gmail): prevent duplicate watch renewals during stalls (#108974) --- src/hooks/gmail-ops.test.ts | 42 +++++++++++++++++++++- src/hooks/gmail-ops.ts | 17 +++++++-- src/hooks/gmail-watcher.test.ts | 62 ++++++++++++++++++++++++++++++++- src/hooks/gmail-watcher.ts | 51 +++++++++++++++++++++------ 4 files changed, 156 insertions(+), 16 deletions(-) diff --git a/src/hooks/gmail-ops.test.ts b/src/hooks/gmail-ops.test.ts index 49db731efe74..c4b3663e7282 100644 --- a/src/hooks/gmail-ops.test.ts +++ b/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?.(); + } + }); }); diff --git a/src/hooks/gmail-ops.ts b/src/hooks/gmail-ops.ts index 1df47dda1487..0462bc7f9889 100644 --- a/src/hooks/gmail-ops.ts +++ b/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 | 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 = () => { diff --git a/src/hooks/gmail-watcher.test.ts b/src/hooks/gmail-watcher.test.ts index 70c4ceea3c2c..f92f40c4ba41 100644 --- a/src/hooks/gmail-watcher.test.ts +++ b/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 { diff --git a/src/hooks/gmail-watcher.ts b/src/hooks/gmail-watcher.ts index 7b5b7fbd45f9..37f127349392 100644 --- a/src/hooks/gmail-watcher.ts +++ b/src/hooks/gmail-watcher.ts @@ -26,6 +26,8 @@ const log = createSubsystemLogger("gmail-watcher"); let watcherProcess: ChildProcess | null = null; let renewInterval: ReturnType | null = null; +let renewalInFlight: Promise | null = null; +let renewalAbortController: AbortController | null = null; let shuttingDown = false; let currentConfig: GmailHookRuntimeConfig | null = null; let respawnTimeout: ReturnType | null = null; @@ -177,6 +179,29 @@ function settleProcess(proc: ChildProcess): Promise { }); } +async function stopPeriodicRenewal(): Promise { + 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 { clearTimeout(respawnTimeout); respawnTimeout = null; } - if (renewInterval) { - clearInterval(renewInterval); - renewInterval = null; - } + await stopPeriodicRenewal(); if (watcherProcess) { log.info("stopping gmail watcher");