feat: pubsub by nugaon · Pull Request #5435 · ethersphere/bee

nugaon · 2026-04-14T14:21:35Z

pubsub

A brokered publish/subscribe protocol over Bee's p2p layer, exposed to end-users via WebSocket.

Overview

One node acts as a broker: it accepts p2p streams, validates incoming messages, and fans them out to all connected receivers. Other nodes connect as either a publisher (publish + receive) or a subscriber (receive only).

Connection flow

WebSocket client
      │  HTTP GET /pubsub/{topic}  (broker peer multiaddr in HTTP request headers)
      ▼
api/pubsub.go  ──►  pubsub.Service.Connect  ──►  mode.Connect (opens p2p stream)
                                                         │
                                               Broker node — brokerHandler
                                                     ├─ readwrite=1  ──►  handlePublisher
                                                     └─ readwrite=0  ──►  handleSubscriber

The connecting node receives a SubscriberConn which is bridged to the WebSocket: inbound WS frames are forwarded to the p2p stream (publishers only); p2p frames from the broker are decoded and forwarded to the WS client (all connections).

Mode system

Protocol-specific behaviour is isolated behind a Mode interface. The Service is mode-agnostic: it delegates header construction, message reading/validation, and broadcast formatting to the active mode. New protocol variants can be added by implementing Mode and registering a mode ID in newMode.

GSOC Ephemeral mode (mode 1)

Messages are Single Owner Chunks (SOC) signed by the holder of the topic's private key. The topic address is derived from the SOC owner public key and an arbitrary topic ID, so only the key holder(s) can publish and the messages carry full SOC wrapping.

Publisher → Broker (p2p)

[ sig: 65 B ][ span: 8 B LE ][ payload: up to 4 KB ]

The broker verifies the ECDSA signature on every message before broadcasting.

Broker → Subscriber (p2p)

Every broker frame begins with a 1-byte message type. 0x01 is reserved at the service level and is valid across all modes; all other type bytes are mode-specific and decoded by the active mode implementation.

First byte	Level	Meaning	Payload
`0x01`	service	Ping	(no fields — keepalive sent every 30 s)
`0x02`	GSOC mode	Handshake + data	`[SOC ID: 32 B][owner: 20 B][sig: 65 B][span: 8 B][payload: N B]`
`0x03`	GSOC mode	Data only	`[sig: 65 B][span: 8 B][payload: N B]`

The handshake frame (SOC identity metadata) is sent once per topic on the first broadcast. Subsequent messages are data-only. Ping frames are consumed by readServiceMessage before mode dispatch and never surfaced to the WebSocket client.

WebSocket (both directions)

Both the publisher (inbound) and subscriber (outbound) see the raw SOC payload:

[ sig: 65 B ][ span: 8 B ][ payload: N B ]

The node handles all p2p framing and signature verification transparently.

Multi-WebSocket multiplexer

Multiple WebSocket sessions to the same topic on the same node share a single p2p stream to the broker via SubscriberConn:

CreateSubscriberConn increments a ref count instead of opening a new stream when one already exists
A single runMux goroutine fans out incoming broker messages to all per-session channels; on stream error it immediately clears the shared conn so new sessions get a fresh stream
RemoveSubscriberConn decrements refs; FullClose is called exactly once when refs reach zero

API

Method	Path	Description
`GET` (WS upgrade)	`/pubsub/{topic}`	Connect to a topic as subscriber or publisher
`GET`	`/pubsub/`	List active topics with role and connection count

Query params for /pubsub/{topic}:

peer (required): multiaddr of the broker peer
gsoc-eth-address + gsoc-topic (optional): upgrade to publisher role

Header:

swarm-keep-alive (optional): WebSocket ping period in seconds (default: 60)

Configuration

Flag	Default	Description
`--pubsub-broker-mode`	`false`	Enable broker role on this node

Subscriber nodes use ConnectAllowLight when dialing the broker, so broker nodes can run in light-node mode.

Extensibility

New modes — implement Mode for a different message format and register a mode ID.
Access control — validatePublisher can enforce allow-lists, or stake checks.
Replay — handleSubscriber can replay on connect.
WASM / event-driven clients — broadcast can forward to an event emitter, making the broker embeddable in a WASM-compiled client without a separate p2p layer.
Per-subscriber transforms — formatBroadcast receives the individual brokerSubscriber, enabling selective filtering or transformation.

Tests

pkg/pubsub (unit)

mode_1_test.go — GSOCEphemeralMode: message formatting (handshake vs data), ReadBrokerMessage for ping/handshake/data/invalid-sig, setGsocParams address mismatch and valid params, SubscriberConn fanout, unsubscribe, and full-channel overflow.
pubsub_test.go — broker handler error paths (disabled, missing/unknown headers) and Service.Topics() (empty list, broker-role registration via spinlock).

pkg/api (integration)

