diff --git a/openapi/Swarm.yaml b/openapi/Swarm.yaml index 9a0eafc96f1..0db1a05b9e3 100644 --- a/openapi/Swarm.yaml +++ b/openapi/Swarm.yaml @@ -1,7 +1,7 @@ openapi: 3.0.3 info: - version: 8.1.0 + version: 8.2.0 title: Bee API description: "API endpoints for interacting with the Swarm network, supporting file operations, messaging, and node management" @@ -859,9 +859,17 @@ paths: $ref: "SwarmCommon.yaml#/components/schemas/SwarmAddress" required: true description: "Single Owner Chunk address (which may have multiple payloads)" + - $ref: "SwarmCommon.yaml#/components/parameters/SwarmSocFieldsParameter" + - $ref: "SwarmCommon.yaml#/components/parameters/SwarmCacheWrappedChunkParameter" responses: "200": - description: Establishes a WebSocket subscription for incoming messages on the Single Owner Chunk address + description: > + Establishes a WebSocket subscription for incoming messages on the + Single Owner Chunk address. Each message is the binary serialization + of the Single Owner Chunk fields requested through the + swarm-soc-fields header (defaults to the wrapped chunk payload). + "400": + $ref: "SwarmCommon.yaml#/components/responses/400" "500": $ref: "SwarmCommon.yaml#/components/responses/500" default: diff --git a/openapi/SwarmCommon.yaml b/openapi/SwarmCommon.yaml index ffcbbac3b8e..ef7fd9b9390 100644 --- a/openapi/SwarmCommon.yaml +++ b/openapi/SwarmCommon.yaml @@ -1144,6 +1144,32 @@ components: required: false description: Associate upload with an existing Tag UID + SwarmSocFieldsParameter: + in: header + name: swarm-soc-fields + schema: + type: string + default: "payload" + required: false + description: > + Comma separated list of Single Owner Chunk fields to be serialized and + channeled on every incoming GSOC message, in the given order. Allowed + values are: address, recoveredPubKey, identifier, signature, + wrappedAddress, span, payload. When omitted it defaults to "payload". + In order to have random access on the response bytes define payload + as the last field in the list since it has variable length. + + SwarmCacheWrappedChunkParameter: + in: header + name: swarm-cache-wrapped-chunk + schema: + type: boolean + required: false + description: > + Indicates whether the wrapped chunk of every incoming GSOC message should + be cached locally so that it can be resolved through the bytes endpoint + (useful when the single owner chunk wraps a root chunk larger than 4KB). + SwarmPinParameter: in: header name: swarm-pin diff --git a/pkg/api/api.go b/pkg/api/api.go index acd838a3ff6..729f5a0812d 100644 --- a/pkg/api/api.go +++ b/pkg/api/api.go @@ -93,6 +93,8 @@ const ( SwarmActTimestampHeader = "Swarm-Act-Timestamp" SwarmActPublisherHeader = "Swarm-Act-Publisher" SwarmActHistoryAddressHeader = "Swarm-Act-History-Address" + SwarmSocFieldsHeader = "Swarm-Soc-Fields" + SwarmCacheWrappedChunkHeader = "Swarm-Cache-Wrapped-Chunk" ImmutableHeader = "Immutable" GasPriceHeader = "Gas-Price" @@ -583,6 +585,7 @@ func (s *Service) corsHandler(h http.Handler) http.Handler { SwarmRedundancyStrategyHeader, SwarmRedundancyFallbackModeHeader, SwarmChunkRetrievalTimeoutHeader, SwarmLookAheadBufferSizeHeader, SwarmFeedIndexHeader, SwarmFeedIndexNextHeader, SwarmSocSignatureHeader, SwarmOnlyRootChunk, GasPriceHeader, GasLimitHeader, ImmutableHeader, SwarmActHeader, SwarmActTimestampHeader, SwarmActPublisherHeader, SwarmActHistoryAddressHeader, + SwarmSocFieldsHeader, SwarmCacheWrappedChunkHeader, } allowedHeadersStr := strings.Join(allowedHeaders, ", ") diff --git a/pkg/api/api_test.go b/pkg/api/api_test.go index babd816dd06..b4953efb5ae 100644 --- a/pkg/api/api_test.go +++ b/pkg/api/api_test.go @@ -136,6 +136,10 @@ type testServerOptions struct { ChequebookDisabled bool SwapDisabled bool Erc20ServiceNil bool + // ServiceOut, when set, receives the constructed *api.Service so tests + // can drive it directly (e.g. via a custom net.Listener) instead of + // through the httptest.Server this function also sets up. + ServiceOut **api.Service } func newTestServer(t *testing.T, o testServerOptions) (*http.Client, *websocket.Conn, string, *chanStorer) { @@ -246,6 +250,10 @@ func newTestServer(t *testing.T, o testServerOptions) (*http.Client, *websocket. s.EnableFullAPI() } + if o.ServiceOut != nil { + *o.ServiceOut = s + } + if o.DirectUpload { chanStore = newChanStore(o.Storer.PusherFeed()) t.Cleanup(chanStore.stop) diff --git a/pkg/api/gsoc.go b/pkg/api/gsoc.go index 60d048ffdc0..fe569982c39 100644 --- a/pkg/api/gsoc.go +++ b/pkg/api/gsoc.go @@ -5,15 +5,116 @@ package api import ( + "bytes" + "context" + "fmt" "net/http" + "slices" + "strings" + "sync" "time" "github.com/ethersphere/bee/v2/pkg/jsonhttp" + "github.com/ethersphere/bee/v2/pkg/soc" "github.com/ethersphere/bee/v2/pkg/swarm" "github.com/gorilla/mux" "github.com/gorilla/websocket" ) +// SOC field identifiers that can be requested through the SwarmSocFieldsHeader +// to be serialized and channeled on every incoming GSOC chunk. +const ( + socFieldAddress = "address" + socFieldRecoveredPubKey = "recoveredpubkey" + socFieldIdentifier = "identifier" + socFieldSignature = "signature" + socFieldWrappedAddress = "wrappedaddress" + socFieldSpan = "span" + socFieldPayload = "payload" +) + +var validSocFields = []string{ + socFieldAddress, + socFieldRecoveredPubKey, + socFieldIdentifier, + socFieldSignature, + socFieldWrappedAddress, + socFieldSpan, + socFieldPayload, +} + +// maxSocFieldsSize is the maximum size of a serialized SOC fields message when +// every field is requested: the whole single owner chunk (identifier + +// signature + span + payload, i.e. SocMaxChunkSize) plus the derived metadata +// fields that are not part of the chunk on the wire (soc address, recovered +// public key and wrapped chunk address). +const maxSocFieldsSize = swarm.SocMaxChunkSize + + swarm.HashSize + // soc address + soc.OwnerPubKeySize + // recovered public key + swarm.HashSize // wrapped chunk address + +// parseSocFields parses the SwarmSocFieldsHeader value into a list of SOC field +// identifiers. When the header is empty it defaults to the payload field only, +// which preserves backward compatibility. Duplicate fields are dropped, keeping +// the first occurrence, so the returned slice never exceeds len(validSocFields) +// entries regardless of how many times a field is repeated in the header. +func parseSocFields(header string) ([]string, error) { + if strings.TrimSpace(header) == "" { + return []string{socFieldPayload}, nil + } + + seen := make(map[string]bool, len(validSocFields)) + parts := strings.Split(header, ",") + fields := make([]string, 0, len(validSocFields)) + for _, p := range parts { + f := strings.ToLower(strings.TrimSpace(p)) + if f == "" { + continue + } + if !slices.Contains(validSocFields, f) { + return nil, fmt.Errorf("unknown soc field: %q", p) + } + if seen[f] { + continue + } + seen[f] = true + fields = append(fields, f) + } + if len(fields) == 0 { + return []string{socFieldPayload}, nil + } + return fields, nil +} + +// socFieldsBytes serializes the requested SOC fields in the same order as they +// were provided in the header. +func socFieldsBytes(c *soc.SOC, fields []string) ([]byte, error) { + buf := bytes.NewBuffer(nil) + for _, f := range fields { + switch f { + case socFieldAddress: + addr, err := c.Address() + if err != nil { + return nil, fmt.Errorf("soc address: %w", err) + } + buf.Write(addr.Bytes()) + case socFieldRecoveredPubKey: + buf.Write(c.OwnerPubKey()) + case socFieldIdentifier: + buf.Write(c.ID()) + case socFieldSignature: + buf.Write(c.Signature()) + case socFieldWrappedAddress: + buf.Write(c.WrappedChunk().Address().Bytes()) + case socFieldSpan: + buf.Write(c.WrappedChunk().Data()[:swarm.SpanSize]) + case socFieldPayload: + buf.Write(c.WrappedChunk().Data()[swarm.SpanSize:]) + } + } + return buf.Bytes(), nil +} + func (s *Service) gsocWsHandler(w http.ResponseWriter, r *http.Request) { logger := s.logger.WithName("gsoc_subscribe").Build() @@ -26,9 +127,31 @@ func (s *Service) gsocWsHandler(w http.ResponseWriter, r *http.Request) { return } + headers := struct { + SocFields string `map:"Swarm-Soc-Fields"` + CacheWrappedChunk bool `map:"Swarm-Cache-Wrapped-Chunk"` + }{} + if response := s.mapStructure(r.Header, &headers); response != nil { + response("invalid header params", logger, w) + return + } + + fields, err := parseSocFields(headers.SocFields) + if err != nil { + logger.Debug("invalid soc fields header", "error", err) + logger.Error(nil, "invalid soc fields header") + jsonhttp.BadRequest(w, "invalid soc fields header") + return + } + upgrader := websocket.Upgrader{ - ReadBufferSize: swarm.ChunkSize, - WriteBufferSize: swarm.ChunkSize, + ReadBufferSize: swarm.SocMaxChunkSize, + // WriteBufferSize is only an I/O buffer hint; it does not cap the + // message size. The serialized output can be the whole single owner + // chunk plus the derived metadata fields (soc address, recovered public + // key, wrapped chunk address), so size it to that maximum to avoid split + // writes. + WriteBufferSize: maxSocFieldsSize, CheckOrigin: s.checkOrigin, } @@ -41,29 +164,51 @@ func (s *Service) gsocWsHandler(w http.ResponseWriter, r *http.Request) { } s.wsWg.Add(1) - go s.gsocListeningWs(conn, paths.Address) + go s.gsocListeningWs(conn, paths.Address, fields, headers.CacheWrappedChunk) } -func (s *Service) gsocListeningWs(conn *websocket.Conn, socAddress swarm.Address) { +func (s *Service) gsocListeningWs(conn *websocket.Conn, socAddress swarm.Address, fields []string, cacheWrappedChunk bool) { defer s.wsWg.Done() var ( - dataC = make(chan []byte) - gone = make(chan struct{}) - ticker = time.NewTicker(s.WsPingPeriod) - err error + dataC = make(chan []byte, 2) // small buffer to decouple producer/consumer + gone = make(chan struct{}) + slow = make(chan struct{}) + slowOnce sync.Once + ticker = time.NewTicker(s.WsPingPeriod) + err error ) defer func() { ticker.Stop() _ = conn.Close() }() - cleanup := s.gsoc.Subscribe(socAddress, func(m []byte) { + cleanup := s.gsoc.Subscribe(socAddress, func(c *soc.SOC) { + if cacheWrappedChunk { + // Caching is a node-local side effect independent of this + // subscriber's connection, so it must not be aborted just + // because the websocket closes mid-write. + if err := s.storer.Cache().Put(context.Background(), c.WrappedChunk()); err != nil { + s.logger.Debug("gsoc ws: cache wrapped chunk failed", "error", err) + } + } + + b, err := socFieldsBytes(c, fields) + if err != nil { + s.logger.Warning("gsoc ws: serialize soc fields failed", "error", err) + return + } + select { - case dataC <- m: + case dataC <- b: case <-gone: - return + case <-slow: case <-s.quit: - return + default: + // The connection writer is single-threaded in the main loop below; + // only signal it here instead of writing/closing the conn from this + // callback goroutine, which can run concurrently with the writer. + s.logger.Warning("gsoc ws: slow consumer, closing connection") + slowOnce.Do(func() { close(slow) }) } }) @@ -105,6 +250,16 @@ func (s *Service) gsocListeningWs(conn *websocket.Conn, socAddress swarm.Address case <-gone: // client gone return + case <-slow: + err = conn.SetWriteDeadline(time.Now().Add(writeDeadline)) + if err != nil { + s.logger.Debug("gsoc ws: set write deadline failed", "error", err) + return + } + _ = conn.WriteControl(websocket.CloseMessage, + websocket.FormatCloseMessage(websocket.ClosePolicyViolation, "slow consumer"), + time.Now().Add(writeDeadline)) + return case <-ticker.C: err = conn.SetWriteDeadline(time.Now().Add(writeDeadline)) if err != nil { diff --git a/pkg/api/gsoc_test.go b/pkg/api/gsoc_test.go index cf3161a9f03..64b926eef69 100644 --- a/pkg/api/gsoc_test.go +++ b/pkg/api/gsoc_test.go @@ -5,16 +5,24 @@ package api_test import ( + "bytes" + "context" "encoding/hex" "fmt" + "net" + "net/http" "net/url" "strings" + "sync" "testing" "time" + "github.com/ethersphere/bee/v2/pkg/api" "github.com/ethersphere/bee/v2/pkg/cac" "github.com/ethersphere/bee/v2/pkg/crypto" "github.com/ethersphere/bee/v2/pkg/gsoc" + "github.com/ethersphere/bee/v2/pkg/jsonhttp" + "github.com/ethersphere/bee/v2/pkg/jsonhttp/jsonhttptest" "github.com/ethersphere/bee/v2/pkg/log" mockbatchstore "github.com/ethersphere/bee/v2/pkg/postage/batchstore/mock" "github.com/ethersphere/bee/v2/pkg/soc" @@ -134,7 +142,359 @@ func TestGsocPong(t *testing.T) { } } +// TestGsocWebsocketWrappedChunkData verifies that the Swarm-Soc-Fields header +// allows requesting the whole wrapped chunk data (span + payload). +func TestGsocWebsocketWrappedChunkData(t *testing.T) { + t.Parallel() + + var ( + id = make([]byte, 32) + headers = http.Header{api.SwarmSocFieldsHeader: []string{"span,payload"}} + g, cl, signer, _, _ = newGsocTestWithOpts(t, id, 0, headers) + respC = make(chan error, 1) + payload = []byte("The most dangerous phrase in the language is: ‘We've always done it this way.’") + ) + + err := cl.SetReadDeadline(time.Now().Add(longTimeout)) + if err != nil { + t.Fatal(err) + } + cl.SetReadLimit(swarm.ChunkSize) + + ch, _ := cac.New(payload) + socCh := soc.New(id, ch) + signedCh, _ := socCh.Sign(signer) + socCh, _ = soc.FromChunk(signedCh) + g.Handle(socCh) + + // span (8 bytes) + payload == full wrapped chunk data + go expectMessage(t, cl, respC, ch.Data()) + if err := <-respC; err != nil { + t.Fatal(err) + } +} + +// TestGsocWebsocketSocFields verifies that multiple SOC fields are serialized in +// the order they are provided in the Swarm-Soc-Fields header. +func TestGsocWebsocketSocFields(t *testing.T) { + t.Parallel() + + var ( + id = make([]byte, 32) + headers = http.Header{api.SwarmSocFieldsHeader: []string{"identifier,wrappedAddress,payload"}} + g, cl, signer, _, _ = newGsocTestWithOpts(t, id, 0, headers) + respC = make(chan error, 1) + payload = []byte("The future is already here — it's just not evenly distributed.") + ) + + err := cl.SetReadDeadline(time.Now().Add(longTimeout)) + if err != nil { + t.Fatal(err) + } + cl.SetReadLimit(swarm.ChunkSize) + + ch, _ := cac.New(payload) + socCh := soc.New(id, ch) + signedCh, _ := socCh.Sign(signer) + socCh, _ = soc.FromChunk(signedCh) + g.Handle(socCh) + + expected := make([]byte, 0, len(id)+swarm.HashSize+len(payload)) + expected = append(expected, id...) + expected = append(expected, ch.Address().Bytes()...) + expected = append(expected, payload...) + + go expectMessage(t, cl, respC, expected) + if err := <-respC; err != nil { + t.Fatal(err) + } +} + +// TestGsocWebsocketSocFieldsDeduplication verifies that repeated field names in +// the Swarm-Soc-Fields header are de-duplicated, keeping only the first +// occurrence, instead of serializing the same field multiple times. +func TestGsocWebsocketSocFieldsDeduplication(t *testing.T) { + t.Parallel() + + var ( + id = make([]byte, 32) + headers = http.Header{api.SwarmSocFieldsHeader: []string{"payload,payload,identifier,payload,identifier"}} + g, cl, signer, _, _ = newGsocTestWithOpts(t, id, 0, headers) + respC = make(chan error, 1) + payload = []byte("Simplicity is the ultimate sophistication.") + ) + + err := cl.SetReadDeadline(time.Now().Add(longTimeout)) + if err != nil { + t.Fatal(err) + } + cl.SetReadLimit(swarm.ChunkSize) + + ch, _ := cac.New(payload) + socCh := soc.New(id, ch) + signedCh, _ := socCh.Sign(signer) + socCh, _ = soc.FromChunk(signedCh) + g.Handle(socCh) + + // each requested field must appear exactly once, in first-occurrence order + expected := make([]byte, 0, len(payload)+len(id)) + expected = append(expected, payload...) + expected = append(expected, id...) + + go expectMessage(t, cl, respC, expected) + if err := <-respC; err != nil { + t.Fatal(err) + } +} + +// TestGsocWebsocketInvalidFieldsHeader verifies that an unknown field name in +// the Swarm-Soc-Fields header is rejected with a 400 Bad Request before the +// websocket upgrade is attempted. +func TestGsocWebsocketInvalidFieldsHeader(t *testing.T) { + t.Parallel() + + var ( + id = make([]byte, 32) + gsocSvc = gsoc.New(log.Noop) + addrHex = hex.EncodeToString(id) + batchStore = mockbatchstore.New() + storer = mockstorer.New() + ) + testutil.CleanupCloser(t, gsocSvc) + + client, _, _, _ := newTestServer(t, testServerOptions{ + Gsoc: gsocSvc, + Storer: storer, + BatchStore: batchStore, + Logger: log.Noop, + }) + + jsonhttptest.Request(t, client, http.MethodGet, "/gsoc/subscribe/"+addrHex, http.StatusBadRequest, + jsonhttptest.WithRequestHeader(api.SwarmSocFieldsHeader, "bogusfield"), + jsonhttptest.WithExpectedJSONResponse(jsonhttp.StatusResponse{ + Message: "invalid soc fields header", + Code: http.StatusBadRequest, + }), + ) +} + +// TestGsocWebsocketSlowConsumer verifies that when a subscriber cannot keep up +// with incoming GSOC messages, the server closes the connection instead of +// blocking indefinitely or racing on the underlying websocket connection. +// +// The connection is served over an in-memory net.Pipe, which is fully +// synchronous (unbuffered): a write only completes once a matching read +// consumes it. This makes the small dataC buffer overflow deterministically +// as soon as the client stops reading, instead of depending on the size of +// the OS's (possibly very large, auto-tuned) TCP socket buffers. +func TestGsocWebsocketSlowConsumer(t *testing.T) { + t.Parallel() + + const messageCount = 10 + + var ( + id = make([]byte, 32) + batchStore = mockbatchstore.New() + storer = mockstorer.New() + gsocSvc = gsoc.New(log.Noop) + svc *api.Service + ) + testutil.CleanupCloser(t, gsocSvc) + + newTestServer(t, testServerOptions{ + Gsoc: gsocSvc, + Storer: storer, + BatchStore: batchStore, + Logger: log.Noop, + ServiceOut: &svc, + }) + + privKey, err := crypto.GenerateSecp256k1Key() + if err != nil { + t.Fatal(err) + } + signer := crypto.NewDefaultSigner(privKey) + owner, err := signer.EthereumAddress() + if err != nil { + t.Fatal(err) + } + chunkAddr, _ := soc.CreateAddress(id, owner.Bytes()) + + ln := newPipeListener() + srv := &http.Server{Handler: svc} + testutil.CleanupCloser(t, srv) + go func() { _ = srv.Serve(ln) }() + + clientConn, serverConn := net.Pipe() + ln.offer(serverConn) + + u := url.URL{Scheme: "ws", Host: "pipe", Path: "/gsoc/subscribe/" + hex.EncodeToString(chunkAddr.Bytes())} + dialer := websocket.Dialer{ + NetDial: func(_, _ string) (net.Conn, error) { return clientConn, nil }, + } + cl, _, err := dialer.Dial(u.String(), nil) + if err != nil { + t.Fatalf("client handshake: %v", err) + } + testutil.CleanupCloser(t, cl) + + // never read from cl, so the dataC buffer (cap 2) fills up almost + // immediately: the first message blocks the single writer goroutine + // (nothing reads the pipe), and the next ones queue up and overflow. + for i := range messageCount { + payload := []byte{byte(i)} + ch, _ := cac.New(payload) + socCh := soc.New(id, ch) + signedCh, _ := socCh.Sign(signer) + socCh, _ = soc.FromChunk(signedCh) + gsocSvc.Handle(socCh) + } + + if err := cl.SetReadDeadline(time.Now().Add(longTimeout)); err != nil { + t.Fatal(err) + } + + // Drain whatever messages had already been handed to the (synchronous) + // pipe before the overflow was detected; the connection must eventually + // be closed instead of the server delivering every message regardless of + // how far behind the consumer falls. + var readErr error + for i := 0; i < messageCount && readErr == nil; i++ { + _, _, readErr = cl.ReadMessage() + } + if readErr == nil { + t.Fatal("expected connection to be closed for a slow consumer") + } +} + +// pipeListener is a net.Listener that hands out pre-established net.Conn +// pairs, so an http.Server can be driven over an in-memory net.Pipe instead +// of a real OS socket. +type pipeListener struct { + connCh chan net.Conn + closed chan struct{} + once sync.Once +} + +func newPipeListener() *pipeListener { + return &pipeListener{ + connCh: make(chan net.Conn, 1), + closed: make(chan struct{}), + } +} + +func (l *pipeListener) offer(conn net.Conn) { l.connCh <- conn } + +func (l *pipeListener) Accept() (net.Conn, error) { + select { + case c := <-l.connCh: + return c, nil + case <-l.closed: + return nil, net.ErrClosed + } +} + +func (l *pipeListener) Close() error { + l.once.Do(func() { close(l.closed) }) + return nil +} + +func (l *pipeListener) Addr() net.Addr { return pipeAddr{} } + +type pipeAddr struct{} + +func (pipeAddr) Network() string { return "pipe" } +func (pipeAddr) String() string { return "pipe" } + +// TestGsocWebsocketMessageOrdering verifies that sequential Handle calls for +// the same GSOC address are delivered to the subscriber in the same order. +func TestGsocWebsocketMessageOrdering(t *testing.T) { + t.Parallel() + + const messageCount = 10 + + var ( + id = make([]byte, 32) + g, cl, signer, _ = newGsocTest(t, id, 0) + ) + + err := cl.SetReadDeadline(time.Now().Add(longTimeout)) + if err != nil { + t.Fatal(err) + } + cl.SetReadLimit(swarm.ChunkSize) + + payloads := make([][]byte, messageCount) + for i := range payloads { + payloads[i] = fmt.Appendf(nil, "message-%d", i) + } + + for _, payload := range payloads { + ch, _ := cac.New(payload) + socCh := soc.New(id, ch) + signedCh, _ := socCh.Sign(signer) + socCh, _ = soc.FromChunk(signedCh) + g.Handle(socCh) + } + + for i, want := range payloads { + _, got, err := cl.ReadMessage() + if err != nil { + t.Fatalf("message %d: %v", i, err) + } + if !bytes.Equal(got, want) { + t.Fatalf("message %d: got %q, want %q", i, got, want) + } + } +} + +// TestGsocWebsocketCacheWrappedChunk verifies that the Swarm-Cache-Wrapped-Chunk +// header causes the wrapped chunk to be stored in the cache so that it can be +// resolved through the bytes endpoint. +func TestGsocWebsocketCacheWrappedChunk(t *testing.T) { + t.Parallel() + + var ( + id = make([]byte, 32) + headers = http.Header{api.SwarmCacheWrappedChunkHeader: []string{"true"}} + g, cl, signer, _, storer = newGsocTestWithOpts(t, id, 0, headers) + respC = make(chan error, 1) + payload = []byte("If you don't like change, you're going to like irrelevance even less.") + ) + + err := cl.SetReadDeadline(time.Now().Add(longTimeout)) + if err != nil { + t.Fatal(err) + } + cl.SetReadLimit(swarm.ChunkSize) + + ch, _ := cac.New(payload) + socCh := soc.New(id, ch) + signedCh, _ := socCh.Sign(signer) + socCh, _ = soc.FromChunk(signedCh) + g.Handle(socCh) + + go expectMessage(t, cl, respC, payload) + if err := <-respC; err != nil { + t.Fatal(err) + } + + got, err := storer.ChunkStore().Get(context.Background(), ch.Address()) + if err != nil { + t.Fatalf("wrapped chunk not cached: %v", err) + } + if !bytes.Equal(got.Data(), ch.Data()) { + t.Fatal("cached wrapped chunk data mismatch") + } +} + func newGsocTest(t *testing.T, socId []byte, pingPeriod time.Duration) (gsoc.Listener, *websocket.Conn, crypto.Signer, string) { + t.Helper() + g, cl, signer, listener, _ := newGsocTestWithOpts(t, socId, pingPeriod, nil) + return g, cl, signer, listener +} + +func newGsocTestWithOpts(t *testing.T, socId []byte, pingPeriod time.Duration, headers http.Header) (gsoc.Listener, *websocket.Conn, crypto.Signer, string, api.Storer) { t.Helper() if pingPeriod == 0 { pingPeriod = 10 * time.Second @@ -161,11 +521,12 @@ func newGsocTest(t *testing.T, socId []byte, pingPeriod time.Duration) (gsoc.Lis _, cl, listener, _ := newTestServer(t, testServerOptions{ Gsoc: gsoc, WsPath: fmt.Sprintf("/gsoc/subscribe/%s", hex.EncodeToString(chunkAddr.Bytes())), + WsHeaders: headers, Storer: storer, BatchStore: batchStore, Logger: log.Noop, WsPingPeriod: pingPeriod, }) - return gsoc, cl, signer, listener + return gsoc, cl, signer, listener, storer } diff --git a/pkg/gsoc/gsoc.go b/pkg/gsoc/gsoc.go index 41e4f54ac2c..343bf24aeaf 100644 --- a/pkg/gsoc/gsoc.go +++ b/pkg/gsoc/gsoc.go @@ -13,8 +13,9 @@ import ( ) // Handler defines code to be executed upon reception of a GSOC sub message. -// it is used as a parameter definition. -type Handler func([]byte) +// it is used as a parameter definition. It receives the recovered single owner +// chunk so the consumer has access to all of its properties. +type Handler func(*soc.SOC) type Listener interface { Subscribe(address swarm.Address, handler Handler) (cleanup func()) @@ -73,7 +74,7 @@ func (l *listener) Handle(c *soc.SOC) { for _, hh := range h { go func(hh Handler) { - hh(c.WrappedChunk().Data()[swarm.SpanSize:]) + hh(c) }(*hh) } } diff --git a/pkg/gsoc/gsoc_test.go b/pkg/gsoc/gsoc_test.go index dc49b0809a8..9beb892da51 100644 --- a/pkg/gsoc/gsoc_test.go +++ b/pkg/gsoc/gsoc_test.go @@ -37,17 +37,17 @@ func TestRegister(t *testing.T) { address1, _ = soc.CreateAddress(socId1, owner.Bytes()) address2, _ = soc.CreateAddress(socId2, owner.Bytes()) - h1 = func(m []byte) { + h1 = func(*soc.SOC) { h1Calls++ msgChan <- struct{}{} } - h2 = func(m []byte) { + h2 = func(*soc.SOC) { h2Calls++ msgChan <- struct{}{} } - h3 = func(m []byte) { + h3 = func(*soc.SOC) { h3Calls++ msgChan <- struct{}{} } diff --git a/pkg/soc/soc.go b/pkg/soc/soc.go index 28ade83eeb8..09a6cd81c2e 100644 --- a/pkg/soc/soc.go +++ b/pkg/soc/soc.go @@ -20,6 +20,10 @@ var ( errWrongChunkSize = errors.New("soc: chunk length is less than minimum") ) +// OwnerPubKeySize is the byte length of a compressed secp256k1 public key, +// as returned by crypto.EncodeSecp256k1PublicKey. +const OwnerPubKeySize = 33 + // ID is a SOC identifier type ID []byte