fix: coalesce concurrent pushCommits to stop genesis double-POST

Two concurrent calls to `Resource.pushCommits()` both entered the
`while (this._pendingCommits.length > 0)` drain loop and both POSTed
the queued commit before either had a chance to `shift()` it. Symptom
in dagger CI:

    Commit for did:ad:… has is_genesis: true, but the resource
    already exists.

emitted from `lib/src/commit.rs:422`. Locally the server tolerates the
duplicate (idempotent on same signature) so the bug is invisible; in
the dagger single-core actix container the lookup→genesis-check window
opens up enough that the second POST loses the TOCTOU and 500s.

Trigger: `WSClient.handleOpen` runs `syncDirtyResources()` on every
WS open. Under flaky network/auth conditions the WS reconnects twice
in close succession (manual `store.reconnect()` + auto-retry on a
slow auth response), each open fires a full `syncDirtyResources()`,
and the second one runs in parallel because there's no re-entrance
guard. Inside, when the resource has no unsaved Loro changes the
sync calls `pushCommits()` directly — bypassing `save()`'s
`hasQueue` / `inProgressCommit` guard.

`pushCommits()` now coalesces concurrent calls onto a single
in-flight drain via `inProgressPush`. The second caller observes the
first call's POST result instead of re-POSTing.

`tests/genesis-double-push.integration.test.ts` reproduces the race
against a real spawned `atomic-server` and asserts exactly one
`/commit` POST hits the network for two `pushCommits()` calls. Pre-fix
the test sees 2 POSTs; post-fix it sees 1. Existing offline integration
tests + unit suite still green; affected e2e specs (sync, file-upload-
offline, offline-chatroom) pass cleanly locally.

browser/lib/src/resource.ts

