Drain pending saves without overlapping writes

Author: ndycode

index.ts

@@ -1681,6 +1681,7 @@ export const OpenAIOAuthPlugin: Plugin = async ({ client }: PluginInput) => {
 			accountManagerPromise = null;
 		};
 
+		// Plugin init can run more than once per process; runCleanup drains this module-level list.
 		registerCleanup(async () => {
 			await cachedAccountManager?.flushPendingSave();
 		});

lib/accounts.ts

@@ -1009,16 +1009,38 @@ export class AccountManager {
 	}
 
 	async flushPendingSave(): Promise<void> {
-		const hadDebouncedSave = !!this.saveDebounceTimer;
-		if (this.saveDebounceTimer) {
-			clearTimeout(this.saveDebounceTimer);
-			this.saveDebounceTimer = null;
-		}
-		if (this.pendingSave) {
-			await this.pendingSave;
-		}
-		if (hadDebouncedSave) {
-			await this.saveToDisk();
+		while (true) {
+			const hadDebouncedSave = !!this.saveDebounceTimer;
+			if (this.saveDebounceTimer) {
+				clearTimeout(this.saveDebounceTimer);
+				this.saveDebounceTimer = null;
+			}
+			if (this.pendingSave) {
+				await this.pendingSave;
+				// Let debounced callbacks waiting on the completed save re-arm pendingSave.
+				await Promise.resolve();
+				if (this.saveDebounceTimer !== null || this.pendingSave !== null) {
+					continue;
+				}
+			}
+			if (!hadDebouncedSave) {
+				return;
+			}
+			if (this.saveDebounceTimer !== null || this.pendingSave !== null) {
+				continue;
+			}
+			const flushSave = this.saveToDisk().finally(() => {
+				if (this.pendingSave === flushSave) {
+					this.pendingSave = null;
+				}
+			});
+			this.pendingSave = flushSave;
+			await flushSave;
+			// Drain saves that were queued while the flush save was in flight.
+			await Promise.resolve();
+			if (this.saveDebounceTimer === null && this.pendingSave === null) {
+				return;
+			}
 		}
 	}
 }

test/accounts.test.ts

