feat(api): bidirectional chunk stream protocol

openapi/Swarm.yaml

@@ -1,7 +1,7 @@
 openapi: 3.0.3
 
 info:
-  version: 8.2.2
+  version: 9.0.0
   title: Bee API
   description: "API endpoints for interacting with the Swarm network, supporting file operations, messaging, and node management"
 
@@ -330,24 +330,43 @@
 
   "/chunks/stream":
     get:
-      summary: "Stream chunks for upload"
-      description: "Establishes a WebSocket connection for streaming chunks. Each uploaded chunk receives a binary acknowledgment (`0`). Chunks are sent as binary messages. When a tag is specified, chunks are stored locally and uploaded to the network after the stream closes. Without a tag, chunks are directly uploaded to the network as they arrive."
+      summary: "Stream chunk downloads and uploads"
+      description: |
+        Establishes a WebSocket connection over which a client pipelines chunk downloads and uploads. Client and node exchange binary Protobuf messages defined by `pkg/api/pb/chunkstream.proto`, one message per WebSocket binary message. Downloads (`GetRequest`) and uploads (`PutRequest`) can be mixed on the same connection.
+
+        This replaces the earlier upload-only stream, which took raw `[8-byte span][data]` messages; a client still sending those has its connection closed with `1003`.
+
+        - Correlation: requests carry no identifier; the chunk address is the key. Each reply echoes the address of its request, and its type (`GetResponse` or `PutResponse`) tells a download reply from an upload reply for the same chunk. A client must not have two requests of the same type for the same address in flight on one connection.
+        - `GetRequest`: retrieves a chunk by its 32-byte `Address`, with an optional `CacheOption` (`CACHE_DEFAULT = 0`, `CACHE_DISABLE = 1`, `CACHE_ENABLE = 2`). Answered by a `GetResponse` carrying the chunk in `Data` on success. Encrypted references (64 bytes) are not supported: fetch an encrypted chunk by its address and decrypt it on the client.
+        - `PutRequest`: uploads a chunk given its `Address` and `Data`. The `Address` must match the address the node computes from `Data`, otherwise the upload is rejected with `STATUS_BAD_REQUEST`; it also settles whether the data is a CAC or a SOC, so no chunk type is sent. Stamping, in order of precedence: a pre-signed `Stamp` on the request, the request's `BatchId`, then the connection's `Swarm-Postage-Batch-Id`. A pre-signed stamp must be valid for the chunk, including its bucket and index. A connection with `Swarm-Postage-Batch-Id` is pinned to that batch: a `BatchId` or pre-signed `Stamp` from a different batch is rejected with `STATUS_BAD_REQUEST`. The batch is checked for every chunk, so one that expires while the connection is open is no longer used. Answered by a `PutResponse`.
+        - Uploads are always direct: `STATUS_OK` confirms that the chunk was pushed to network peers. There is no deferred mode on the stream, so `Swarm-Tag` / `swarm-tag` is rejected with `400` at the handshake.
+        - `Status`: `STATUS_UNSPECIFIED = 0` (never sent), `STATUS_OK = 1`, `STATUS_NOT_FOUND = 2`, `STATUS_ERROR = 3`, `STATUS_BAD_REQUEST = 4`, `STATUS_BUSY = 5` (queue full — retry with backoff), `STATUS_UNAVAILABLE = 6` (no peer could serve the chunk, because none was usable or every peer was tried; it may or may not exist — retry later, a limited number of times). Failed replies carry a sanitized `Error` message.
+        - Reading replies: clients should read replies while they send requests. A client that stops reading fills the socket, and the node then stops reading its requests until it does.
+        - Frame size limit: 8 KiB maximum per WebSocket message.
+        - Keepalive: the node pings the client periodically. A connection is closed after 15 minutes without a request, a pong or a ping from the client, so a client that is only waiting for replies stays connected as long as it answers pings (browsers do this automatically).
+        - WebSocket Close Codes:
+          - `1000` (Normal Closure): Client-initiated clean disconnect.
+          - `1001` (Going Away): Node is shutting down.
+          - `1003` (Unsupported Data): Non-binary frame, invalid protobuf payload, or a request with no body (which carries no address to reply with).
+          - `1009` (Message Too Big): Message exceeds the 8 KiB size limit.
+          - `1006` (Abnormal Closure): the connection was dropped without a close frame, for example after 15 minutes without a request or a pong.
+        - Fault Tolerance & Recovery: individual chunk errors (e.g. not found, bad request, queue full) are answered with a reply for that address and do not close the stream. `STATUS_BUSY` indicates worker queue saturation; clients should retry with exponential backoff. If the connection drops or is closed, clients should reconnect and re-send every request that received no reply.
       tags:
         - Chunk
       parameters:
-        - $ref: "SwarmCommon.yaml#/components/parameters/SwarmTagParameter"
         - in: query
-          name: swarm-tag
+          name: cache
           schema:
-            $ref: "SwarmCommon.yaml#/components/schemas/Uid"
+            type: boolean
           required: false
-          description: "Associate upload with an existing Tag UID (use when WebSocket client cannot set custom headers)"
+          description: "Enable or disable chunk caching when downloading (default: true). Overridden by `Swarm-Cache` header if present."
         - in: header
           name: swarm-postage-batch-id
           description: "ID of Postage Batch that is used to upload data with. Optional when chunks include pre-signed postage stamps."
           required: false
           schema:
             $ref: "SwarmCommon.yaml#/components/schemas/SwarmAddress"
