add suspend controller for replication

[tur-nr]

This adds a way to pause replication without dropping connections. SuspendController works like an AbortController: you hold the controller and pass controller.signal into cores via the new suspendSignal option. One signal can be shared by any number of cores, and a core opened while the signal is suspended comes up suspended.

While suspended, a core sends no requests, uploads no blocks and drops incoming data (clearing the request so it gets re-fetched later). Incoming requests are queued in the existing receiver queue and answered on resume, and the replicator listens for the signal's resume event to serve that backlog and re-request whatever was dropped. Wants are still answered since they're served from memory. replicator._replicationBusy() lets the caller wait until in-flight serves/verifies have finished after suspending.

The reason we need this: consumers can briefly suspend hypercore to release resources and locks, for example suspending the underlying storage, whilst keeping the swarm and peers warm. Without it, any replication traffic against suspended storage parks forever, so the only safe option today is tearing connections down first. With this, storage can suspend underneath live connections and everything picks up where it left off on resume, no reconnect needed.

Cores that don't pass a signal are completely unaffected.

[lejeunerenard]

While reviewing I found a few fixes and since they were simple, I just pushed them. There were:

• replicator.busy didn't reflect push mode requests which also use storage.
• Possible deadlock if throwing an error while handling a request after resuming suspension.

Also pointed out how there is still an issue around guarding against suspending while handling requests that were queued.

Reviewed with help from AI.

[lejeunerenard]

The fact that isActive() is based on suspended means that fulfilling request can bail here leaving requests not fulfilled. This can deadlock in the following test:

test('suspended replication queues incoming requests, doesnt throw when suspending microtick after resume', async function (t) {
  const controller = new Hypercore.SuspendController()

  const a = await create(t, null, { suspendSignal: controller.signal })
  const b = await create(t, a.key)
  await a.append(['a', 'b', 'c'])

  const bAppended = once(b, 'append')
  replicate(a, b, t)
  await bAppended // ensure b has latest length

  controller.suspend()
  const get = b.get(0)
  const early = await Promise.race([get.then(() => 'served'), sleep(300).then(() => 'pending')])
  t.is(early, 'pending', 'request not served while suspended')

  controller.resume()
  await eventFlush()
  controller.suspend()

  await eventFlush()
  controller.resume()

  t.alike(await get, Buffer.from('a'), 'queued request served after resume')
})

Note you need to mine to hit the scenario.

So when handling queued requests the while checks if the replicator is suspended, but it can become suspended mid-_handleRequest().

[tur-nr]

Fixed in 6df919f

[lejeunerenard]

Thanks for the fixes. If I understand the problem we're trying to solve, IO is the culprit for issues suspending, not networking traffic. This means we could keep more network sent while suspended. Then we should refactor so isActive() doesn't include the replicator being suspended as it was primarily used for 'bad behaving' peers with the exception of the core being frozen due to a conflict.

[lejeunerenard]

This fix requeues requests already completed to avoid the isActive() check below (mentioned here: https://github.com/holepunchto/hypercore/pull/851/changes#r3617257798). This is the wrong approach because the issue is not that the request shouldn't be fulfilled, but that a valid response is being dropped. The replicator can be suspended while its going to storage for fulfilling the request. Afterwards there is no more storage IO to block suspension, so we can send it correctly.

The issue is that isActive() is used here as a guard against 'bad behaving' peers rather than whether the replicator is active. I think this actually points out that using isActive() as a guard is misusing it as it is a per Peer method. We should add replicator.suspend instead to all places that .isActive() is used currently. The following line would need to be reverted too as it was excluding suspension as the reason to not reply to wants:
https://github.com/holepunchto/hypercore/pull/851/changes#diff-3344a84994d8e172768f33ec4bf514ba2bc28eb7eeffff10fcbbd294efe56edcR2863

There are other places that currently were skipped when suspended (such as broadcasting the range) that could be allowed when suspended since they dont use IO / storage, much like responding to wants.

If we do want to stop messages sending while suspended even after the IO is done, we can requeue the msg, but:

1. dont need to check this.removed as it cannot be true given we just checked that the remoteRequest for it. It would be cleared if removed.
2. Can't check proof.block as the fulfilled proof could be null etc.
3. Should be moved to before we remove the request from the remoteRequests right above.

[tur-nr]

Thanks for review again. Sorry I misunderstood what is active means so yeah putting the suspended check there isn't the solution.

Yes you are correct with network can continue and suspend only when idle.

I've reverted the isActive work and added distinct .suspended checks at points where new or upgraded blocks are request.

Fixed: 36c46f2