fix(pubsub): eagerly initialize topic in subscribe() to prevent race

[Faolain]

WIP - But would love a review/comments (Goal to eliminate flakes)

Edit 1: ongoing research here Faolain#5 - which bundles in the 589 fix with another hypothesized shared-log issue fix...tests run multiple times pass locally but thenI saw a failure in CI...investigating

Edit2: Had codex try its hand and all tests are passing Faolain#6 which looks promising, builds on this branch and tests pass

tl;dr on an app I've been developing with peerbit I had some patches (one of which was the previously merged #538) where when I updated all my peerbit dependencies for the first time in a while today, and removed my patches (assuming things were far more performant/stable) I noticed tests breaking and connectivity being far worse.

I figured to make a few PRs for those patches I had aside from the initial one. However I realized the initial commit here was having relayers track subscribers (here a2b642e) and I wasn't sure if that was intended so the subsequent commits I removed that but maybe it's something needed for peerbit?

The PR explicitly says they avoided initializing topics in the incoming Subscribe handler because it broke “rejoin with different subscriptions,” and they want to preserve “only track topics you care about.”
That’s coherent if DirectSub nodes are not supposed to act as full routing relays for topics they aren’t subscribed to. If your “relay” use case depends on tracking remote subscribers even when the relay itself isn’t subscribed, then this PR intentionally does not solve that (it fixes only the “I called subscribe but debounce hasn’t fired yet” case).

In any case as I continued to slog forward (this patch improved connectivity) I realized it led to issues elsewhere with the tests, since it led to far faster connections it meant that race conditions became more apparent (hence the failures in the CI, maybe?) So this document is a living bible to try and squash these race conditions/flaky tests once and for all! hehe I figured the more information the better, apologies for the walls of text.

Summary

subscribe() debounces through debounceSubscribeAggregator, but the actual _subscribe() handler (which calls initializeTopic() and sets subscriptions) only fires after the debounce window. If a remote Subscribe message arrives in that window, two things go wrong:

1. this.topics.get(topic) returns undefined → the remote subscription is silently dropped
2. Even if the topic IS initialized, getSubscriptionOverlap() returns empty (subscriptions not set yet) → the requestSubscribers response is empty → the sender never learns about our subscription

This causes a race condition: when two peers subscribe to the same topic concurrently, one peer's Subscribe message may arrive at the other before the debounced _subscribe() has fired, causing asymmetric subscription state where only one peer knows about the other.

Problem

In DirectSub, subscribe() delegates to a debounce aggregator:

async subscribe(topic: string) {
    return this.debounceSubscribeAggregator.add({ key: topic });
}

The debounced _subscribe() eventually calls listenForSubscribers() → initializeTopic() and sets this.subscriptions. But there's a timing gap between calling subscribe() and _subscribe() firing.

During this gap, incoming Subscribe messages hit two problems:

Problem 1 — Topic not initialized:

const peers = this.topics.get(topic);  // ← undefined
if (peers == null) { return; }         // ← silently drops

Problem 2 — Response empty even if topic IS initialized:

const mySubscriptions = this.getSubscriptionOverlap(pubsubMessage.topics);
// getSubscriptionOverlap checks this.subscriptions, which isn't set yet
// → returns [] → no response sent → sender never learns about us

Downstream impact

Discovered while building a browser-based P2P application using Peerbit. Peers would intermittently fail to discover each other's subscriptions due to this timing-dependent race, causing message delivery failures.

Fix (two parts)

Part 1: Eagerly initialize topic in subscribe()

async subscribe(topic: string) {
    if (!this.topics.has(topic)) {
        this.initializeTopic(topic);
    }
    return this.debounceSubscribeAggregator.add({ key: topic });
}

This ensures the topic Map entry exists immediately, so incoming Subscribe messages aren't dropped.

Part 2: Include pending subscriptions in requestSubscribers response

if (pubsubMessage.requestSubscribers) {
    const mySubscriptions = this.getSubscriptionOverlap(pubsubMessage.topics);
    // Also include topics with a pending subscribe (debounce not yet fired).
    // This handles the race where subscribe() was called but _subscribe()
    // hasn't executed yet, so subscriptions isn't set but we should still respond.
    for (const topic of pubsubMessage.topics) {
        if (!mySubscriptions.includes(topic) &&
            this.debounceSubscribeAggregator.has(topic)) {
            mySubscriptions.push(topic);
        }
    }
    if (mySubscriptions.length > 0) { /* send response */ }
}

