chore: revert #5497

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:

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

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, ", ")
 

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)

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 {