55package api
66
77import (
8+ "bytes"
9+ "context"
10+ "fmt"
811 "net/http"
12+ "slices"
13+ "strings"
14+ "sync"
915 "time"
1016
1117 "github.com/ethersphere/bee/v2/pkg/jsonhttp"
18+ "github.com/ethersphere/bee/v2/pkg/soc"
1219 "github.com/ethersphere/bee/v2/pkg/swarm"
1320 "github.com/gorilla/mux"
1421 "github.com/gorilla/websocket"
1522)
1623
24+ // SOC field identifiers that can be requested through the SwarmSocFieldsHeader
25+ // to be serialized and channeled on every incoming GSOC chunk.
26+ const (
27+ socFieldAddress = "address"
28+ socFieldRecoveredPubKey = "recoveredpubkey"
29+ socFieldIdentifier = "identifier"
30+ socFieldSignature = "signature"
31+ socFieldWrappedAddress = "wrappedaddress"
32+ socFieldSpan = "span"
33+ socFieldPayload = "payload"
34+ )
35+
36+ var validSocFields = []string {
37+ socFieldAddress ,
38+ socFieldRecoveredPubKey ,
39+ socFieldIdentifier ,
40+ socFieldSignature ,
41+ socFieldWrappedAddress ,
42+ socFieldSpan ,
43+ socFieldPayload ,
44+ }
45+
46+ // maxSocFieldsSize is the maximum size of a serialized SOC fields message when
47+ // every field is requested: the whole single owner chunk (identifier +
48+ // signature + span + payload, i.e. SocMaxChunkSize) plus the derived metadata
49+ // fields that are not part of the chunk on the wire (soc address, recovered
50+ // public key and wrapped chunk address).
51+ const maxSocFieldsSize = swarm .SocMaxChunkSize +
52+ swarm .HashSize + // soc address
53+ soc .OwnerPubKeySize + // recovered public key
54+ swarm .HashSize // wrapped chunk address
55+
56+ // parseSocFields parses the SwarmSocFieldsHeader value into a list of SOC field
57+ // identifiers. When the header is empty it defaults to the payload field only,
58+ // which preserves backward compatibility. Duplicate fields are dropped, keeping
59+ // the first occurrence, so the returned slice never exceeds len(validSocFields)
60+ // entries regardless of how many times a field is repeated in the header.
61+ func parseSocFields (header string ) ([]string , error ) {
62+ if strings .TrimSpace (header ) == "" {
63+ return []string {socFieldPayload }, nil
64+ }
65+
66+ seen := make (map [string ]bool , len (validSocFields ))
67+ parts := strings .Split (header , "," )
68+ fields := make ([]string , 0 , len (validSocFields ))
69+ for _ , p := range parts {
70+ f := strings .ToLower (strings .TrimSpace (p ))
71+ if f == "" {
72+ continue
73+ }
74+ if ! slices .Contains (validSocFields , f ) {
75+ return nil , fmt .Errorf ("unknown soc field: %q" , p )
76+ }
77+ if seen [f ] {
78+ continue
79+ }
80+ seen [f ] = true
81+ fields = append (fields , f )
82+ }
83+ if len (fields ) == 0 {
84+ return []string {socFieldPayload }, nil
85+ }
86+ return fields , nil
87+ }
88+
89+ // socFieldsBytes serializes the requested SOC fields in the same order as they
90+ // were provided in the header.
91+ func socFieldsBytes (c * soc.SOC , fields []string ) ([]byte , error ) {
92+ buf := bytes .NewBuffer (nil )
93+ for _ , f := range fields {
94+ switch f {
95+ case socFieldAddress :
96+ addr , err := c .Address ()
97+ if err != nil {
98+ return nil , fmt .Errorf ("soc address: %w" , err )
99+ }
100+ buf .Write (addr .Bytes ())
101+ case socFieldRecoveredPubKey :
102+ buf .Write (c .OwnerPubKey ())
103+ case socFieldIdentifier :
104+ buf .Write (c .ID ())
105+ case socFieldSignature :
106+ buf .Write (c .Signature ())
107+ case socFieldWrappedAddress :
108+ buf .Write (c .WrappedChunk ().Address ().Bytes ())
109+ case socFieldSpan :
110+ buf .Write (c .WrappedChunk ().Data ()[:swarm .SpanSize ])
111+ case socFieldPayload :
112+ buf .Write (c .WrappedChunk ().Data ()[swarm .SpanSize :])
113+ }
114+ }
115+ return buf .Bytes (), nil
116+ }
117+
17118func (s * Service ) gsocWsHandler (w http.ResponseWriter , r * http.Request ) {
18119 logger := s .logger .WithName ("gsoc_subscribe" ).Build ()
19120
@@ -26,9 +127,31 @@ func (s *Service) gsocWsHandler(w http.ResponseWriter, r *http.Request) {
26127 return
27128 }
28129
130+ headers := struct {
131+ SocFields string `map:"Swarm-Soc-Fields"`
132+ CacheWrappedChunk bool `map:"Swarm-Cache-Wrapped-Chunk"`
133+ }{}
134+ if response := s .mapStructure (r .Header , & headers ); response != nil {
135+ response ("invalid header params" , logger , w )
136+ return
137+ }
138+
139+ fields , err := parseSocFields (headers .SocFields )
140+ if err != nil {
141+ logger .Debug ("invalid soc fields header" , "error" , err )
142+ logger .Error (nil , "invalid soc fields header" )
143+ jsonhttp .BadRequest (w , "invalid soc fields header" )
144+ return
145+ }
146+
29147 upgrader := websocket.Upgrader {
30- ReadBufferSize : swarm .ChunkSize ,
31- WriteBufferSize : swarm .ChunkSize ,
148+ ReadBufferSize : swarm .SocMaxChunkSize ,
149+ // WriteBufferSize is only an I/O buffer hint; it does not cap the
150+ // message size. The serialized output can be the whole single owner
151+ // chunk plus the derived metadata fields (soc address, recovered public
152+ // key, wrapped chunk address), so size it to that maximum to avoid split
153+ // writes.
154+ WriteBufferSize : maxSocFieldsSize ,
32155 CheckOrigin : s .checkOrigin ,
33156 }
34157
@@ -41,29 +164,51 @@ func (s *Service) gsocWsHandler(w http.ResponseWriter, r *http.Request) {
41164 }
42165
43166 s .wsWg .Add (1 )
44- go s .gsocListeningWs (conn , paths .Address )
167+ go s .gsocListeningWs (conn , paths .Address , fields , headers . CacheWrappedChunk )
45168}
46169
47- func (s * Service ) gsocListeningWs (conn * websocket.Conn , socAddress swarm.Address ) {
170+ func (s * Service ) gsocListeningWs (conn * websocket.Conn , socAddress swarm.Address , fields [] string , cacheWrappedChunk bool ) {
48171 defer s .wsWg .Done ()
49172
50173 var (
51- dataC = make (chan []byte )
52- gone = make (chan struct {})
53- ticker = time .NewTicker (s .WsPingPeriod )
54- err error
174+ dataC = make (chan []byte , 2 ) // small buffer to decouple producer/consumer
175+ gone = make (chan struct {})
176+ slow = make (chan struct {})
177+ slowOnce sync.Once
178+ ticker = time .NewTicker (s .WsPingPeriod )
179+ err error
55180 )
56181 defer func () {
57182 ticker .Stop ()
58183 _ = conn .Close ()
59184 }()
60- cleanup := s .gsoc .Subscribe (socAddress , func (m []byte ) {
185+ cleanup := s .gsoc .Subscribe (socAddress , func (c * soc.SOC ) {
186+ if cacheWrappedChunk {
187+ // Caching is a node-local side effect independent of this
188+ // subscriber's connection, so it must not be aborted just
189+ // because the websocket closes mid-write.
190+ if err := s .storer .Cache ().Put (context .Background (), c .WrappedChunk ()); err != nil {
191+ s .logger .Debug ("gsoc ws: cache wrapped chunk failed" , "error" , err )
192+ }
193+ }
194+
195+ b , err := socFieldsBytes (c , fields )
196+ if err != nil {
197+ s .logger .Warning ("gsoc ws: serialize soc fields failed" , "error" , err )
198+ return
199+ }
200+
61201 select {
62- case dataC <- m :
202+ case dataC <- b :
63203 case <- gone :
64- return
204+ case <- slow :
65205 case <- s .quit :
66- return
206+ default :
207+ // The connection writer is single-threaded in the main loop below;
208+ // only signal it here instead of writing/closing the conn from this
209+ // callback goroutine, which can run concurrently with the writer.
210+ s .logger .Warning ("gsoc ws: slow consumer, closing connection" )
211+ slowOnce .Do (func () { close (slow ) })
67212 }
68213 })
69214
@@ -105,6 +250,16 @@ func (s *Service) gsocListeningWs(conn *websocket.Conn, socAddress swarm.Address
105250 case <- gone :
106251 // client gone
107252 return
253+ case <- slow :
254+ err = conn .SetWriteDeadline (time .Now ().Add (writeDeadline ))
255+ if err != nil {
256+ s .logger .Debug ("gsoc ws: set write deadline failed" , "error" , err )
257+ return
258+ }
259+ _ = conn .WriteControl (websocket .CloseMessage ,
260+ websocket .FormatCloseMessage (websocket .ClosePolicyViolation , "slow consumer" ),
261+ time .Now ().Add (writeDeadline ))
262+ return
108263 case <- ticker .C :
109264 err = conn .SetWriteDeadline (time .Now ().Add (writeDeadline ))
110265 if err != nil {
0 commit comments