pubsub_test.go — nil service → 404, empty list → 200, missing peer/invalid multiaddr/bad GSOC params → 400, subscriber WS connect/teardown, and a two-topic scenario: opens two concurrent WS sessions, writes broker pings through the mock p2p streams, queries GET /pubsub/ asserting both topics are listed, then closes each connection sequentially verifying the topic count decrements to 0.

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)

SWIPs #93 — PubSub protocol — this PR implements Milestone 1 (Direct messaging) of that SWIP.

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 · 2026-04-16T01:59:43Z

+	defer bc.mu.Unlock()
+
+	for _, sub := range bc.subscribers {
+		msg := bc.mode.FormatBroadcast(bc, sub, rawMsg)


I dont think this is needed inside the fanout loop. Why would you reserialise for every subscriber?

the message serialization varies between subscribers e.g. one of them did not get the SOC ID and address for message verification, another subscriber already got it. Then, these metadata will be either attached or not in the message payload.

why would it vary between subscribers? In fact it must be the same, otherwise same msg can be repeated.

…send during create

…and msgType refactor

acud

Thanks @nugaon , once these changes are addressed I will give it another whirl. It seems that for what we are trying to achieve this is too complex. It can be slimmed down significantly

acud · 2026-07-17T16:44:10Z


 	s.Handler = web.ChainHandlers(
 		httpaccess.NewHTTPAccessLogHandler(s.logger, s.tracer, "api access"),
-		handlers.CompressHandler,


how is it that this is needed? we've had websockets running in the API for years now. by what you're saying, they couldn't have worked so far. it would be good to know why this is needed

acud · 2026-07-17T16:45:19Z

+// ResponseWriter, causing "response.Write on hijacked connection" errors.
 func (rr *responseRecorder) Hijack() (net.Conn, *bufio.ReadWriter, error) {
-	return rr.ResponseWriter.(http.Hijacker).Hijack()
+	h, ok := rr.ResponseWriter.(http.Hijacker)


not sure why this is needed. can you please explain why the existing websocket machinery that is used to stream chunks cannot be reused? also you are affecting that machinery with this change.

acud · 2026-07-17T16:47:13Z

+	go func() {
+		pubsub.ListeningWs(ctx, conn, pubsub.WsOptions{PingPeriod: pingPeriod, Cancel: cancel}, logger, mode, isPublisher)
+		cancel()
+		_ = conn.Close()


please use defer statements inside the closure before the ListeningWs call for both cleanup commands. when that call panics, the runtime will do a stack unwind and will execute the defer statements. in your version here, since the calls are made after the potential panic, they never get cleaned up

acud · 2026-07-17T16:52:57Z

+	}
+
+	var topicAddr [32]byte
+	if decoded, err := hex.DecodeString(paths.Topic); err == nil && len(decoded) == swarm.HashSize {


can you please add some comments on what topic could be? there's zero comments on the handler and zero inline comments, which is not so nice to read (i'd rather not have to reach out to the openapi spec or BOS to check what a parameter might mean). ideally you can add a comment in the anonymous type on that field, explaining what it is.
also just to be a little bit more specific. if a parameter is a hex address OR an arbitrary length string, then it is ONLY an arbitrary length string. whether someone, by convention, chooses to use one over the other, it is a bad practice to disambiguate them. i would suggest to have just one type and use it however using conventions. i.e. always hash the topic.

acud · 2026-07-17T16:57:07Z

+
+	peerMultiaddr := r.URL.Query().Get("peer")
+	if peerMultiaddr == "" {
+		jsonhttp.BadRequest(w, "missing peer query param")


query params are usually optional, and rest path parameters are mandatory. if the peer multiaddr is mandatory, it should be in the path params, and you avoid these sort of checks then

acud · 2026-07-17T17:08:43Z

+// verifying them, and returning the payload to forward to the WebSocket.
+// If the subscriber is a Publisher, it also reads from the WebSocket
+// and writes raw messages to the p2p stream.
+func ListeningWs(ctx context.Context, conn *websocket.Conn, options WsOptions, logger log.Logger, mode Mode, isPublisher bool) {


i can't see why anything apart from the api package should leak websocket types. this does not make much sense and i would request that the abstractions handle this requirement properly. just in the same way that we don't leak cobra/viper types outside of the cmd package, in the same way no HTTP types should leak out of the api package. thanks

acud · 2026-07-17T17:14:07Z

+	}
+
+	s.logger.Info("connecting to broker peer", "underlay", underlay)
+	bzzAddr, err := s.p2p.ConnectAllowLight(ctx, []ma.Multiaddr{underlay})


let's please not mix these things. a protocol should not be able to dial to another peer. this is not good news. there is already existing functionality in the api package to dial to peers. any sort of connection orchestration should be done from there by a different client that would know to tell bee to connect to another bee. for me having a protocol that can coerce a node to create connections via other messages (be they websocket or p2p messages) is a serious attack surface. a protocol should be "read message, write message". look at pushsync/pullsync and their relevant orchestration layers pusher/puller respectively. we are doing things with this architecture for a good reason. if this deviation is required we should at least be shown some justification as to why.

acud · 2026-07-17T17:16:33Z

+	defer sc.subsMu.Unlock()
+	for _, ch := range sc.subs {
+		select {
+		case ch <- msg:


one reader stalling delays the rest. and you're holding the mutex all that time

acud · 2026-07-17T17:19:30Z

+}
+
+// writeRaw writes raw bytes to the stream.
+func writeRaw(stream p2p.Stream, data []byte) error {


why can't you just Write once? not sure why the helper is needed. libp2p streams can handle enough data unless you can prove otherwise?

acud · 2026-07-17T17:22:17Z

+	defer sc.subsMu.Unlock()
+	id := sc.nextID
+	sc.nextID++
+	ch := make(chan []byte, 16)


channel buffer size is either 0 or 1 please. this is a recurring pattern i see that in PRs this rule is starting to be broken. this is against go styleguides and against our adopted styleguide.

nugaon force-pushed the feat/pubsub branch from 8cfd40e to 8de8fac Compare April 14, 2026 14:42

zelig assigned nugaon Apr 16, 2026

zelig reviewed Apr 16, 2026

View reviewed changes

nugaon force-pushed the feat/pubsub branch 4 times, most recently from 7011490 to 8539f8c Compare May 15, 2026 13:15

nugaon closed this May 15, 2026

nugaon force-pushed the feat/pubsub branch from 8539f8c to 088c616 Compare May 15, 2026 13:16

nugaon reopened this May 15, 2026

nugaon mentioned this pull request May 21, 2026

BeeJS and collab app aligment, demo setup Apiary-Suite/planning#2

Open

nugaon and others added 19 commits May 28, 2026 15:08

feat: pubsub

21c071b

refactor: mode id enum type

6e38c54

refactor: rename participant to publisher

cbc882d

docs: openapi

6575b7d

fix: eth address instead of public key on api

e5155a1

fix: pubsub header parsing

155ca50

fix: config cli option for broker mode

63a1435

fix: http hijacked write error

d64a383

chore: debugging

b6436c7

fix: libp2p backoff clear

4486a40

fix: http hijack handler

1b8e90c

fix: hijacked websocket error

d6541e4

chore: logs debugging

f7ff741

fix: pingpong with subscriber

5dd81fd

fix: upgrade ws no compress

e1d7dc7

fix: readdeadline on pong

873f9b7

fix(broker): eof return

cd8dbbf

chore: logs

096902b

fix: wrong connection mapping and many refactors

a805913

bosi95 and others added 24 commits May 28, 2026 15:09

fix: span size 8

7aea4c5

fix: swap pubsub ws upgrade and connect order

7c87b9b

fix: revert ws and svc connect order and cancel subscriberConn if pre…

27460ff

…send during create

fix: unregister delete overlay

df93f38

feat: allow light node mode

20e8cfd

fix: remove and create sub conn race conditions

d35f391

fix: multiple ws on the same topic

010f8fd

refactor: remove unnecessary param in runMux

8d87fd5

refactor: move connection related struct defs and methods to service

4f1f5f9

refactor: subtract the pinging service from mode

552f123

docs: correction of size in the span

c9a4301

fix: removing duplicated stream close, leave only cancel

c7a622d

refactor: log out dropping messages when ws buffer is full

ce3cc21

fix: clear subscriberConn in runMux defer to prevent stale reference …

d2b3c7c

…and msgType refactor

fix: msg type in mode

73dff46

fix: remove devmode

594c93d

refactor: remove api header and use only query

0e344ae

refactor: remove pubsub max connections

62540ad

fix: ci linting

1076add

refactor: move pubsub schemas from headers to components section

e0bf3a9

fix: subscriberConn null set concurrency

0d2d78c

test: init

2b91b2c

perf: bmt SIMD hasher (#5381)

da8ccfd

fix: linting

acbface

nugaon force-pushed the feat/pubsub branch 3 times, most recently from c3e2d28 to acbface Compare June 1, 2026 14:00

nugaon mentioned this pull request Jun 4, 2026

feat: soc pubkey #5492

Merged

7 tasks

refactor: msg type numbers

e9f6d45

acud requested changes Jul 17, 2026

View reviewed changes

Uh oh!

Conversation

nugaon commented Apr 14, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

pubsub

Overview

Connection flow

Mode system

GSOC Ephemeral mode (mode 1)

Publisher → Broker (p2p)

Broker → Subscriber (p2p)

WebSocket (both directions)

Multi-WebSocket multiplexer

API

Configuration

Extensibility

Tests

Checklist

Description

Open API Spec Version Changes (if applicable)

Motivation and Context (Optional)

Related Issue (Optional)

Screenshots (if appropriate):

AI Disclosure

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

acud left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

nugaon commented Apr 14, 2026 •

edited

Loading