feat(bps): explicit multi-publisher mode (SWIP-87)

openapi/Swarm.yaml

@@ -993,14 +993,20 @@ paths:
     get:
       summary: Subscribe to a BPS cohort
       description: |
-        WebSocket. The node joins the BPS-lite (SWIP-74) cohort `{topic, FEED_TOPIC, owner}`
-        at `broker` as `identity` (`broker` must be a full node started with `bps-broker`) and sends every update as a binary frame
-        `challenge (32) | index (8, big-endian) | soc`, where `soc` is the full single-owner
-        chunk (`id | signature | span | payload`) signed by `owner` with
-        `id = keccak256(keccak256(topic | challenge) | index)`. Updates that do not verify,
-        or whose index is below the subscriber's cursor, are dropped.
+        WebSocket. The node joins the BPS cohort `{topic, FEED_TOPIC, owner}` (SWIP-74, with
+        the multi-publisher mode of SWIP-87) at `broker` as `identity` and sends every
+        verified delivery as a binary frame
+        `kind (1) | owner (20) | challenge (32) | index (8, big-endian) | soc`, where `soc` is
+        the full single-owner chunk (`id | signature | span | payload`) signed by `owner` with
+        `id = keccak256(sessionTopic | index)`. `kind` is 1 (DATA): an update, by the cohort
+        owner or by an address on its latest roster; 4 (ROSTER): the owner's publisher set,
+        payload a `Roster` protobuf of 20-byte addresses; or 3 (EOS): the owner ended the
+        channel, after which the socket closes with 4004. `sessionTopic` is
+        `keccak256(topic | challenge)` for DATA and `keccak256("bps-service:v1" | kind | topic | challenge)`
+        otherwise. Deliveries that do not verify, by an owner the roster does not admit, or
+        below the subscriber's cursor are dropped.
         The endpoint is read-only: any client frame closes the socket with 4002.
-        Close codes: 4002 invalid message, 4003 broker stream closed.
+        Close codes: 4002 invalid message, 4003 broker stream closed, 4004 end of stream.
       tags:
         - BPS
       parameters:
@@ -1033,7 +1039,7 @@ paths:
           schema:
             type: integer
           required: false
-          description: Lowest update index to deliver. Defaults to 0.
+          description: Lowest index of the owner's updates to deliver. Defaults to 0.
       responses:
         "200":
           description: Establishes a WebSocket subscription for incoming broadcasts
@@ -1048,25 +1054,42 @@ paths:
 
   "/bps/publish/{owner}/{topic}":
     get:
-      summary: Publish to a BPS cohort as its principal
+      summary: Publish to a BPS cohort as its owner or as a rostered seat
       description: |
