Skip to content

fix(gsoc): client too slow handling #5593

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
nugaon wants to merge 3 commits into from
+10 −42
Closed

fix(gsoc): client too slow handling #5593

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
b15df1b
fix: handle gsoc messages sequentially
nugaon Sep 2, 2026
0be4122
fix: increase gsoc websocket buffer to absorb legitimate bursts
nugaon Sep 2, 2026
7375a24
docs: cap number
nugaon Sep 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion pkg/api/gsoc.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)

@aloknerurkar aloknerurkar Sep 8, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not sure why this is needed in this PR. Is there any reasoning behind why 16 is the right value to be used here? If there is truly a burst > 16 we will still trip the slow-consumer detection below. Even the previous value of 2 is questionable.

If this is an actual problem, the correct fix would be something like a time-windowed rate check, or a slightly larger drain-then-reconsider window, or just accepting that a truly overwhelmed consumer should be disconnected and tuning based on expected real-world message rates than using some arbitrary constant. Having this change in this PR feels like the value is made up so the tests pass. It would be better to not have this in the current PR and making a proper fix in a subsequent PR.

@nugaon nugaon Sep 9, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The time-windowed rate check wouldn't be the best fit since the data feed is sporadic.
I don't know what would be the optimal value, wdyt?
we can remove the whole condition as well if you are fine with that.

@nugaon nugaon Sep 9, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

anyway, I think 16 is a fine number, ofc the test scenario a bit unreal but in case of a really popular gsoc topic it can happen 16 messages arrive the same time. ofc, even more popular topic would get maybe 32...

@martinconic martinconic Sep 11, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please check my comment from the previous PR: #5497 (comment)

gone = make(chan struct{})
slow = make(chan struct{})
slowOnce sync.Once
Expand Down
4 changes: 2 additions & 2 deletions pkg/api/gsoc_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -338,7 +338,7 @@ func TestGsocWebsocketSlowConsumer(t *testing.T) {
}
testutil.CleanupCloser(t, cl)

// never read from cl, so the dataC buffer (cap 2) fills up almost
// never read from cl, so the dataC buffer (cap 16) 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 {
Expand Down
4 changes: 1 addition & 3 deletions pkg/gsoc/gsoc.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
}
}

Expand Down
39 changes: 3 additions & 36 deletions pkg/gsoc/gsoc_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,6 @@ package gsoc_test

import (
"testing"
"time"

"github.com/ethersphere/bee/v2/pkg/cac"
"github.com/ethersphere/bee/v2/pkg/crypto"
Expand All @@ -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!")
Expand All @@ -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)
Expand All @@ -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)

Expand All @@ -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)
Expand All @@ -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)
Expand All @@ -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")
}
}
}
Loading