fix(gsoc): sequential messaging

pkg/api/gsoc.go

@@ -171,7 +171,10 @@ 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
+		// Buffered enough to absorb a legitimate burst of concurrently delivered
+		// GSOC messages (e.g. several chunks pushed to this address at once)
+		// without tripping the slow-consumer detection below.
+		dataC    = make(chan []byte, 16)
 		gone     = make(chan struct{})
 		slow     = make(chan struct{})
 		slowOnce sync.Once

pkg/api/gsoc_test.go

@@ -290,7 +290,7 @@ func TestGsocWebsocketInvalidFieldsHeader(t *testing.T) {
 func TestGsocWebsocketSlowConsumer(t *testing.T) {
 	t.Parallel()
 
-	const messageCount = 10
+	const messageCount = 32 // exceeds dataC's buffer so the overflow is hit deterministically
 
 	var (
 		id         = make([]byte, 32)

pkg/gsoc/gsoc.go

@@ -73,9 +73,7 @@ func (l *listener) Handle(c *soc.SOC) {
 	l.logger.Debug("new incoming GSOC message", "GSOC Address", addr, "wrapped chunk address", c.WrappedChunk().Address())
 
 	for _, hh := range h {
-		go func(hh Handler) {
-			hh(c)
-		}(*hh)
+		(*hh)(c)
 	}
 }
 

pkg/gsoc/gsoc_test.go

@@ -6,7 +6,6 @@ package gsoc_test
 
 import (
 	"testing"
-	"time"
 
 	"github.com/ethersphere/bee/v2/pkg/cac"
 	"github.com/ethersphere/bee/v2/pkg/crypto"
@@ -25,7 +24,6 @@ func TestRegister(t *testing.T) {
 		h1Calls = 0
 		h2Calls = 0
 		h3Calls = 0
-		msgChan = make(chan struct{})
 
 		payload1    = []byte("Hello there!")
 		payload2    = []byte("General Kenobi. You are a bold one. Kill him!")
@@ -37,20 +35,9 @@ func TestRegister(t *testing.T) {
 		address1, _ = soc.CreateAddress(socId1, owner.Bytes())
 		address2, _ = soc.CreateAddress(socId2, owner.Bytes())
 
-		h1 = func(*soc.SOC) {
-			h1Calls++
-			msgChan <- struct{}{}
-		}
-
-		h2 = func(*soc.SOC) {
-			h2Calls++
-			msgChan <- struct{}{}
-		}
-
-		h3 = func(*soc.SOC) {
-			h3Calls++
-			msgChan <- struct{}{}
-		}
+		h1 = func(*soc.SOC) { h1Calls++ }
+		h2 = func(*soc.SOC) { h2Calls++ }
+		h3 = func(*soc.SOC) { h3Calls++ }
 	)
 	_ = g.Subscribe(address1, h1)
 	_ = g.Subscribe(address2, h2)
@@ -68,8 +55,6 @@ func TestRegister(t *testing.T) {
 	// trigger soc upload on address1, check that only h1 is called
 	g.Handle(socCh1)
 
-	waitHandlerCallback(t, &msgChan, 1)
-
 	ensureCalls(t, &h1Calls, 1)
 	ensureCalls(t, &h2Calls, 0)
 
@@ -78,8 +63,6 @@ func TestRegister(t *testing.T) {
 
 	g.Handle(socCh1)
 
-	waitHandlerCallback(t, &msgChan, 2)
-
 	ensureCalls(t, &h1Calls, 2)
 	ensureCalls(t, &h2Calls, 0)
 	ensureCalls(t, &h3Calls, 1)
@@ -88,16 +71,12 @@ func TestRegister(t *testing.T) {
 
 	g.Handle(socCh1)
 
-	waitHandlerCallback(t, &msgChan, 1)
-
 	ensureCalls(t, &h1Calls, 3)
 	ensureCalls(t, &h2Calls, 0)
 	ensureCalls(t, &h3Calls, 1)
 
 	g.Handle(socCh2)
 
-	waitHandlerCallback(t, &msgChan, 1)
-
 	ensureCalls(t, &h1Calls, 3)
 	ensureCalls(t, &h2Calls, 1)
 	ensureCalls(t, &h3Calls, 1)
@@ -110,15 +89,3 @@ func ensureCalls(t *testing.T, calls *int, exp int) {
 		t.Fatalf("expected %d calls, found %d", exp, *calls)
 	}
 }
-
-func waitHandlerCallback(t *testing.T, msgChan *chan struct{}, count int) {
-	t.Helper()
-
-	for range count {
-		select {
-		case <-*msgChan:
-		case <-time.After(1 * time.Second):
-			t.Fatal("reached timeout while waiting for handler message")
-		}
-	}
-}