@@ -1854,6 +1854,12 @@ describe("AccountManager", () => {
   });
 
   describe("flushPendingSave", () => {
+    const drainMicrotasks = async (turns = 10): Promise<void> => {
+      for (let turn = 0; turn < turns; turn++) {
+        await Promise.resolve();
+      }
+    };
+
     it("flushes pending debounced save", async () => {
       const { saveAccounts } = await import("../lib/storage.js");
       const mockSaveAccounts = vi.mocked(saveAccounts);
@@ -1998,6 +2004,183 @@ describe("AccountManager", () => {
         vi.useRealTimers();
       }
     });
+
+    it("drains saves queued during flush without overlapping writes", async () => {
+      vi.useFakeTimers();
+      try {
+        const { saveAccounts } = await import("../lib/storage.js");
+        const mockSaveAccounts = vi.mocked(saveAccounts);
+        mockSaveAccounts.mockClear();
+
+        let activeWrites = 0;
+        let maxConcurrentWrites = 0;
+        const settleWrites: Array<() => void> = [];
+        const writePromises: Promise<void>[] = [];
+
+        mockSaveAccounts.mockImplementation(() => {
+          activeWrites++;
+          maxConcurrentWrites = Math.max(maxConcurrentWrites, activeWrites);
+          const writePromise = new Promise<void>((resolve) => {
+            settleWrites.push(() => {
+              activeWrites--;
+              resolve();
+            });
+          });
+          writePromises.push(writePromise);
+          return writePromise;
+        });
+
+        const now = Date.now();
+        const stored = {
+          version: 3 as const,
+          activeIndex: 0,
+          accounts: [
+            { refreshToken: "token-1", addedAt: now, lastUsed: now },
+          ],
+        };
+
+        const manager = new AccountManager(undefined, stored);
+
+        manager.saveToDiskDebounced(50);
+        await vi.advanceTimersByTimeAsync(60);
+        expect(mockSaveAccounts).toHaveBeenCalledTimes(1);
+
+        const armGapSave = writePromises[0]!.then(() => {
+          manager.saveToDiskDebounced(0);
+        });
+
+        manager.saveToDiskDebounced(50);
+        const flushPromise = manager.flushPendingSave();
+
+        settleWrites[0]!();
+        await armGapSave;
+        await drainMicrotasks();
+
+        expect(mockSaveAccounts).toHaveBeenCalledTimes(2);
+
+        manager.saveToDiskDebounced(0);
+        await vi.advanceTimersByTimeAsync(0);
+        expect(mockSaveAccounts).toHaveBeenCalledTimes(2);
+
+        settleWrites[1]!();
+        await drainMicrotasks();
+
+        expect(mockSaveAccounts).toHaveBeenCalledTimes(3);
+
+        settleWrites[2]!();
+        await flushPromise;
+
+        expect(maxConcurrentWrites).toBe(1);
+      } finally {
+        vi.useRealTimers();
+      }
+    });
+
+    it("persists disabled auth-failure state after flushing over an in-flight save", async () => {
+      vi.useFakeTimers();
+      try {
+        const { saveAccounts } = await import("../lib/storage.js");
+        const mockSaveAccounts = vi.mocked(saveAccounts);
+        mockSaveAccounts.mockClear();
+
+        let activeWrites = 0;
+        let maxConcurrentWrites = 0;
+        const settleWrites: Array<() => void> = [];
+        const savedSnapshots: Array<
+          Array<{
+            refreshToken?: string;
+            enabled?: false;
+            disabledReason?: string;
+            coolingDownUntil?: number;
+            cooldownReason?: string;
+          }>
+        > = [];
+
+        mockSaveAccounts.mockImplementation(async (storage) => {
+          activeWrites++;
+          maxConcurrentWrites = Math.max(maxConcurrentWrites, activeWrites);
+          savedSnapshots.push(
+            storage.accounts.map((account) => ({
+              refreshToken: account.refreshToken,
+              enabled: account.enabled,
+              disabledReason: account.disabledReason,
+              coolingDownUntil: account.coolingDownUntil,
+              cooldownReason: account.cooldownReason,
+            })),
+          );
+          await new Promise<void>((resolve) => {
+            settleWrites.push(() => {
+              activeWrites--;
+              resolve();
+            });
+          });
+        });
+
+        const now = Date.now();
+        const stored = {
+          version: 3 as const,
+          activeIndex: 0,
+          accounts: [
+            {
+              refreshToken: "shared-refresh",
+              addedAt: now,
+              lastUsed: now,
+              coolingDownUntil: now + 10_000,
+              cooldownReason: "auth-failure" as const,
+            },
+            {
+              refreshToken: "shared-refresh",
+              addedAt: now + 1,
+              lastUsed: now + 1,
+              coolingDownUntil: now + 20_000,
+              cooldownReason: "auth-failure" as const,
+            },
+          ],
+        };
+
+        const manager = new AccountManager(undefined, stored);
+
+        manager.saveToDiskDebounced(50);
+        await vi.advanceTimersByTimeAsync(60);
+        expect(mockSaveAccounts).toHaveBeenCalledTimes(1);
+
+        const account = manager.getAccountsSnapshot()[0]!;
+        manager.disableAccountsWithSameRefreshToken(account);
+        manager.saveToDiskDebounced(50);
+
+        const flushPromise = manager.flushPendingSave();
+
+        settleWrites[0]!();
+        await drainMicrotasks();
+
+        expect(mockSaveAccounts).toHaveBeenCalledTimes(2);
+
+        const flushedSharedAccounts = savedSnapshots[1]!.filter(
+          (savedAccount) => savedAccount.refreshToken === "shared-refresh",
+        );
+        expect(flushedSharedAccounts).toEqual([
+          expect.objectContaining({
+            enabled: false,
+            disabledReason: "auth-failure",
+            coolingDownUntil: undefined,
+            cooldownReason: undefined,
+          }),
+          expect.objectContaining({
+            enabled: false,
+            disabledReason: "auth-failure",
+            coolingDownUntil: undefined,
+            cooldownReason: undefined,
+          }),
+        ]);
+
+        settleWrites[1]!();
+        await flushPromise;
+
+        expect(maxConcurrentWrites).toBe(1);
+      } finally {
+        vi.useRealTimers();
+      }
+    });
   });
 
   describe("health and token tracking methods", () => {