feat(bps): lite pubsub

[acud]

Checklist

• I have read the coding guide.
• My change requires a documentation update, and I have done it.
• I have added tests to cover my changes.
• I have filled out the description and linked the related issues.

Description

Open API Spec Version Changes (if applicable)

Motivation and Context (Optional)

Related Issue (Optional)

Screenshots (if appropriate):

AI Disclosure

• This PR contains code that has been generated by an LLM.
• I have reviewed the AI generated code thoroughly.
• I possess the technical expertise to responsibly review the code generated in this PR.

[zelig]

Review against SWIP-74 rev 3 (ethersphere/SWIPs #111), which now specifies the challenge–response claim this PR started from. The findings were checked adversarially against the code, the spec and bee's p2p wrapper on master. (Edited: cursor semantics, test flow and counter list aligned with SWIP-74 rev 3 as pushed.)

State of the PR. Seven commits 2026-09-12..24; 424 lines of non-generated Go and
proto plus generated bps.pb.go; the description is the empty template; not wired into
the node; CI red — TestJoin fails with write join msg: stream closed (fullNode is
never set, so the handler resets every stream), and lint fails on the commit message
bps init. It is a scaffold — most
functions are stubs with "replace later" comments — so this review is about the shape
it commits to, not the stubs. It is also a restart: acud's previous bee PR #5597 (closed
2026-09-10, ~9,900 lines) implemented SWIP-60 rev 3 in full — bindings, broker, cohort,
session, WS bridge, metrics, hostile tests — and none of it is carried over yet; the
validation, queue and metrics machinery there is the obvious quarry for what is missing.

Headline

The scaffold follows SWIP-74 in three places and diverges from it in one that matters;
and it has one lifecycle error that no amount of filling in the stubs will fix.

