feat: download mode for the chunk stream endpoint - #5618
martinconic wants to merge 1 commit into
Conversation
| s.metrics.ChunkStreamOpenConnections.WithLabelValues("download").Inc() | ||
| defer s.metrics.ChunkStreamOpenConnections.WithLabelValues("download").Dec() | ||
|
|
||
| ctx, cancel := context.WithCancel(context.Background()) |
There was a problem hiding this comment.
you probably also want to have another goroutine that selects on s.quit. if the node gets shut down, this context doesn't cancel. ideally you cancel the context once if there's a shutdown or if there's an error, then you save selecting on s.quit later on in the method
| return | ||
| } | ||
|
|
||
| if len(msg) < 1+swarm.HashSize || (len(msg)-1)%swarm.HashSize != 0 { |
There was a problem hiding this comment.
some comment as to why is this off by one needed?
| // because ReadMessage allocates a fresh buffer per message; a pooled or | ||
| // reused read buffer would corrupt addresses across concurrent workers. | ||
| addrs := make([]swarm.Address, 0, batchCount) | ||
| for i := 0; i < len(payload); i += swarm.HashSize { |
There was a problem hiding this comment.
what about if there's an encrypted chunk address here? won't work no..?
| func (s *Service) fetchAndSendChunk( | ||
| streamCtx context.Context, | ||
| logger log.Logger, | ||
| loggerV1 log.Logger, |
| s.metrics.ChunkStreamDeliveryCount.WithLabelValues("success").Inc() | ||
|
|
||
| chunkData := chunk.Data() | ||
| resp := make([]byte, 1+swarm.HashSize+len(chunkData)) |
There was a problem hiding this comment.
not sure about this custom serialization by hand thing... it is just specced out in the openapi spec and assumed that implementers should implement it by hand. it is fragile and breakable. why not use some sort of standard serialization format to both decode the request and encode the response?
aloknerurkar
left a comment
aloknerurkar
left a comment
There was a problem hiding this comment.
Since we are making breaking changes and also building on the serialization thread — I'd suggest going further than reworking the download framing: use one connection for both directions, with request/response semantics and a request id.
Two things this fixes beyond tidiness:
-
Upload is round-trip bound. The upload loop is strictly sequential — read → put → ack → read, one chunk in flight. Each chunk costs a full RTT, so on a 50ms link that's ~20 chunks/sec regardless of bandwidth. Download already has a 16-worker pool; upload has none. stamper.Stamp already takes issuer.mtx (pkg/postage/stamper.go:43), so concurrent stamping is safe — the ceiling is the protocol, not the storage layer.
-
Neither direction is recoverable. Download replies carry no id, and the upload ack is successWsMsg = []byte{} — an empty frame with no address at all. Clients correlate purely by ordering. On any error both paths call sendErrorClose and drop the connection, so a client that had N requests outstanding cannot tell which completed. For bulk upload that means restarting from zero.
A shared envelope — [type][8-byte request-id][payload] for requests, [type][8-byte request-id][status][payload] for responses — would give:
- one read loop feeding one typed job channel, with a worker pool serving both Get and Put (removes the duplicated read/deadline/close-handler logic between the two handlers)
- pipelined uploads instead of one-at-a-time
- per-request error status instead of connection teardown, so a single bad chunk no longer kills the stream
- correlation for recovery after an error close
I'd keep this as raw binary rather than JSON-RPC or protobuf. JSON-RPC costs more on the wire with base64 and ~15x encode/decode — self-defeating for an endpoint that exists to cut per-chunk overhead. Protobuf is bee's p2p convention but has never appeared in pkg/api; requiring a schema compiler would be a new burden on bee-js. The binary envelope stays a four-line DataView parse in the browser with no dependency.
The upload stream endpoint was not written with care. I was going through the code and there is a lot of scope to cleanup. Multiple putters defined, deferred/direct upload semantics, stamped and unstamped chunks etc. I feel that this would make it much more usable for clients and also close to what elad mentioned about having grpc style chunk get/put API.
| defer func() { | ||
| cancel() | ||
| close(jobs) | ||
| wg.Wait() |
There was a problem hiding this comment.
A worker blocked in conn.WriteMessage is not released by cancel but only by the WriteDeadline or socket closing. Since the defer ordering is such that conn.Close is called after wg.Wait, teardown waits out chunkDeliveryWriteDeadline which is 5 mins. api.Close has a shorter timeline which is why we will exit with open websockets.
Suggested fix: Collapse into 1 defer so that socket is closed before wg.Wait.
| // connection is closed out from under it. newTestServer closes the api.Service | ||
| // during cleanup and fails the test if Close reports open websockets, which is | ||
| // where this test's assertion lives. | ||
| func TestChunkDownloadStream_ShutdownWithOpenStream(t *testing.T) { |
There was a problem hiding this comment.
This test does not assert anything about shutdown which is why it would pass with or without the shutdown bug pointed above.
It dials, spins until a ping succeeds, and returns. The service is closed by t.Cleanup — i.e. after the test body has finished — so nothing observes whether Close() succeeded or hit its 1s timeout. The comment says the connection "has to still be open when the service shuts down, otherwise this asserts nothing," which is true, but the assertion it is protecting does not exist yet.
To make it catch the ordering bug: capture the *api.Service, queue enough work that workers are blocked writing to a client that has stopped reading, then assert svc.Close() returns nil inside the budget. That is the exact shape that produced err=api shutting down with open websockets for me.
| } | ||
|
|
||
| // fetchAndSendChunk retrieves a single chunk and writes exactly one response | ||
| // frame for it. That one-frame-per-requested-address invariant is what lets a |
There was a problem hiding this comment.
Exactly one reply per requested address, and "a dropped frame is indistinguishable from a slow one" because replies carry no request id. But the protocol-error paths in the read loop (sendErrorClose + return on bad message type, bad length, unknown opcode, oversized batch) void every in-flight request silently.
A client that sends 256 addresses and then trips CloseMessageTooBig on its next frame has no way to determine which of the first 256 were answered. With no request id and no per-batch boundary marker in the reply stream, the only recovery is to discard everything and re-request.
The OpenAPI spec documents the close codes, but not that pending replies are lost when they fire. At minimum that consequence belongs in the spec. A framing with a request/batch id — which is largely what @acud point about a standard serialization format would give you for free — would make it recoverable instead.
3 participants
Checklist
Description
Adds a download mode to
GET /chunks/stream. A client opens one websocket and pulls many chunks over it, instead of paying for an HTTP request per chunk.Closes #5417, closes #5599.
Protocol
Mode is fixed at handshake — via
Sec-WebSocket-Protocol: swarm-chunk-download, or?mode=downloadfor browser clients that cannot set headers. A connection is an upload stream or a download stream, never both.Request frame:
['D'][32-byte address]..., up to 256 addresses.'D'is the command byte from #5417; batching sits under it so the two compose, and it leaves room to add commands later without breaking clients.Reply frame:
[status][32-byte address][payload]—0x00success,0x01not found,0x02error. Replies arrive out of order as retrievals complete, so clients match on the address. Exactly one reply per requested address.swarm-cacheis honoured on the download path, matchingGET /chunks/{address}.From #5599
Batched requests (256/frame), a bounded worker pool rather than an uncontrolled request storm, streamed partial results, per-address success/failure, and relief from the browser's six-connections-per-host limit. Implemented over websocket — one of the response options that issue listed — rather than
POST /chunks/batch.The video-streaming motivation in #5599 is not served by this endpoint and has been split out. Playback wants HTTP Range and server-side read-ahead on
/bzz; this endpoint has no cancellation and strict FIFO delivery, so a seek would mean dropping the connection. It suits random-access and bulk workloads: manifest traversal, SQLite/Parquet-style page reads, bulk sync, pinning.Notes for review
topology.ErrNotFoundmaps to0x01, matching howbzz.gotreats the same error from the samestorer.Downloadcall.getter.DefaultFetchTimeout, the constant the joiner already uses, so one unreachable chunk can't park a worker indefinitely.WriteControlis safe concurrently, and holding the mutex would let a slow delivery delay the close frame.ReadMessagereturns, instead ofapi.Close()timing out against a 15-minute read deadline.AI Disclosure