-        WebSocket. The node joins the BPS-lite (SWIP-74) cohort `{topic, FEED_TOPIC, owner}`
-        at `broker` declaring `owner` as its address.
-
-        1. Server sends `{"type":"challenge","challenge":"<hex>","broker":"<hex>","dataTopic":"<hex>","authTopic":"<hex>"}`.
-           `dataTopic` is `keccak256(topic | challenge)`, the session feed for updates;
-           `authTopic` is `keccak256("bps-service:v1" | 0x02 | topic | challenge)`.
-        2. Client sends binary frames `kind (1) | index (8, big-endian) | soc`, where `kind` is
-           1 (DATA) or 2 (AUTH) and `soc` is a single-owner chunk signed by `owner` with
-           `id = keccak256(sessionTopic | index)`. The first frame must be the AUTH — the empty
-           chunk at index 0 of `authTopic`, which authenticates the stream — and must arrive within
-           30 seconds; sent again later it is a heartbeat. DATA frames are updates of
-           `dataTopic`. Indexes must not be reused across sessions; frames below the broker's
-           cursor are dropped silently.
-
-        Close codes: 4001 invalid soc (an AUTH at another index or with a payload included),
-        4002 invalid message (wrong frame type, any frame before the AUTH, unknown kind, auth
-        timeout), 4003 broker stream closed.
+        WebSocket. The node joins the BPS cohort `{topic, FEED_TOPIC, owner}` (SWIP-74, with
+        the multi-publisher mode of SWIP-87) at `broker` declaring `identity` — the owner by
+        default, or an address the owner's roster names or will name (a seat). A seat's
+        stream receives from the start, like a subscriber's; once a roster naming the seat
+        has reached it, the server sends `{"type":"granted"}` and the seat may send its
+        AUTH, which upgrades the stream in place. A seat that sends anything before the
+        grant is closed with 4002. A seat whose address a later roster drops finds the
+        socket closed after that roster (4003).
+
+        1. Server sends `{"type":"challenge","challenge":"<hex>","broker":"<hex>","dataTopic":"<hex>","authTopic":"<hex>"}`,
+           and to the owner also `"rosterTopic"` and `"eosTopic"`.
+           `dataTopic` is `keccak256(topic | challenge)`, the session feed for updates; the
+           others are `keccak256("bps-service:v1" | kind | topic | challenge)` with kind 0x02
+           (AUTH), 0x04 (ROSTER), 0x03 (EOS).
+        2. Client sends binary frames `kind (1) | index (8, big-endian) | soc`, where `soc` is
+           a single-owner chunk with `id = keccak256(sessionTopic | index)`, signed by
+           `identity` for kind 1 (DATA) and 2 (AUTH) and by `owner` for kind 4 (ROSTER) and 3
+           (EOS), which only the owner may send. The first frame must be the AUTH — the empty
+           chunk at index 0 of `authTopic`, which authenticates the stream — and must arrive
+           within 25 seconds — of the connection for the owner, of the grant for a seat;
+           sent again later it is a heartbeat. DATA frames are updates of
+           `dataTopic`. A ROSTER carries a `Roster` protobuf naming every address that may
+           publish besides the owner, full state at each index; an EOS — the empty chunk at
+           index 0 of `eosTopic` — ends the channel. An owner that has a roster MUST publish
+           it again, at its next roster index, right after its AUTH: the broker may hold
+           none, after an inactivity reclaim or a restart. Indexes must not be reused across
+           sessions; frames below the broker's cursor are dropped silently.
+        3. Server sends what the other publishers publish, and the owner's rosters — to
+           the owner too — as binary frames in the subscribe endpoint's format.
+
+        Close codes: 4001 invalid soc (an AUTH or EOS at another index or with a payload
+        included), 4002 invalid message (wrong frame type, any frame before the AUTH, unknown
+        kind, a ROSTER or EOS from a seat, a roster that does not decode, a frame from a seat
+        before its grant, auth timeout), 4003 broker stream closed, 4004 end of stream.
       tags:
         - BPS
       parameters:
@@ -1088,6 +1111,12 @@ paths:
             $ref: "SwarmCommon.yaml#/components/schemas/SwarmAddress"
           required: true
           description: Overlay address of the broker node. Must not be this node.
+        - in: query
+          name: identity
+          schema:
+            $ref: "SwarmCommon.yaml#/components/schemas/EthereumAddress"
+          required: false
+          description: The address the stream publishes as. Defaults to the owner; any other may publish once the owner's roster names it.
       responses:
         "200":
           description: Establishes a WebSocket subscription for publishing broadcasts

pkg/api/bps.go

@@ -28,17 +28,21 @@
 	bpsCloseInvalidSOC     = 4001 // a chunk that does not validate under the session challenge
 	bpsCloseInvalidMessage = 4002 // frame violates the endpoint protocol
 	bpsCloseBrokerGone     = 4003 // the p2p stream to the broker ended
