test: remaining

Author: nugaon

pkg/api/api_test.go

@@ -137,6 +137,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) {
@@ -251,6 +255,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)

pkg/api/gsoc_test.go

@@ -9,16 +9,20 @@ import (
 	"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"
@@ -243,6 +247,207 @@ func TestGsocWebsocketSocFieldsDeduplication(t *testing.T) {
 	}
 }
 
+// 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.