feat: bps protocol draft

cmd/bee/cmd/cmd.go

@@ -50,6 +50,7 @@ const (
 	optionNamePaymentThreshold             = "payment-threshold"
 	optionNamePaymentTolerance             = "payment-tolerance-percent"
 	optionNamePaymentEarly                 = "payment-early-percent"
+	optionNamePubsubCapacity               = "pubsub-capacity"
 	optionNameResolverEndpoints            = "resolver-options"
 	optionNameBootnodeMode                 = "bootnode-mode"
 	optionNameBzzTokenAddress              = "bzz-token-address"
@@ -347,6 +348,7 @@ func (c *command) setAllFlags(cmd *cobra.Command) {
 	cmd.Flags().String(optionNamePaymentThreshold, "13500000", "threshold in BZZ where you expect to get paid from your peers")
 	cmd.Flags().Int64(optionNamePaymentTolerance, 25, "excess debt above payment threshold in percentages where you disconnect from your peer")
 	cmd.Flags().Int64(optionNamePaymentEarly, 50, "percentage below the peers payment threshold when we initiate settlement")
+	cmd.Flags().Int(optionNamePubsubCapacity, 32, "per-topic connection capacity this node offers as a pubsub broker")
 	cmd.Flags().StringSlice(optionNameResolverEndpoints, []string{}, "ENS compatible API endpoint for a TLD and with contract address, can be repeated, format [tld:][contract-addr@]url")
 	cmd.Flags().Bool(optionNameBootnodeMode, false, "cause the node to always accept incoming connections")
 	cmd.Flags().String(optionNameBlockchainRpcEndpoint, "", "rpc blockchain endpoint")

cmd/bee/cmd/start.go

@@ -308,6 +308,7 @@ func buildBeeNode(ctx context.Context, c *command, cmd *cobra.Command, logger lo
 		BlockSyncInterval:             c.config.GetUint64(optionNameBlockSyncInterval),
 		BootnodeMode:                  bootNode,
 		Bootnodes:                     networkConfig.bootNodes,
+		BpsCapacity:                   c.config.GetInt(optionNamePubsubCapacity),
 		CacheCapacity:                 c.config.GetUint64(optionNameCacheCapacity),
 		AutoTLSCAEndpoint:             c.config.GetString(optionAutoTLSCAEndpoint),
 		ChainID:                       networkConfig.chainID,

openapi/Swarm.yaml

@@ -1,7 +1,7 @@
 openapi: 3.0.3
 
 info:
-  version: 8.1.1
+  version: 8.2.0
   title: Bee API
   description: "API endpoints for interacting with the Swarm network, supporting file operations, messaging, and node management"
 
@@ -979,6 +979,142 @@ paths:
         default:
           description: Default response
 
+  "/pubsub":
+    get:
+      summary: List active pubsub topics
+      tags:
+        - PubSub
+      responses:
+        "200":
+          description: List of pubsub topics this node currently participates in
+          content:
+            application/json:
+              schema:
+                type: array
+                items:
+                  $ref: "SwarmCommon.yaml#/components/schemas/BpsTopicStatus"
+        "501":
+          description: Pubsub is not enabled on this node
+        default:
+          description: Default response
+
+  "/pubsub/{topic}":
+    get:
+      summary: Open or join a pubsub topic (WebSocket)
+      description: >
+        Upgrades the connection to a WebSocket attached to one pubsub topic. Presence of any
+        cohort query parameter (binding, publishers, admin, publisher-list, closed, history)
+        opens the cohort for that topic; their absence subscribes to a topic that is already
+        open. Presence of the owner query parameter makes the session a publisher.
+
+
+        Signing of publish frames is performed client-side; the node holds no publisher keys.
+        For an ANCHOR-bound topic, inbound publisher frames are binary
+        `sig(65) || span(8) || payload`, and the Single Owner Chunk id is the topic itself. For a
+        FEED_TOPIC-bound topic, inbound publisher frames are binary
+        `index(8, big-endian) || sig(65) || span(8) || payload`, and the id is
+        keccak256(topic || index). There is no `index` query parameter: the feed index rides on
+        every publish frame.
+
+
+        Outbound (subscriber) messages are payload-only binary frames by default, or JSON text
+        frames with hex-encoded fields when Swarm-Soc-Fields names more than payload alone. Under
+        an explicit publisher regime, a subscriber has no way to independently verify a
+        publisher's cohort spec ahead of the broker's own enforcement; treat delivered messages
+        accordingly until the broker vouches for the cohort.
+
+
+        A second local WebSocket client attaching to a topic that is already open on this node is
+        muxed onto the existing p2p session rather than admitted as a new one: closed-cohort and
+        capacity checks are enforced by the broker per p2p session, not per local WS client.
+      tags:
+        - PubSub
+      parameters:
+        - in: path
+          name: topic
+          schema:
+            type: string
+          required: true
+          description: >
+            Topic name: either a 64-hex-character string naming a 32-byte topic directly, or any
+            other string, which is hashed (keccak256) into a topic as a mnemonic.
+        - in: query
+          name: peer
+          schema:
+            type: string
+          required: true
+          description: Underlay multiaddr of the broker peer for this topic.
+        - in: query
+          name: binding
+          schema:
+            type: string
+            enum:
+              - anchor
+              - feed
+          required: false
+          description: How a publish frame's Single Owner Chunk id is derived. Presence of this or any other cohort parameter opens the cohort.
+        - in: query
+          name: publishers
+          schema:
+            type: string
+            enum:
+              - single
+              - list
+          required: false
+          description: Whether the cohort has one publisher (single) or an explicit publisher list (list).
+        - in: query
+          name: admin
+          schema:
+            $ref: "SwarmCommon.yaml#/components/schemas/EthereumAddress"
+          required: false
+          description: Ethereum address of the cohort's admin.
+        - in: query
+          name: publisher-list
+          schema:
+            type: string
+          required: false
+          description: Comma-separated list of 20-byte hex Ethereum addresses allowed to publish, used with `publishers=list`.
+        - in: query
+          name: closed
+          schema:
+            type: boolean
+          required: false
+          description: Whether the cohort is closed to publishers not on the publisher list.
+        - in: query
+          name: history
+          schema:
+            type: boolean
+          required: false
+          description: "Whether the cohort should deliver history to new subscribers. Not supported by this node; a cohort spec naming `history: true` is refused."
+        - in: query
+          name: owner
+          schema:
+            $ref: "SwarmCommon.yaml#/components/schemas/EthereumAddress"
+          required: false
+          description: Ethereum address of the publisher signing outbound frames. Presence of this parameter makes the session a publisher.
+        - $ref: "SwarmCommon.yaml#/components/parameters/SwarmKeepAlive"
+        - $ref: "SwarmCommon.yaml#/components/parameters/SwarmSocFields"
+        - $ref: "SwarmCommon.yaml#/components/parameters/SwarmCacheWrappedChunk"
+      responses:
+        "101":
+          description: Switching protocols; a pubsub WebSocket session for the topic is established
+        "400":
+          $ref: "SwarmCommon.yaml#/components/responses/400"
+        "403":
+          description: The request was rejected — the cohort is closed, or the caller is not a publisher of the cohort
+        "404":
+          description: Unknown topic
+        "409":
+          description: The requested cohort specification conflicts with the cohort's live specification
+        "500":
+          $ref: "SwarmCommon.yaml#/components/responses/500"
+        "501":
+          description: Pubsub is not enabled on this node
+        "503":
+          description: The broker is at capacity for this topic
+        default:
+          description: Default response
+
   "/soc/{owner}/{id}":
     post:
       summary: Upload a Single Owner Chunk

openapi/SwarmCommon.yaml

@@ -1075,6 +1075,56 @@ components:
       properties:
         transactionHash:
           $ref: "#/components/schemas/TransactionHash"
+
+    BpsCohortSpec:
+      type: object
+      description: The cohort specification of a pubsub topic, when known.
+      properties:
+        binding:
+          type: string
+          enum:
+            - anchor
+            - feed
+          description: How a publish frame's Single Owner Chunk id is derived. "anchor" uses the topic itself as the id; "feed" derives the id from the topic and the 8-byte big-endian index carried on every publish frame.
+        publishers:
+          type: string
+          enum:
+            - single
+            - list
+          description: Whether the cohort has one publisher (single) or an explicit publisher list (list).
+        admin:
+          type: string
+          description: Hex-encoded (no 0x prefix) 20-byte Ethereum address of the cohort's admin, or empty when unset.
+        publisherList:
+          type: array
+          items:
+            type: string
+            description: Hex-encoded (no 0x prefix) 20-byte Ethereum address.
+        closed:
+          type: boolean
+          description: Whether the cohort is closed to publishers not on the publisher list.
+        history:
+          type: boolean
+          description: Whether the cohort delivers history to new subscribers. Not supported by this node.
+
+    BpsTopicStatus:
+      type: object
+      description: The status of one pubsub topic this node participates in.
+      properties:
+        topic:
+          $ref: "#/components/schemas/SwarmAddress"
+        role:
+          type: string
+          enum:
+            - broker
+            - client
+          description: Whether this node brokers the topic or is a client of it.
+        peers:
+          type: integer
+          description: Number of peers currently attached to the topic on this node.
+        cohort:
+          $ref: "#/components/schemas/BpsCohortSpec"
+
   headers:
     SwarmTag:
       description: "Tag UID"
@@ -1331,6 +1381,38 @@ components:
       required: false
       description: "ACT history Unix timestamp"
 
+    SwarmKeepAlive:
+      in: header
+      name: Swarm-Keep-Alive
+      schema:
+        type: integer
+        format: int64
+        default: 60
+      required: false
+      description: "Ping period, in seconds, of the pubsub websocket session. Must be a positive integer. Default: 60"
+
+    SwarmSocFields:
+      in: header
+      name: Swarm-Soc-Fields
+      schema:
+        type: string
+      required: false
+      description: >
+        Comma-separated list of Single Owner Chunk fields to include in outbound pubsub websocket
+        messages: address, recoveredPubKey, identifier, signature, wrappedAddress, span, payload.
+        Default: payload. When the selection is payload only, messages are sent as binary frames
+        containing the raw payload; any other selection is sent as a JSON text frame with
+        hex-encoded values for exactly the requested fields.
+
+    SwarmCacheWrappedChunk:
+      in: header
+      name: Swarm-Cache-Wrapped-Chunk
+      schema:
+        type: boolean
+        default: "false"
+      required: false
+      description: "Indicates whether each message's wrapped chunk should be stored in the local cache. Default: false"
+
   responses:
     "200":
       description: Success

pkg/api/api.go

@@ -27,6 +27,7 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethersphere/bee/v2/pkg/accesscontrol"
 	"github.com/ethersphere/bee/v2/pkg/accounting"
+	"github.com/ethersphere/bee/v2/pkg/bps"
 	"github.com/ethersphere/bee/v2/pkg/crypto"
 	"github.com/ethersphere/bee/v2/pkg/feeds"
 	"github.com/ethersphere/bee/v2/pkg/file/pipeline"
@@ -96,6 +97,9 @@ const (
 	SwarmActTimestampHeader           = "Swarm-Act-Timestamp"
 	SwarmActPublisherHeader           = "Swarm-Act-Publisher"
 	SwarmActHistoryAddressHeader      = "Swarm-Act-History-Address"
+	SwarmKeepAliveHeader              = "Swarm-Keep-Alive"
+	SwarmSocFieldsHeader              = "Swarm-Soc-Fields"
+	SwarmCacheWrappedChunkHeader      = "Swarm-Cache-Wrapped-Chunk"
 
 	ImmutableHeader = "Immutable"
 	GasPriceHeader  = "Gas-Price"
@@ -149,6 +153,12 @@ type Storer interface {
 	storer.NeighborhoodStats
 }
 
+// BpsBridge is the surface of pkg/bps the API needs.
+type BpsBridge interface {
+	Attach(ctx context.Context, o bps.AttachOptions) (bps.Attachment, error)
+	Status() []bps.TopicStatus
+}
+
 type PinIntegrity interface {
 	Check(ctx context.Context, logger log.Logger, pin string, out chan storer.PinStat)
 }
@@ -158,6 +168,7 @@ type Service struct {
 	resolver        resolver.Interface
 	pss             pss.Interface
 	gsoc            gsoc.Listener
+	bps             BpsBridge
 	steward         steward.Interface
 	logger          log.Logger
 	loggerV1        log.Logger
@@ -264,6 +275,7 @@ type ExtraOptions struct {
 	Resolver        resolver.Interface
 	Pss             pss.Interface
 	Gsoc            gsoc.Listener
+	Bps             BpsBridge
 	FeedFactory     feeds.Factory
 	Post            postage.Service
 	AccessControl   accesscontrol.Controller
@@ -345,6 +357,7 @@ func (s *Service) Configure(signer crypto.Signer, tracer *tracing.Tracer, o Opti
 	s.resolver = e.Resolver
 	s.pss = e.Pss
 	s.gsoc = e.Gsoc
+	s.bps = e.Bps
 	s.feedFactory = e.FeedFactory
 	s.post = e.Post
 	s.accesscontrol = e.AccessControl

pkg/api/api_test.go

@@ -95,6 +95,7 @@ type testServerOptions struct {
 	Resolver           resolver.Interface
 	Pss                pss.Interface
 	Gsoc               gsoc.Listener
+	Bps                api.BpsBridge
 	WsPath             string
 	WsPingPeriod       time.Duration
 	Logger             log.Logger
@@ -202,6 +203,7 @@ func newTestServer(t *testing.T, o testServerOptions) (*http.Client, *websocket.
 		Resolver:        o.Resolver,
 		Pss:             o.Pss,
 		Gsoc:            o.Gsoc,
+		Bps:             o.Bps,
 		FeedFactory:     o.Feeds,
 		Post:            o.Post,
 		AccessControl:   o.AccessControl,