+	bpsCloseEndOfStream    = 4004 // the owner ended the channel with an EOS
 
 	bpsMaxCloseReason = 123   // websocket control frame payload limit minus the code
 	bpsFrameHeader    = 1 + 8 // publish frame header: kind | index (big-endian)
 	bpsMaxSOCSize     = swarm.HashSize + swarm.SocSignatureSize + swarm.SpanSize + swarm.ChunkSize
 	bpsMaxFrameSize   = bpsFrameHeader + bpsMaxSOCSize
+	// delivery frame header: kind | owner | challenge | index (big-endian)
+	bpsDeliveryHeader = 1 + common.AddressLength + bps.ChallengeSize + 8
 )
 
-// bpsAuthTimeout bounds how long a publisher may take to send its AUTH. It is
-// counted from before the join, so that it runs out before the broker's auth
-// deadline does and the broker never blocklists this node for a slow dApp.
-var bpsAuthTimeout = 30 * time.Second
+// bpsAuthTimeout bounds how long a publisher may take to send its AUTH. For the
+// owner it is counted from before the join and is shorter than the broker's
+// auth deadline, so that the broker never blocklists this node for a slow
+// dApp; for a seat it is counted from the grant.
+var bpsAuthTimeout = 25 * time.Second
 
 var errBPSBrokerGone = errors.New("broker stream closed")
 
@@ -50,12 +54,41 @@
 	cursor   uint64
 }
 