This ensures the requestSubscribers response includes topics where subscribe() was called but _subscribe() hasn't executed yet, using the existing debounceSubscribeAggregator.has() check (same pattern as unsubscribe()).

Why this approach

An earlier iteration added initializeTopic() unconditionally in the Subscribe handler, but this broke the "rejoin with different subscriptions" test — after a peer restarts and subscribes to only topic B, it would incorrectly initialize topic A from a remote Subscribe message.

The eager-init + debounce-check approach is more targeted: it only affects topics the local node has explicitly called subscribe() for, preserving the existing behavior that nodes only track topics they care about.

Testing

Included three new test cases in test/bug1-initializeTopic-race.spec.ts:

1. Unit test — eager initialization: Verifies topics.has(topic) is true immediately after subscribe(), before the debounce fires.

2. Integration — concurrent subscribe + connect: Both peers subscribe and connect simultaneously. Verifies both peers discover each other's subscriptions despite the race-prone ordering.

3. Integration — subscribe after connect (normal path): Peers connect first, then subscribe. Verifies the normal (non-race) path still works correctly.

All 40 existing pubsub tests pass (including "rejoin with different subscriptions") plus the replicate test in packages/log (192/192 pass) plus 3 new regression tests.

Notes

• initializeTopic() is idempotent (guards with this.topics.get(topic) ||), so calling it before the debounce is safe.
• debounceSubscribeAggregator.has() is already used in unsubscribe() for the same purpose (checking pending subscribes).
• The Subscribe handler in onDataMessage() is unchanged — the peers == null guard remains.
• I am happy to allow edits by maintainers on this PR.

CI Failures: Pre-existing shared-log race condition

Definitive finding: The shared-log test failures are caused by a pre-existing TOCTOU race condition in @peerbit/shared-log, not by this PR's pubsub changes.

Verification

• Master (clean, no patch): test:ci:part-4 passes 1744/1744 with 0 failures
• With this PR's fix: 3 test failures appear in shared-log (events, migration-8-9)

The failures are exposed by faster subscription discovery (this fix removes the ~50ms debounce delay), but the underlying bug is in shared-log.

Root cause (shared-log onMessage handler)

In SharedLog.onMessage(), incoming AllReplicatingSegmentsMessage is processed via async IIFEs (~line 2971). When subscriptions resolve faster, two concurrent AllReplicatingSegmentsMessage handlers for the same peer both:

1. Check prevCount === 0 in the replication index
2. Both see 0 (neither has written yet)
3. Both set isNewReplicator = true
4. Both emit replicator:join → duplicate event

This is a classic TOCTOU race: check-then-act without serialization. On master, the slower sequential handshake (debounce delay) serializes these messages, hiding the race.

Specific test failures

Test	Expected	Got	Cause
events > replicate:join not emitted on update	1 join event	2 join events	Duplicate AllReplicatingSegmentsMessage processing
migration-8-9 > replicates database of 1 entry	1 replicated entry	0 entries	Timing shift in RequestReplicationInfoMessage

Evidence this is pre-existing

• Master CI has had shared-log failures (Jan 27: "Timeout waiting for mature replicators")
• PR fix(pubsub): eagerly initialize topic in subscribe() to prevent race #589's first two CI runs failed in @peerbit/log (timeout), not shared-log
• @peerbit/document tests flake across multiple unrelated branches
• The concurrent addReplicationRange without per-peer serialization was always a latent bug

Recommendation

The shared-log TOCTOU race should be fixed separately (e.g., serializing replication info processing per-peer, or deduplicating handleSubscriptionChange calls during setup). A CI re-run may pass due to timing variance.

[Faolain]

AI Review 1:

AI Review 2:
https://chatgpt.com/share/69856c83-5a10-8002-9936-3bb84d811b3c

AI Review 3:
https://chatgpt.com/share/69856cad-16a0-8002-a9b6-c81596594eef

[Faolain]

Closing in favor of #593