Follows: one create-or-attach join (Jopen in the registry: "joins an existing cohort
or creates one by joining a previously unknown topic"); a per-cohort lastSeen index
"used to prevent replay and circumvent dedup logic" — that is the cursor; one frame type
both directions after the handshake (SWIP-74's Message, though the frame must carry the
whole chunk — see the wire table).

Diverges — and the divergence is the right one. The publisher role is claimed by a
challenge–response (JoinAck{challenge} → Claim{sig}), which is what SWIP-74 rev 3
now says: a joining peer is a receiver at once, and a peer that signs the challenge
upgrades to a publisher stream. SWIP-74 rev 2's static-preimage Auth-in-Join was
replayable by design, and a replayed signature would have bought an identity — a stream
carried as the admin's, exempt from the fan-out bound, and in SWIP-60 admitted to a
closed cohort; the challenge makes the identity worth exactly the key. What the scaffold
gets wrong is the shape of the challenge, in four ways (SWIP-74 rev 3, Handshake):

• the challenge is a per-cohort value (Cohort.challenge, rotated after each claim)
  handed to every joiner — it must be derived per address from a boot secret:
  S = H(S_C ‖ S_c ‖ addr) with S_C drawn once at boot and never persisted,
  S_c = H(S_C ‖ H(Marshal(spec))) (the spec's canonical serialisation, which also keys
  the registry), addr the address the joiner declares in its Join; the broker stores
  nothing and recomputes S at claim time, and an address that moves to another node
  keeps its claim;
• Claim opens a second stream — it must be sent on the joined stream, which
  upgrades in place; the second stream is why the admin's Join stream would otherwise
  stay a fan-out target receiving the admin's own messages;
• the claim takes over the cohort and rotates the challenge "so that the publisher
  can reclaim later" — no takeover: a reconnecting admin joins and claims on its new
  stream, concurrent admin streams are all publisher streams, the cursor arbitrates;
• Claim{sig} carries a bare signature over a bare challenge, on a stream with no
  topic — a claim is Claim{addr, index, auth} with auth = {r, s, v} over
  H("bps-claim:v1" ‖ S ‖ O_B ‖ index): domain-separated from SOC signatures, bound
  through O_B (the broker's overlay) to the verifier — otherwise a relay can forward
  the challenge and land the admin's signature on its own stream at the honest broker —
  and carrying the publisher's cursor, signed: the claim that its next message has an
  index of at least index, which a reconnecting admin uses to move the broker's cursor
  forward.

Lifecycle: the broker handler is fire-and-forget — it spawns the per-stream goroutine
on the handler's context and returns nil at once. In bee's libp2p wrapper the handler
context is cancelled the moment the handler returns (libp2p.go removes the stream and
calls its cancel), so on a real node every subscriber and publisher stream is torn down
right after the ack. streamtest hands the handler context.Background(), which is why
the test does not show it — and why the goroutine will instead leak and trip goleak
once fullNode is set. Every long-lived bee handler (pushsync, pullsync, retrieval)
blocks for the stream's lifetime for exactly this reason, and it is also the only way to
return p2p.NewBlockPeerError for a violation, which the spec needs (below).

Wire, message by message

PR #5626	SWIP-74	note
SystemMessage{oneof Join ǀ Claim}	no envelope	Join is the only first frame; after it, what a frame is follows from the stream's role — a subscriber stream sends at most one Claim, a publisher stream sends Message — so nothing needs a discriminator
Join{topic}	Join{CohortSpec{topic, binding, admin}, addr, claim?}	the broker needs topic to rebuild keccak256(topic ‖ index) and admin to check the owner; a hash of both (the registry comment: "feed topic + owner hash") gives it neither. The spec is the cohort's identity and the registry key — carry it, hash its canonical serialisation for the key. addr declares the address the stream will publish as; a returning publisher claims right here
JoinAck{challenge}	Ack{status, challenge} — OK / FULL / REJECTED	the challenge is right, and issued only to a Join that declared an address; the status is missing: the peer cannot tell FULL (capacity — back off and rejoin) from REJECTED (the spec itself is refused — rejoining unchanged is pointless); SWIP-74 mandates backoff after either
Claim{sig} on a fresh stream	Claim{addr, index, auth} on the joined stream	same stream, upgrade in place; signature over H("bps-claim:v1" ‖ S ‖ O_B ‖ index), not over the bare challenge — as sent, Claim names neither cohort, verifier nor cursor
ClaimAck{}	nothing	the client waits for it and the handler never sends it: Claim blocks until its context ends. No reply at all: a publisher pipelines its first Message behind the Claim, and a bad claim resets the stream
Broadcast{soc}	Message{address, data}	the chunk data alone has no address, so the ordinary SOC validation is vacuous (ecrecover always yields an address): carry the address, and the frame is a whole chunk validated by the ordinary path
bps/1.0.0, stream bps	pubsub/1.0.0	bee registers /swarm/<name>/<version>/<stream>; decide the name once, in the spec, and match it

The takeover semantics in the Claim comment ("the current stream becomes the publisher
stream and the next challenge changes randomly, so that ... the publisher can reclaim
later") is the supersede rule SWIP-74 considered and dropped: a reconnecting admin needs
no reclaim, its new stream is simply another publisher stream, and a stale one closing
later is a no-op.

Streams and state

1. Broker streams die after the ack (above). Make the handler resident: the read loop
   inline on a publisher stream, the writer loop draining a bounded per-stream queue
   inline on a subscriber stream; exits are the peer's EOF, the queue overflow, the
   inactivity reclaim. A violation returns p2p.NewBlockPeerError(...) so the wrapper
   resets and blocklists; a non-OK ack returns an error so the wrapper closes.
2. Subscriber streams are never read: the join branch is write-only. SWIP-74 item 5
   (a Message on a subscriber stream: drop, reset, blocklist, count wrong_stream) is
   unsatisfiable in this shape, and a subscriber that closes is not noticed until the
   next fan-out write fails, so idle streams keep counting against the per-cohort bound.
3. Client-side lifetimes: every error path after NewStream returns without
   stream.Reset() (bee's idiom is Reset on error, FullClose on a clean end — the
   CI failure is such a path); the subscription's whole life is the caller's context, so
   a request-scoped one kills it on return and Background makes it uncancellable, and
   Join returns no handle; rxCh is never closed, so a consumer ranging over it hangs
   after the stream dies; in Claim, once the writer goroutine exits on a broker reset
   nobody reads ch and the publisher's next send blocks forever. Join/Claim should
   return a handle (close, done, error), the Service should own a long-lived context.
4. Subscriber drops deliveries: select { case rxCh <- msg.Soc: default: } on an
   unbuffered channel discards any frame the consumer is not ready for at that instant.
   Loss policy is the broker's (bounded queue, reset on overflow); the subscriber blocks
   or buffers, and re-verifies (SWIP-74 item 10).
5. Identity is modelled as a node: Cohort.publisher swarm.Address, Jopen(overlay, topic), members map[string]string. SWIP-74 binds the identity to the stream, never
   the node — the same key works from any node, and several streams may be the admin's
   at once — and a cohort is a set of streams each with its own queue, which a map of
   strings cannot hold. And lastSeen models the cursor the wrong way round: SWIP-74's
   cursor is the lowest index accepted next, a plain uint64 initially 0, moved to
   n + 1 on acceptance and to max(cursor, index) on a claim, so index 0 is accepted
   on a fresh cohort without an "absent" value.

Not started, and expected not to be

Validation on the publish path (id substitution, SOC check with the address, owner ==
admin, index ≥ cursor — lastSeen is declared and never read), fan-out from the registry,
the bounds and the inactivity deadline, the three statuses with backoff, and the eight
counters. All of it exists in some form in #5597.

Tests

TestJoin asserts a challenge round trip and never sends a Message, so it would pass
with no validation, no cursor and no fan-out; streamtest.Records waits for the handler
to return, so it can only pass with the fire-and-forget handler that has to go, and once
fullNode is set the leaked goroutines fail goleak. Suggested flow: join a subscriber,
join the admin declaring addr, claim on the challenge from its Ack (and once more with
the claim in the Join), publish index 0, receive it on the subscriber, close both, then
Records. Cases worth their own test: index 0 accepted on a fresh cohort; n < cursor
counted as a retransmit; owner ≠ admin dropped; a subscriber Message → reset and
blocklist; two concurrent admin streams arbitrated by the cursor; inactivity reclaim.

Small things

Claim sends Sig: topic; two log lines say "read join ack" in the write loop and the
broadcast loop; the package doc still says "hello-world wire protocol used to greet other
peers"; TestJoin prints the challenge and its failure message says "want 2" for a check
of 1; registry_test.go is a bare package line; a SystemMessage with neither member
set falls through to return nil with the stream neither closed nor reset.

What to ask for

• Join{CohortSpec, addr, claim?}, Ack{status, challenge}, Claim{addr, index, auth},
  Message{address, data} — field numbers as SWIP-74 rev 3 fixes them; keep Jopen, key
  the registry by the canonical serialisation of the spec; rename Broadcast to Message;
  drop SystemMessage and ClaimAck.
• A resident handler; a read loop on subscriber streams that accepts at most one Claim
  and treats everything else as a violation; BlockPeerError on violation.
• No challenge state: one boot secret, S recomputed at claim time; the declared addr
  and the role per stream; several admin streams; no takeover; the cursor as "lowest index
  accepted next", set forward by the claim.
• The p2p handshake should verify that a peer's signed address record names the
  connection's authenticated peer ID — the claim's binding to the broker's overlay rests on
  it, and bee's handshake checks the record, not the binding, today. Separate PR.
• Client handles with close/done; Reset on error paths; close rxCh; no drop at the
  subscriber.
• Then the validation steps and the cursor, fan-out to subscriber streams only, bounded
  per-stream queues, the bounds and the inactivity deadline, statuses with backoff, the
  counters — from feat: bps protocol draft #5597 where it fits.
• The protocol name decided in one place.

Open

• The protocol name, bps in code versus pubsub in both SWIPs.
• The subscribers-per-cohort bound admits one extra stream while the admin is absent — a
  Join declaring the admin's address — and disconnects it if it has not claimed within
  the claim deadline (SWIP-74 rev 3, Resource bounds).