fix(browser): prevent stuck leadership state on acquireLeadership errors

kevin-dp · claude · kevin-dp · commit 51630874b247 · 2026-03-05T13:25:33.000+01:00
Previously, `state.isLeader = true` was set before the setup code that
calls `getStreamPosition()`. If `getStreamPosition` threw (e.g. due to
a UNIQUE constraint violation from React StrictMode double-mounting),
`isLeader` remained permanently stuck at `true` because the `finally`
block that resets it was inside an inner try/finally that was never
entered.

Fix: Wrap the entire lock callback body in a single try/finally. Set
`state.isLeader = true` only after successful setup (stream position
restore and term increment). The finally block always runs and resets
`isLeader = false` + cleans up the heartbeat timer.

Also refactors the coordinator to support lazy adapter wiring via
`setAdapter()`, allowing `createBrowserWASQLitePersistence` to inject
the adapter after construction. This enables the demo to construct the
coordinator without requiring the adapter upfront.

Co-Authored-By: Claude Opus 4.6 &lt;noreply@anthropic.com&gt;
diff --git a/packages/db-browser-wa-sqlite-persisted-collection/src/browser-coordinator.ts b/packages/db-browser-wa-sqlite-persisted-collection/src/browser-coordinator.ts
@@ -113,7 +113,7 @@ type AdapterWithPullSince = PersistenceAdapter<
 
 export type BrowserCollectionCoordinatorOptions = {
   dbName: string
-  adapter: AdapterWithPullSince
+  adapter?: AdapterWithPullSince
 }
 
 // ---------------------------------------------------------------------------
