diff --git a/pkg/api/gsoc.go b/pkg/api/gsoc.go index fe569982c39..843cb235744 100644 --- a/pkg/api/gsoc.go +++ b/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 diff --git a/pkg/api/gsoc_test.go b/pkg/api/gsoc_test.go index 64b926eef69..56832929c9b 100644 --- a/pkg/api/gsoc_test.go +++ b/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) diff --git a/pkg/gsoc/gsoc.go b/pkg/gsoc/gsoc.go index 343bf24aeaf..f4ad79a6ccc 100644 --- a/pkg/gsoc/gsoc.go +++ b/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) } } diff --git a/pkg/gsoc/gsoc_test.go b/pkg/gsoc/gsoc_test.go index 9beb892da51..2ad61fff16d 100644 --- a/pkg/gsoc/gsoc_test.go +++ b/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") - } - } -}