@@ -124,6 +124,15 @@ export class Resource<C extends OptionalClass = any> {
 
   private inProgressCommit: Promise<void> | undefined;
   private hasQueue = false;
+  /**
+   * Coalesces concurrent {@link pushCommits} invocations onto a single
+   * in-flight drain. Two callers must NOT both POST the same queued
+   * commit — under load the second POST hits the server's
+   * lookup→genesis-check TOCTOU and 500s with
+   * "is_genesis: true, but the resource already exists". Reproduced in
+   * `tests/genesis-double-push.integration.test.ts`.
+   */
+  private inProgressPush: Promise<string | undefined> | undefined;
 
   private _store?: Store;
   private eventManager = new EventManager<
@@ -1492,12 +1501,37 @@ export class Resource<C extends OptionalClass = any> {
    *
    * After a successful push the resource's `lastCommit` is updated from the
    * server response and the local queue is cleared.
+   *
+   * Concurrent invocations (e.g. two `syncDirtyResources` calls fired by a
+   * fast WS reconnect flap) are coalesced onto the same in-flight drain via
+   * {@link inProgressPush} — the second caller observes the first's POST
+   * result instead of double-POSTing the queued commits. Without that
+   * guard the dagger CI runner reproducibly hits
+   * "Commit for did:ad:… has is_genesis: true, but the resource already
+   * exists" when both pushes race the server's lookup→apply window.
    */
   public async pushCommits(): Promise<string | undefined> {
     if (this._pendingCommits.length === 0) {
       return undefined;
     }
 
+    if (this.inProgressPush) {
+      return this.inProgressPush;
+    }
+
+    const drain = this._drainPendingCommits();
+    this.inProgressPush = drain;
+    try {
+      return await drain;
+    } finally {
+      // Clear AFTER await so a concurrent caller arriving mid-drain still
+      // joins this same promise; only post-resolution does the next call
+      // start a fresh drain.
+      this.inProgressPush = undefined;
+    }
+  }
+
+  private async _drainPendingCommits(): Promise<string | undefined> {
     const endpoint = this.getCommitEndpoint();
     const wasNew =
       this._pendingCommits.length > 0 &&

browser/lib/tests/genesis-double-push.integration.test.ts

@@ -0,0 +1,155 @@
+/**
+ * Reproduces the dagger-CI symptom:
+ *
+ *     Commit for did:ad:... has is_genesis: true, but the resource
+ *     already exists.
+ *
+ * Root cause hypothesis: when the WebSocket reconnects twice in quick
+ * succession (which dagger's slow single-core actix container makes
+ * common), `WSClient.handleOpen` fires `Store.syncDirtyResources` twice
+ * in parallel. Each call iterates the dirty subjects and reaches:
+ *
+ *     if (resource.hasUnsavedChanges()) await resource.save();
+ *     else                              await resource.pushCommits();
+ *
+ * `save()` has a `hasQueue` guard (via `inProgressCommit`), so two
+ * parallel `save()`s on the same resource serialise. `pushCommits()`
+ * has no such guard — both invocations enter the
+ * `while (this._pendingCommits.length > 0)` loop, both read
+ * `pendingCommits[0]`, both `await postCommit()`. The first POST
+ * creates the resource server-side; the second POST fails with the
+ * "is_genesis: true, but resource already exists" error.
+ *
+ * The repro below skips the WebSocket layer entirely. It manually
+ * queues a genesis commit (mimicking the offline → reconnect path),
+ * then calls `pushCommits()` twice in parallel against the real
+ * server's `/commit` endpoint and checks how many POSTs the server
+ * actually rejected. Without the fix this fails with the genesis
+ * already-exists error coming back from the second push.
+ */
+
+import { describe, it, expect, beforeAll, afterAll } from 'vitest';
+import path from 'node:path';
+import { fileURLToPath } from 'node:url';
+import { setTimeout as delay } from 'node:timers/promises';
+
+import { startServer, type ServerHandle } from './server-fixture.js';
+import { Agent } from '../src/agent.js';
+import { Store } from '../src/store.js';
+import { NodeClientDb } from '../src/client-db.node.js';
+import type { ClientDbWorker } from '../src/client-db.js';
+
+const here = path.dirname(fileURLToPath(import.meta.url));
+const wasmPath = path.resolve(here, '../../../wasm/pkg/atomic_wasm_bg.wasm');
+
+describe('parallel pushCommits race on genesis commits', () => {
+  let server: ServerHandle;
+
+  beforeAll(async () => {
+    server = await startServer();
+  }, 60_000);
+
+  afterAll(async () => {
+    await server?.stop();
+  });
+
+  it('two parallel pushCommits() calls on a queued genesis commit must not double-POST', async () => {
+    const agent = await Agent.fromSecret(server.agentSecret);
+
+    const clientDb = new NodeClientDb({
+      wasmPath,
+      baseUrl: server.serverUrl,
+    });
+    await clientDb.init();
+
+    const store = new Store({ serverUrl: server.serverUrl, agent });
+    store.setClientDb(clientDb as unknown as ClientDbWorker);
+
+    // Let WS handshake settle (we won't use WS for the actual race,
+    // but the resource subscribe path may try to use it).
+    await delay(300);
+
+    const drive = server.initialDrive ?? `${server.serverUrl}/`;
+
+    // Count `/commit` POSTs at the HTTP layer. The race is "two
+    // pushCommits both fire postCommit", so we instrument fetch.
+    const realFetch = globalThis.fetch.bind(globalThis);
+    let commitPostCount = 0;
+    let alreadyExistsCount = 0;
+    store.injectFetch((async (
+      input: RequestInfo | URL,
+      init?: RequestInit,
+    ) => {
+      const url =
+        typeof input === 'string' ? input : (input as Request | URL).toString();
+      if (url.endsWith('/commit') && init?.method === 'POST') {
+        commitPostCount++;
+        // Add a small delay so the second parallel push has a chance
+        // to enter the while-loop before the first's POST resolves.
+        // Without this delay the race window is too small to repro
+        // reliably under Node's single-threaded scheduler.
+        await delay(80);
+        const res = await realFetch(input as RequestInfo, init);
+        if (!res.ok) {
+          const body = await res.clone().text();
+          if (body.includes('but the resource already exists')) {
+            alreadyExistsCount++;
+          }
+        }
+
+        return res;
+      }
+
+      return realFetch(input as RequestInfo, init);
+    }) as typeof fetch);
+
+    // Build a brand-new DID-eligible resource. `store.newResource` with
+    // `did: true` does the placeholder + properties + initial signChanges
+    // dance for us, and crucially does NOT push — the signed genesis
+    // commit is sitting in `_pendingCommits` waiting for a `pushCommits`
+    // / `save` call.
+    const resource = await store.newResource({
+      isA: 'https://atomicdata.dev/classes/Folder',
+      parent: drive,
+      did: true,
+      propVals: {
+        'https://atomicdata.dev/properties/name': 'parallel-push race repro',
+      },
+    });
+
+    // Confirm the resource really has a queued genesis commit before we
+    // race it — without this we could be testing "two pushes against an
+    // empty queue both no-op" which is uninteresting.
+    expect(resource.hasPendingCommits, 'resource must have a queued genesis commit').toBe(true);
+
+    // The signed genesis commit is now sitting in resource._pendingCommits.
+    // Mirror what `syncDirtyResources` does on a re-entrant call: invoke
+    // pushCommits() twice in parallel. Both should resolve to the same
+    // server-recorded commit and only one /commit POST should reach the
+    // server (the second call should observe the empty queue or wait on
+    // the first).
+    const [a, b] = await Promise.allSettled([
+      resource.pushCommits(),
+      resource.pushCommits(),
+    ]);
+
+    // Diagnostic dump on failure so the next reader knows what to look at.
+    const summary = {
+      commitPostCount,
+      alreadyExistsCount,
+      resultA: a.status === 'fulfilled' ? a.value : (a.reason as Error).message,
+      resultB: b.status === 'fulfilled' ? b.value : (b.reason as Error).message,
+    };
+
+    expect(
+      alreadyExistsCount,
+      `server rejected at least one duplicate genesis: ${JSON.stringify(summary)}`,
+    ).toBe(0);
+    expect(
+      commitPostCount,
+      `expected exactly one /commit POST, got ${commitPostCount}: ${JSON.stringify(summary)}`,
+    ).toBe(1);
+
+    store.disconnect();
+  }, 60_000);
+});