diff --git a/openapi/Swarm.yaml b/openapi/Swarm.yaml index 9140a8b4783..fe753efb560 100644 --- a/openapi/Swarm.yaml +++ b/openapi/Swarm.yaml @@ -1,7 +1,7 @@ openapi: 3.0.3 info: - version: 8.2.0 + version: 8.1.1 title: Bee API description: "API endpoints for interacting with the Swarm network, supporting file operations, messaging, and node management" @@ -967,17 +967,9 @@ 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. 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" + description: Establishes a WebSocket subscription for incoming messages on the Single Owner Chunk address "500": $ref: "SwarmCommon.yaml#/components/responses/500" default: diff --git a/openapi/SwarmCommon.yaml b/openapi/SwarmCommon.yaml index cdacafb3f7f..a1b9718eb3b 100644 --- a/openapi/SwarmCommon.yaml +++ b/openapi/SwarmCommon.yaml @@ -1156,32 +1156,6 @@ 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 426cbcb1837..63a04c390ff 100644 --- a/pkg/api/api.go +++ b/pkg/api/api.go @@ -96,8 +96,6 @@ 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" @@ -609,7 +607,6 @@ 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 c7376d6bbee..23782bcbc6b 100644 --- a/pkg/api/api_test.go +++ b/pkg/api/api_test.go @@ -137,10 +137,6 @@ 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) { @@ -255,10 +251,6 @@ 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 fe569982c39..60d048ffdc0 100644 --- a/pkg/api/gsoc.go +++ b/pkg/api/gsoc.go @@ -5,116 +5,15 @@ 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() @@ -127,31 +26,9 @@ 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.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, + ReadBufferSize: swarm.ChunkSize, + WriteBufferSize: swarm.ChunkSize, CheckOrigin: s.checkOrigin, } @@ -164,51 +41,29 @@ func (s *Service) gsocWsHandler(w http.ResponseWriter, r *http.Request) { } s.wsWg.Add(1) - go s.gsocListeningWs(conn, paths.Address, fields, headers.CacheWrappedChunk) + go s.gsocListeningWs(conn, paths.Address) } -func (s *Service) gsocListeningWs(conn *websocket.Conn, socAddress swarm.Address, fields []string, cacheWrappedChunk bool) { +func (s *Service) gsocListeningWs(conn *websocket.Conn, socAddress swarm.Address) { defer s.wsWg.Done() var ( - 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 + dataC = make(chan []byte) + gone = make(chan struct{}) + ticker = time.NewTicker(s.WsPingPeriod) + err error ) defer func() { ticker.Stop() _ = conn.Close() }() - 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 - } - + cleanup := s.gsoc.Subscribe(socAddress, func(m []byte) { select { - case dataC <- b: + case dataC <- m: case <-gone: - case <-slow: + return case <-s.quit: - 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) }) + return } }) @@ -250,16 +105,6 @@ 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 64b926eef69..cf3161a9f03 100644 --- a/pkg/api/gsoc_test.go +++ b/pkg/api/gsoc_test.go @@ -5,24 +5,16 @@ 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" @@ -142,359 +134,7 @@ 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 @@ -521,12 +161,11 @@ func newGsocTestWithOpts(t *testing.T, socId []byte, pingPeriod time.Duration, h _, 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, storer + return gsoc, cl, signer, listener } diff --git a/pkg/gsoc/gsoc.go b/pkg/gsoc/gsoc.go index 343bf24aeaf..41e4f54ac2c 100644 --- a/pkg/gsoc/gsoc.go +++ b/pkg/gsoc/gsoc.go @@ -13,9 +13,8 @@ import ( ) // Handler defines code to be executed upon reception of a GSOC sub message. -// 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) +// it is used as a parameter definition. +type Handler func([]byte) type Listener interface { Subscribe(address swarm.Address, handler Handler) (cleanup func()) @@ -74,7 +73,7 @@ func (l *listener) Handle(c *soc.SOC) { for _, hh := range h { go func(hh Handler) { - hh(c) + hh(c.WrappedChunk().Data()[swarm.SpanSize:]) }(*hh) } } diff --git a/pkg/gsoc/gsoc_test.go b/pkg/gsoc/gsoc_test.go index 9beb892da51..dc49b0809a8 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(*soc.SOC) { + h1 = func(m []byte) { h1Calls++ msgChan <- struct{}{} } - h2 = func(*soc.SOC) { + h2 = func(m []byte) { h2Calls++ msgChan <- struct{}{} } - h3 = func(*soc.SOC) { + h3 = func(m []byte) { h3Calls++ msgChan <- struct{}{} } diff --git a/pkg/soc/soc.go b/pkg/soc/soc.go index 09a6cd81c2e..28ade83eeb8 100644 --- a/pkg/soc/soc.go +++ b/pkg/soc/soc.go @@ -20,10 +20,6 @@ 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