From b15df1b897e88b77f1743a4d441a385965d30e5b Mon Sep 17 00:00:00 2001 From: nugaon Date: Wed, 2 Sep 2026 17:15:43 +0200 Subject: [PATCH 1/2] fix: handle gsoc messages sequentially gsoc.Handle spawned a goroutine per subscriber handler, so message delivery order to a subscriber was not guaranteed. Call handlers synchronously in registration order instead. --- pkg/gsoc/gsoc.go | 4 +--- pkg/gsoc/gsoc_test.go | 39 +++------------------------------------ 2 files changed, 4 insertions(+), 39 deletions(-) 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") - } - } -} From 0be412236b81f97bb5434d53d42e572955dccbf0 Mon Sep 17 00:00:00 2001 From: nugaon Date: Wed, 2 Sep 2026 17:52:24 +0200 Subject: [PATCH 2/2] fix: increase gsoc websocket buffer to absorb legitimate bursts dataC's buffer of 2 was too small for a legitimate burst of GSOC messages delivered concurrently to the same subscriber (e.g. several chunks pushed at once), causing a false-positive slow-consumer disconnect. Bump it to 16 and adjust the slow-consumer test's message count so overflow is still hit deterministically. --- pkg/api/gsoc.go | 5 ++++- pkg/api/gsoc_test.go | 2 +- 2 files changed, 5 insertions(+), 2 deletions(-) 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)