fix(sync): gate peer 'started' status on the peer's own Synchronize, not core.update()

Flip status to 'started' when the peer's own first Synchronize wire
message is processed - hypercore records exactly this per-peer fact as
peer.remoteSynced, set synchronously at the top of peer.onsync(). We
intercept by shadowing onsync on the peer instance, the same
interception already used for onbitfield/onrange (hypercore dispatches
wire messages by property lookup on the channel's userData). This fixes
all three failure modes of the update()-based signal covered by the
previous commit's tests. It also removes an unhandled-rejection hazard:
the old .then() had no rejection handler, and hypercore rejects pending
update() calls when a core closes.

One real case exists where the first Synchronize never arrives on a live
channel: a close/open crossing on the same protomux - both sides disable
and re-enable a namespace around the same time, which routinely happens
during the invite flow's capability churn. One side ends up with a fresh
peer holding no baseline while the remote believes its old session is
still live, so it never re-sends its state, and hypercore does not
recover on its own. There is no reliable in-band repair: no message we
can send guarantees a Synchronize in reply, the duplicate channel open
is dropped by protomux before hypercore sees it, and forcing a channel
renegotiation from this observation layer destabilizes the layers that
own the channel lifecycle (tried; it could strand a peer with no channel
at all). So after 5s the peer is treated as started with its
channel-level state unknown. This does not fabricate sync state: a
peer's blocks are counted from its pre-have bitfields (broadcast over
the project creator core at connect and on append) merged with the
channel-level bitfield, so a peer that has data is still counted
correctly - the fallback stops gating counts that are already correct,
and a Synchronize arriving later still updates state. The root fix is
upstream: protomux channel renegotiation needs a close reason / channel
generation (hypercore's own Peer#onclose has a TODO to this effect).

Peers whose Synchronize was processed before the core was attached are
detected via remoteSynced and marked started immediately - defensive
only: production wiring attaches cores synchronously on 'add-core',
before a channel handshake can complete, but nothing in this class
enforces that.

Adds tests for the fallback (a silent session must not block sync state
forever, asserted to have resolved without a Synchronize) and for the
attach-late invariant.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CzqT9t19AV5voYFM2vYV8h

Author: claude

src/sync/core-sync-state.js

@@ -43,6 +43,15 @@ import { InvalidBitfieldIndexError } from '../errors.js'
 // This is a 32-bit number with all bits set
 const WANT_FULL = 2 ** 32 - 1
 
+/**
+ * How long after `peer-add` to wait for the peer's first Synchronize message
+ * before treating the peer as started with its channel-level state unknown
+ * (see `#onPeerAdd`). Normally the Synchronize arrives in the same batch as
+ * the channel open, so this is only hit when a channel close/open crossing
+ * swallowed the handshake.
+ */
+const FIRST_SYNC_FALLBACK_MS = 5_000
+
 /**
  * Track sync state for a core identified by `discoveryId`. Can start tracking
  * state before the core instance exists locally, via the "preHave" messages
@@ -279,12 +288,6 @@ export class CoreSyncState {
     const peerState = this.#getOrCreatePeerState(peerId)
     peerState.status = 'starting'
 
-    this.#core?.update({ wait: true }).then(() => {
-      peerState.status = 'started'
-      peerState.contiguousLength = peer.remoteContiguousLength
-      this.#update()
-    })
-
     // A peer can have a pre-emptive "have" bitfield received via an extension
     // message, but when the peer actually connects then we switch to the actual
     // bitfield from the peer object
@@ -294,7 +297,9 @@ export class CoreSyncState {
     this.#update()
 
     // We want to emit state when a peer's bitfield changes, which can happen as
-    // a result of these two internal calls.
+    // a result of these two internal calls. Hypercore dispatches wire messages
+    // by property lookup on this peer object (its channel's userData), so
+    // shadowing the methods on the instance is the interception point.
     const originalOnBitfield = peer.onbitfield
     const originalOnRange = peer.onrange
     peer.onbitfield = (...args) => {
@@ -307,6 +312,79 @@ export class CoreSyncState {
       peerState.contiguousLength = peer.remoteContiguousLength
       this.#update()
     }
+
+    // The peer is "started" once its first Synchronize message has been
+    // processed (hypercore sets `peer.remoteSynced`): only then do we know
+    // the peer's length and fork, so only then can counts about this peer be
+    // trusted. Every channel sends a Synchronize immediately on open, so this
+    // always arrives. Do NOT use `core.update({ wait: true })` as a proxy for
+    // this: it answers a core-global question ("am I up to date with the
+    // swarm?") which (a) resolves immediately for writable cores, (b)
+    // resolves for all waiters when *any* peer supplies an upgrade, and (c)
+    // samples at most 3 peers (hypercore's MAX_PEERS_UPGRADE) — so it can
+    // report a peer as started before we have heard anything from it.
+    const markStarted = () => {
+      peerState.status = 'started'
+      peerState.contiguousLength = peer.remoteContiguousLength
+      this.#update()
+    }
+    if (peer.remoteSynced) {
+      // Defensive only — not reachable in production wiring: cores are
+      // attached synchronously on the core manager's 'add-core' event (or at
+      // construction), before a channel handshake could possibly complete,
+      // so peers replayed by `attachCore` have never synced yet. But nothing
+      // in this class enforces that callers attach before handshakes
+      // complete, so handle an already-synced peer correctly rather than
+      // waiting for a Synchronize that already happened.
+      markStarted()
+    } else {
+      const originalOnSync = peer.onsync
+      peer.onsync = (...args) => {
+        const result = originalOnSync.apply(peer, args)
+        // `remoteSynced` is set synchronously at the top of `onsync`
+        if (peer.remoteSynced) {
+          if (peerState.status === 'starting') {
+            clearTimeout(firstSyncFallback)
+            markStarted()
+          } else {
+            // A Synchronize arriving after the fallback below fired still
+            // carries state (the peer's length and fork), so re-derive
+            peerState.contiguousLength = peer.remoteContiguousLength
+            this.#update()
+          }
+        }
+        return result
+      }
+      // The remote sends its first Synchronize as soon as the channel
+      // opens, so normally this timer never fires. The exception is a
+      // renegotiation bug in hypercore/protomux (their `Peer#onclose` has a
+      // TODO about it): when both sides close and re-open a channel at the
+      // same time — e.g. a namespace toggling off and on around a role
+      // change — the messages cross and the remote keeps its old session,
+      // never re-sending its state, so we would wait forever. This is not
+      // repairable from here (no message forces a re-send, and
+      // closing/re-opening the channel from this layer destabilizes the
+      // replication layers that own it), only upstream. So stop waiting and
+      // mark the peer started. The cost is the same as any reconnect, and
+      // the same as the previous `core.update()`-based code: until this
+      // peer reconnects we only know about this core what its pre-haves
+      // told us (what it had at connect, plus its own writer-core appends),
+      // not blocks it downloaded from third devices mid-session, so its
+      // `have`/`wanted` counts can run low.
+      const firstSyncFallback = setTimeout(() => {
+        if (peer.remoteSynced || peer.removed) return
+        if (peerState.status !== 'starting') return
+        this.#l.log(
+          'Peer %S sent no Synchronize %dms after channel open, marking started',
+          peerId,
+          FIRST_SYNC_FALLBACK_MS
+        )
+        markStarted()
+      }, FIRST_SYNC_FALLBACK_MS)
+      // Don't keep the process alive for this; it matters only while the
+      // connection (which itself holds the process open) is live
+      firstSyncFallback.unref?.()
+    }
   }
 
   /**

test/sync/core-sync-state.js

@@ -640,6 +640,81 @@ test('peer status stays "starting" for a 4th concurrent peer (hypercore upgrade
   await waitForStatus(crs, emitter, fourthPeerId, 'started')
 })
 
+test('peer status falls back to "started" if the first Synchronize never arrives', async (t) => {
+  // A channel close/open crossing can leave the remote never re-sending its
+  // Synchronize (it believes its old session is still live) — see the
+  // comment in CoreSyncState's #onPeerAdd. A live channel in that state
+  // must not block sync state forever: after a fallback timeout the peer is
+  // treated as started, with its blocks still counted via pre-haves.
+  const localCore = await createCore(t)
+  await localCore.append(['a', 'b', 'c'])
+
+  const emitter = new EventEmitter()
+  const crs = new CoreSyncState({
+    onUpdate: () => emitter.emit('update'),
+    peerSyncControllers: new Map(),
+    namespace: 'auth',
+    deviceId: '',
+    hasDownloadFilter: () => false,
+  })
+  crs.attachCore(localCore)
+
+  // Never released: the Synchronize never arrives
+  holdSynchronize(localCore)
+  const remoteCore = await createCore(t, localCore.key)
+  const kp2 = NoiseSecretStream.keyPair(Buffer.allocUnsafe(32).fill(1))
+  const peerId = kp2.publicKey.toString('hex')
+  const destroy = replicate(localCore, remoteCore, { kp2 })
+  t.after(destroy)
+
+  await once(localCore, 'peer-add')
+  assert.equal(
+    crs.getState().remoteStates[peerId].status,
+    'starting',
+    'peer is "starting" while the Synchronize is missing'
+  )
+
+  // 5000ms fallback (FIRST_SYNC_FALLBACK_MS) plus margin
+  await waitForStatus(crs, emitter, peerId, 'started', 8_000)
+  const livePeer = localCore.peers.find((p) =>
+    p.remotePublicKey.equals(kp2.publicKey)
+  )
+  assert.equal(
+    livePeer?.remoteSynced,
+    false,
+    'sanity: "started" came from the fallback, not a Synchronize'
+  )
+})
+
+test('peer status becomes "started" for a peer that synced before the core was attached', async (t) => {
+  // Defensive invariant: production wiring attaches cores before their
+  // channel handshakes can complete, but nothing in CoreSyncState enforces
+  // that. A peer whose Synchronize was already processed when the core is
+  // attached must become "started" without waiting for another message.
+  const localCore = await createCore(t)
+  await localCore.append(['a', 'b', 'c'])
+  const remoteCore = await createCore(t, localCore.key)
+  const kp2 = NoiseSecretStream.keyPair(Buffer.allocUnsafe(32).fill(1))
+  const peerId = kp2.publicKey.toString('hex')
+  const destroy = replicate(localCore, remoteCore, { kp2 })
+  t.after(destroy)
+
+  await once(localCore, 'peer-add')
+  await waitFor(() => localCore.peers[0].remoteSynced === true)
+
+  const emitter = new EventEmitter()
+  const crs = new CoreSyncState({
+    onUpdate: () => emitter.emit('update'),
+    peerSyncControllers: new Map(),
+    namespace: 'auth',
+    deviceId: '',
+    hasDownloadFilter: () => false,
+  })
+  crs.attachCore(localCore)
+
+  await waitForStatus(crs, emitter, peerId, 'started')
+})
+
 /**
  * Intercept and queue the first Synchronize message(s) from peers that are
  * added to `core` after this is called, so tests can hold a peer in the