@@ -123,7 +123,7 @@ export type BrowserCollectionCoordinatorOptions = {
 export class BrowserCollectionCoordinator implements PersistedCollectionCoordinator {
   private readonly nodeId = crypto.randomUUID()
   private readonly dbName: string
-  private readonly adapter: AdapterWithPullSince
+  private adapter: AdapterWithPullSince | null
   private readonly channel: BroadcastChannel
   private readonly collections = new Map<string, CollectionState>()
   private readonly pendingRPCs = new Map<string, PendingRPC>()
@@ -135,15 +135,33 @@ export class BrowserCollectionCoordinator implements PersistedCollectionCoordina
     return this.disposed
   }
 
+  private requireAdapter(): AdapterWithPullSince {
+    if (!this.adapter) {
+      throw new Error(
+        `BrowserCollectionCoordinator: adapter not set. Call setAdapter() before using leader-side operations.`,
+      )
+    }
+    return this.adapter
+  }
+
   constructor(options: BrowserCollectionCoordinatorOptions) {
     this.dbName = options.dbName
-    this.adapter = options.adapter
+    this.adapter = options.adapter ?? null
     this.channel = new BroadcastChannel(`tsdb:coord:${this.dbName}`)
     this.channel.onmessage = (event: MessageEvent) => {
       this.onChannelMessage(event.data)
     }
   }
 
+  /**
+   * Set or replace the persistence adapter used for leader-side RPC handling.
+   * Called by `createBrowserWASQLitePersistence` to wire the internally-created
+   * adapter into the coordinator.
+   */
+  setAdapter(adapter: AdapterWithPullSince): void {
+    this.adapter = adapter
+  }
+
   // -----------------------------------------------------------------------
   // PersistedCollectionCoordinator interface
   // -----------------------------------------------------------------------
@@ -207,7 +225,7 @@ export class BrowserCollectionCoordinator implements PersistedCollectionCoordina
     spec: PersistedIndexSpec,
   ): Promise<void> {
     if (this.isLeader(collectionId)) {
-      await this.adapter.ensureIndex(collectionId, signature, spec)
+      await this.requireAdapter().ensureIndex(collectionId, signature, spec)
       return
     }
 
@@ -328,26 +346,27 @@ export class BrowserCollectionCoordinator implements PersistedCollectionCoordina
         lockName,
         { signal: abortController.signal },
         async () => {
-          if (this.disposed) return
-
-          state.isLeader = true
-
-          // Restore stream position from DB
-          if (this.adapter.getStreamPosition) {
-            const pos = await this.adapter.getStreamPosition(collectionId)
-            state.latestTerm = pos.latestTerm
-            state.latestSeq = pos.latestSeq
-            state.latestRowVersion = pos.latestRowVersion
-          }
+          if (this.isDisposed()) return
 
-          state.latestTerm++
+          try {
+            // Restore stream position from DB before claiming leadership
+            const adapter = this.requireAdapter()
+            if (adapter.getStreamPosition) {
+              const pos =
+                await adapter.getStreamPosition(collectionId)
+              state.latestTerm = pos.latestTerm
+              state.latestSeq = pos.latestSeq
+              state.latestRowVersion = pos.latestRowVersion
+            }
+
+            state.latestTerm++
+            state.isLeader = true
 
-          this.emitHeartbeat(collectionId, state)
-          state.heartbeatTimer = setInterval(() => {
             this.emitHeartbeat(collectionId, state)
-          }, HEARTBEAT_INTERVAL_MS)
+            state.heartbeatTimer = setInterval(() => {
+              this.emitHeartbeat(collectionId, state)
+            }, HEARTBEAT_INTERVAL_MS)
 
-          try {
             // Hold the lock until disposed or aborted
             await new Promise<void>((resolve) => {
               const onAbort = () => {
@@ -362,8 +381,10 @@ export class BrowserCollectionCoordinator implements PersistedCollectionCoordina
             })
           } finally {
             state.isLeader = false
-            clearInterval(state.heartbeatTimer)
-            state.heartbeatTimer = null
+            if (state.heartbeatTimer) {
+              clearInterval(state.heartbeatTimer)
+              state.heartbeatTimer = null
+            }
           }
         },
       )
@@ -592,7 +613,7 @@ export class BrowserCollectionCoordinator implements PersistedCollectionCoordina
     },
   ): Promise<RPCResponse> {
     await this.withWriterLock(() =>
-      this.adapter.ensureIndex(collectionId, request.signature, request.spec),
+      this.requireAdapter().ensureIndex(collectionId, request.signature, request.spec),
     )
     return {
       type: `rpc:ensurePersistedIndex:res`,
@@ -654,7 +675,7 @@ export class BrowserCollectionCoordinator implements PersistedCollectionCoordina
     }
 
     await this.withWriterLock(() =>
-      this.adapter.applyCommittedTx(collectionId, tx),
+      this.requireAdapter().applyCommittedTx(collectionId, tx),
     )
 
     // Track envelope for dedup
@@ -714,7 +735,8 @@ export class BrowserCollectionCoordinator implements PersistedCollectionCoordina
   ): Promise<PullSinceResponse> {
     const state = this.collections.get(collectionId)
 
-    if (!this.adapter.pullSince) {
+    const adapter = this.requireAdapter()
+    if (!adapter.pullSince) {
       return {
         type: `rpc:pullSince:res`,
         rpcId: request.rpcId,
@@ -726,7 +748,7 @@ export class BrowserCollectionCoordinator implements PersistedCollectionCoordina
       }
     }
 
-    const result = await this.adapter.pullSince(
+    const result = await adapter.pullSince(
       collectionId,
       request.fromRowVersion,
     )
diff --git a/packages/db-browser-wa-sqlite-persisted-collection/src/browser-persistence.ts b/packages/db-browser-wa-sqlite-persisted-collection/src/browser-persistence.ts
@@ -2,6 +2,7 @@ import {
   SingleProcessCoordinator,
   createSQLiteCorePersistenceAdapter,
 } from '@tanstack/db-sqlite-persisted-collection-core'
+import { BrowserCollectionCoordinator } from './browser-coordinator'
 import { BrowserWASQLiteDriver } from './wa-sqlite-driver'
 import type {
   PersistedCollectionCoordinator,
@@ -135,6 +136,13 @@ export function createBrowserWASQLitePersistence<
       ...(schemaVersion === undefined ? {} : { schemaVersion }),
     })
     adapterCache.set(cacheKey, adapter)
+
+    // Wire the adapter into the multi-tab coordinator so it can handle
+    // leader-side RPCs (applyCommittedTx, pullSince, ensureIndex, etc.)
+    if (resolvedCoordinator instanceof BrowserCollectionCoordinator) {
+      resolvedCoordinator.setAdapter(adapter)
+    }
+
     return adapter
   }
 