+// bpsGrantedMessage tells a seat's dApp that a roster naming its identity has
+// reached the stream: it may now send its AUTH.
+type bpsGrantedMessage struct {
+	Type string `json:"type"`
+}
+
 type bpsChallengeMessage struct {
-	Type      string `json:"type"`
-	Challenge string `json:"challenge"`
-	Broker    string `json:"broker"`
-	DataTopic string `json:"dataTopic"`
-	AuthTopic string `json:"authTopic"`
+	Type        string `json:"type"`
+	Challenge   string `json:"challenge"`
+	Broker      string `json:"broker"`
+	DataTopic   string `json:"dataTopic"`
+	AuthTopic   string `json:"authTopic"`
+	RosterTopic string `json:"rosterTopic,omitempty"` // the owner's only
+	EOSTopic    string `json:"eosTopic,omitempty"`    // the owner's only
+}
+
+// bpsDeliveryFrame encodes a delivery for the dApp: kind | owner | challenge |
+// index | soc — what it needs to re-verify the chunk, whoever published it.
+func bpsDeliveryFrame(msg bps.Message) []byte {
+	data := make([]byte, 0, bpsDeliveryHeader+len(msg.SOC))
+	data = append(data, byte(msg.Kind))
+	data = append(data, msg.Owner...)
+	data = append(data, msg.Challenge...)
+	data = binary.BigEndian.AppendUint64(data, msg.Index)
+	return append(data, msg.SOC...)
+}
+
+// bpsSessionEnded closes the socket for a session whose stream ended: with
+// 4004 after an EOS, 4003 otherwise.
+func (s *Service) bpsSessionEnded(conn *websocket.Conn, sess bps.Session) {
+	if err := sess.Err(); errors.Is(err, bps.ErrEndOfStream) {
+		s.bpsClose(conn, bpsCloseEndOfStream, "end of stream")
+		return
+	}
+	s.bpsClose(conn, bpsCloseBrokerGone, bpsErrReason(sess.Err()))
 }
 
 func (s *Service) bpsSubscribeWsHandler(w http.ResponseWriter, r *http.Request) {
@@ -88,6 +121,10 @@
 	if !ok {
 		return
 	}
+	// the owner publishes by default; a seat names its own address
+	if len(req.identity) == 0 {
+		req.identity = req.owner.Bytes()
+	}
 	conn, ok := s.bpsUpgrade(w, r, logger)
 	if !ok {
 		return
@@ -244,7 +281,7 @@
 		Broker:   req.broker,
 		Spec:     bps.CohortSpec{Topic: req.topic, Principal: req.owner.Bytes()},
 		Identity: req.identity,
-		Cursor:   req.cursor,
+		Position: bps.Position{Cursors: map[string]uint64{string(req.owner.Bytes()): req.cursor}},
 	})
 	if err != nil {
 		logger.Debug("join failed", "broker", req.broker, "error", err)
@@ -259,23 +296,18 @@
 		select {
 		case msg, ok := <-sess.Messages():
 			if !ok {
-				s.bpsClose(conn, bpsCloseBrokerGone, bpsErrReason(sess.Err()))
+				s.bpsSessionEnded(conn, sess)
 				return
 			}
-			// challenge | index | soc: what the client needs to re-verify the chunk
-			data := make([]byte, 0, bps.ChallengeSize+8+len(msg.SOC))
-			data = append(data, msg.Challenge...)
-			data = binary.BigEndian.AppendUint64(data, msg.Index)
-			data = append(data, msg.SOC...)
-			if err := s.bpsWrite(conn, websocket.BinaryMessage, data); err != nil {
+			if err := s.bpsWrite(conn, websocket.BinaryMessage, bpsDeliveryFrame(msg)); err != nil {
 				logger.Debug("write broadcast failed", "error", err)
 				return
 			}
 		case <-frames:
 			s.bpsClose(conn, bpsCloseInvalidMessage, "subscribe endpoint is read-only")
 			return
 		case <-sess.Done():
-			s.bpsClose(conn, bpsCloseBrokerGone, bpsErrReason(sess.Err()))
+			s.bpsSessionEnded(conn, sess)
 			return
 		case <-gone:
 			return
@@ -304,14 +336,29 @@
 	if !authenticated && kind != bps.KindAuth {
 		return 0, 0, nil, bpsCloseInvalidMessage, "expected auth"
 	}
-	if kind != bps.KindData && kind != bps.KindAuth {
+	owner := req.identity
+	switch kind {
+	case bps.KindData, bps.KindAuth:
+	case bps.KindRoster, bps.KindEOS:
+		// the service feeds are the owner's alone
+		if !bytes.Equal(req.identity, req.owner.Bytes()) {
+			return 0, 0, nil, bpsCloseInvalidMessage, "roster and eos are the owner's"
+		}
+		owner = req.owner.Bytes()
+	default:
 		return 0, 0, nil, bpsCloseInvalidMessage, "unknown kind"
 	}
 	index := binary.BigEndian.Uint64(f.data[1:bpsFrameHeader])
 	chunk := f.data[bpsFrameHeader:]
-	if err := bps.Verify(kind, challenge, index, chunk, req.topic, req.owner.Bytes()); err != nil {
+	if err := bps.Verify(kind, challenge, index, chunk, req.topic, owner); err != nil {
 		return 0, 0, nil, bpsCloseInvalidSOC, "invalid soc"
 	}
+	// the broker drops a roster that does not decode without a word
+	if kind == bps.KindRoster {
+		if _, err := bps.ParseRoster(chunk); err != nil {
+			return 0, 0, nil, bpsCloseInvalidMessage, "invalid roster"
+		}
+	}
 	return kind, index, chunk, 0, ""
 }
 
@@ -338,11 +385,15 @@
 	authTimer := time.NewTimer(bpsAuthTimeout)
 	defer authTimer.Stop()
 
-	// the publisher declares the principal as its identity and authenticates with its AUTH
+	// the publisher declares its identity — the owner's, or a seat the owner's
+	// roster names — and authenticates with its AUTH: the owner at once, a
+	// seat once a roster naming it has reached the stream, which upgrades it
+	// in place
+	owner := bytes.Equal(req.identity, req.owner.Bytes())
 	sess, err := s.bps.Join(ctx, bps.JoinRequest{
 		Broker:   req.broker,
 		Spec:     bps.CohortSpec{Topic: req.topic, Principal: req.owner.Bytes()},
-		Identity: req.owner.Bytes(),
+		Identity: req.identity,
 	})
 	if err != nil {
 		logger.Debug("join failed", "broker", req.broker, "error", err)
@@ -351,6 +402,14 @@
 	}
 	defer sess.Close()
 
+	// a seat listens until it is granted; its auth deadline starts then
+	granted := owner
+	var grantedC <-chan struct{}
+	if !owner {
+		authTimer.Stop()
+		grantedC = sess.Granted()
+	}
+
 	challenge := sess.Challenge()
 	if len(challenge) != bps.ChallengeSize {
 		logger.Debug("invalid challenge length", "broker", req.broker, "length", len(challenge))
@@ -359,7 +418,7 @@
 	}
 	dataTopic, err := bps.SessionTopic(bps.KindData, req.topic, challenge)
 	if err != nil {
 		s.bpsClose(conn, bpsCloseBrokerGone, "session topic")
 		return
 	}
 	authTopic, err := bps.SessionTopic(bps.KindAuth, req.topic, challenge)
@@ -367,13 +426,27 @@
 		s.bpsClose(conn, bpsCloseBrokerGone, "session topic")
 		return
 	}
-	if err := s.bpsWriteJSON(conn, bpsChallengeMessage{
+	msg := bpsChallengeMessage{
 		Type:      "challenge",
 		Challenge: hex.EncodeToString(challenge),
 		Broker:    req.broker.String(),
 		DataTopic: hex.EncodeToString(dataTopic),
 		AuthTopic: hex.EncodeToString(authTopic),
-	}); err != nil {
+	}
+	if owner {
+		rosterTopic, err := bps.SessionTopic(bps.KindRoster, req.topic, challenge)
+		if err != nil {
+			s.bpsClose(conn, bpsCloseBrokerGone, "session topic")
+			return
+		}
+		eosTopic, err := bps.SessionTopic(bps.KindEOS, req.topic, challenge)
+		if err != nil {
+			s.bpsClose(conn, bpsCloseBrokerGone, "session topic")
+			return
+		}
+		msg.RosterTopic, msg.EOSTopic = hex.EncodeToString(rosterTopic), hex.EncodeToString(eosTopic)
+	}
+	if err := s.bpsWriteJSON(conn, msg); err != nil {
 		logger.Debug("write challenge failed", "error", err)
 		return
 	}
@@ -383,7 +456,19 @@
 
 	for {
 		select {
+		case <-grantedC:
+			grantedC, granted = nil, true
+			authTimer.Reset(bpsAuthTimeout)
+			if err := s.bpsWriteJSON(conn, bpsGrantedMessage{Type: "granted"}); err != nil {
+				logger.Debug("write granted failed", "error", err)
+				return
+			}
 		case f := <-frames:
+			// a seat sends nothing before a roster names it
+			if !granted {
+				s.bpsClose(conn, bpsCloseInvalidMessage, "not granted")
+				return
+			}
 			kind, index, chunk, code, reason := bpsParseFrame(f, req, challenge, authenticated)
 			if code != 0 {
 				s.bpsClose(conn, code, reason)
@@ -397,13 +482,22 @@
 				authenticated = true
 				authTimer.Stop()
 			}
+		case msg, ok := <-sess.Messages():
+			if !ok {
+				s.bpsSessionEnded(conn, sess)
+				return
+			}
+			if err := s.bpsWrite(conn, websocket.BinaryMessage, bpsDeliveryFrame(msg)); err != nil {
+				logger.Debug("write broadcast failed", "error", err)
+				return
+			}
 		case <-authTimer.C:
 			if !authenticated {
 				s.bpsClose(conn, bpsCloseInvalidMessage, "auth timeout")
 				return
 			}
 		case <-sess.Done():
-			s.bpsClose(conn, bpsCloseBrokerGone, bpsErrReason(sess.Err()))
+			s.bpsSessionEnded(conn, sess)
 			return
 		case <-gone:
 			return