fix(pubsub): disconnect hygiene and once-per-subscription mcache replay

[gmelodie]

What was wrong?

Issue #1403

Follow-ups from the review of #1398:

• _handle_dead_peer returned too early. A peer's subscriptions arrive on its inbound stream, so it can sit in peer_topics while the outbound stream was never registered in peers. The if peer_id not in self.peers: return guard ran before the peer_topics cleanup, so a peer that dropped while half-registered left stale subscription entries behind forever.
• Replay had no gating. _send_recent_messages_to_new_peer replayed the whole mcache window for every topic the peer appeared subscribed to, with no equivalent of handle_subscription's was_newly_added guard. Combined with the stale entries above, a messy disconnect + reconnect could hand a peer the full window again; the only thing saving us was receiver-side seen-cache dedup.
• hasattr probes. pubsub.py reached into the router with four hasattr(self.router, ...) checks plus # type: ignore[attr-defined] on every call, so the optional hooks were invisible to type checkers and to anyone implementing a router.

How was it fixed?

Disconnect hygiene — _handle_dead_peer now runs _clear_pending_announce_retries_for_peer and a new _forget_peer_subscriptions before the peers early return. Stream teardown, queue close, and router.remove_peer still run only for fully-registered peers, since router registration happens in lockstep with self.peers (no await between router.add_peer and self.peers[peer_id] = stream). _forget_peer_subscriptions also replaces the open-coded loop in _teardown_if_connected, so the blacklist path gets the same cleanup.

Sender-side replay gating — all mcache replay now goes through one method, _replay_recent_messages(peer_id, topic), called by both _handle_new_peer and handle_subscription. It records (peer, topic) in _replayed_recent_topics so a peer receives the window at most once per subscription. Two details matter:

• The gate is only spent when the peer is actually writable (peer_id in self.peers). Marking it while the outbound stream is still pending would make the later _handle_new_peer replay a no-op and reintroduce the exact bug fix(pubsub): replay recent messages once a peer's outbound stream is registered #1398 fixed.
• The entry is dropped on unsubscribe, on _handle_dead_peer, and on blacklist, so a genuine resubscribe or reconnect replays again.

Router interface — IPubsubRouter declares flush_pending_messages and send_recent_messages as non-abstract async hooks with no-op bodies. GossipSub's existing implementations become real overrides, FloodSub inherits the no-ops, and all four hasattr probes and their type: ignore comments are gone.

Not in this PR: the GossipSub.publish queueing item from the issue. It is filed there as a separate design — peers known only via in-flight connection setup are in neither peers nor peer_topics at publish time, and changing when publish queues speculatively affects every pubsub user.

Tests

• TestRecentMessageReplayGate (4 tests): replay runs once per subscription; is deferred but not spent before the peer is writable; runs again after unsubscribe/resubscribe; runs again after disconnect/reconnect.
• test_dead_peer_clears_subscriptions_of_half_registered_peer — a peer in peer_topics but never in peers is cleaned up.
• test_router_replay_hooks_default_to_noop — FloodSub is callable through the new interface hooks.
• test_publish_before_identify_with_subscription_before_stream now waits on wait_for_peer instead of trio.sleep(publisher_stream_delay + 4), as the issue suggested. Runtime drops from ~7s to ~1.4s.

tests/core/pubsub is green (408 passed), as are ruff check, ruff format --check on the touched files, mypy -p libp2p, and pyrefly check.

To-Do

• Clean up commit history
• Add or update documentation related to these changes — docstrings on the new IPubsubRouter hooks and _replay_recent_messages; no user-facing docs describe these internals
• Add entry to the release notes — 1403.bugfix.rst and 1403.internal.rst

Cute Animal Picture