+        - $ref: "SwarmCommon.yaml#/components/parameters/SwarmCache"
       responses:
         "200":
           description: "Connection established"

pkg/api/api.go

@@ -225,6 +225,10 @@ type Service struct {
 	lightNodes  *lightnode.Container
 	blockTime   time.Duration
 
+	chunkDeliveryWriteDeadline  time.Duration
+	chunkDownloadRequestTimeout time.Duration
+	chunkStreamReadTimeout      time.Duration
+
 	statusSem        *semaphore.Weighted
 	postageSem       *semaphore.Weighted
 	stakingSem       *semaphore.Weighted
@@ -268,6 +272,15 @@ func (s *Service) SetRedistributionAgent(redistributionAgent *storageincentives.
 type Options struct {
 	CORSAllowedOrigins []string
 	WsPingPeriod       time.Duration
+	// ChunkDeliveryWriteDeadline bounds a single response write on
+	// /chunks/stream. Zero selects chunkDeliveryWriteDeadline.
+	ChunkDeliveryWriteDeadline time.Duration
+	// ChunkDownloadRequestTimeout bounds a single chunk retrieval on
+	// /chunks/stream. Zero selects chunkDownloadRequestTimeout.
+	ChunkDownloadRequestTimeout time.Duration
+	// ChunkStreamReadTimeout bounds how long /chunks/stream waits for a
+	// request, a pong or a ping. Zero selects streamReadTimeout.
+	ChunkStreamReadTimeout time.Duration
 }
 
 type ExtraOptions struct {
@@ -382,6 +395,21 @@ func (s *Service) Configure(signer crypto.Signer, tracer *tracing.Tracer, o Opti
 	s.pseudosettle = e.Pseudosettle
 	s.blockTime = e.BlockTime
 
+	s.chunkDeliveryWriteDeadline = o.ChunkDeliveryWriteDeadline
+	if s.chunkDeliveryWriteDeadline <= 0 {
+		s.chunkDeliveryWriteDeadline = chunkDeliveryWriteDeadline
+	}
+
+	s.chunkDownloadRequestTimeout = o.ChunkDownloadRequestTimeout
+	if s.chunkDownloadRequestTimeout <= 0 {
+		s.chunkDownloadRequestTimeout = chunkDownloadRequestTimeout
+	}
+
+	s.chunkStreamReadTimeout = o.ChunkStreamReadTimeout
+	if s.chunkStreamReadTimeout <= 0 {
+		s.chunkStreamReadTimeout = streamReadTimeout
+	}
+
 	s.statusSem = semaphore.NewWeighted(1)
 	s.postageSem = semaphore.NewWeighted(1)
 	s.stakingSem = semaphore.NewWeighted(1)

pkg/api/api_test.go

@@ -90,14 +90,20 @@ func init() {
 }
 
 type testServerOptions struct {
-	Storer             api.Storer
-	Tracer             *tracing.Tracer
-	StateStorer        storage.StateStorer
-	Resolver           resolver.Interface
-	Pss                pss.Interface
-	Gsoc               gsoc.Listener
-	WsPath             string
-	WsPingPeriod       time.Duration
+	Storer                      api.Storer
+	Tracer                      *tracing.Tracer
+	StateStorer                 storage.StateStorer
+	Resolver                    resolver.Interface
+	Pss                         pss.Interface
+	Gsoc                        gsoc.Listener
+	WsPath                      string
+	WsPingPeriod                time.Duration
+	ChunkDeliveryWriteDeadline  time.Duration
+	ChunkDownloadRequestTimeout time.Duration
+	ChunkStreamReadTimeout      time.Duration
+	// Service, when set, receives the api.Service and makes the caller
+	// responsible for closing it, so a test can assert on Close directly.
+	Service            **api.Service
 	Logger             log.Logger
 	PreventRedirect    bool
 	Feeds              feeds.Factory
@@ -227,7 +233,11 @@ func newTestServer(t *testing.T, o testServerOptions) (*http.Client, *websocket.
 	}
 
 	s := api.New(o.PublicKey, o.PSSPublicKey, o.EthereumAddress, []string{o.WhitelistedAddr}, o.Logger, transaction, o.BatchStore, o.BeeMode, !o.ChequebookDisabled, !o.SwapDisabled, backend, o.CORSAllowedOrigins, inmemstore.New())
-	testutil.CleanupCloser(t, s)
+	if o.Service != nil {
+		*o.Service = s
+	} else {
+		testutil.CleanupCloser(t, s)
+	}
 
 	s.SetP2P(o.P2P)
 
@@ -252,8 +262,11 @@ func newTestServer(t *testing.T, o testServerOptions) (*http.Client, *websocket.
 	}
 
 	s.Configure(signer, tracer, api.Options{
-		CORSAllowedOrigins: o.CORSAllowedOrigins,
-		WsPingPeriod:       o.WsPingPeriod,
+		CORSAllowedOrigins:          o.CORSAllowedOrigins,
+		WsPingPeriod:                o.WsPingPeriod,
+		ChunkDeliveryWriteDeadline:  o.ChunkDeliveryWriteDeadline,
+		ChunkDownloadRequestTimeout: o.ChunkDownloadRequestTimeout,
+		ChunkStreamReadTimeout:      o.ChunkStreamReadTimeout,
 	}, extraOpts, 1, erc20APIService)
 
 	s.Mount()