Skip to content

Commit b15df1b

Browse files
committed
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.
1 parent e0fb3a6 commit b15df1b

2 files changed

Lines changed: 4 additions & 39 deletions

File tree

pkg/gsoc/gsoc.go

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -73,9 +73,7 @@ func (l *listener) Handle(c *soc.SOC) {
7373
l.logger.Debug("new incoming GSOC message", "GSOC Address", addr, "wrapped chunk address", c.WrappedChunk().Address())
7474

7575
for _, hh := range h {
76-
go func(hh Handler) {
77-
hh(c)
78-
}(*hh)
76+
(*hh)(c)
7977
}
8078
}
8179

pkg/gsoc/gsoc_test.go

Lines changed: 3 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,6 @@ package gsoc_test
66

77
import (
88
"testing"
9-
"time"
109

1110
"github.com/ethersphere/bee/v2/pkg/cac"
1211
"github.com/ethersphere/bee/v2/pkg/crypto"
@@ -25,7 +24,6 @@ func TestRegister(t *testing.T) {
2524
h1Calls = 0
2625
h2Calls = 0
2726
h3Calls = 0
28-
msgChan = make(chan struct{})
2927

3028
payload1 = []byte("Hello there!")
3129
payload2 = []byte("General Kenobi. You are a bold one. Kill him!")
@@ -37,20 +35,9 @@ func TestRegister(t *testing.T) {
3735
address1, _ = soc.CreateAddress(socId1, owner.Bytes())
3836
address2, _ = soc.CreateAddress(socId2, owner.Bytes())
3937

40-
h1 = func(*soc.SOC) {
41-
h1Calls++
42-
msgChan <- struct{}{}
43-
}
44-
45-
h2 = func(*soc.SOC) {
46-
h2Calls++
47-
msgChan <- struct{}{}
48-
}
49-
50-
h3 = func(*soc.SOC) {
51-
h3Calls++
52-
msgChan <- struct{}{}
53-
}
38+
h1 = func(*soc.SOC) { h1Calls++ }
39+
h2 = func(*soc.SOC) { h2Calls++ }
40+
h3 = func(*soc.SOC) { h3Calls++ }
5441
)
5542
_ = g.Subscribe(address1, h1)
5643
_ = g.Subscribe(address2, h2)
@@ -68,8 +55,6 @@ func TestRegister(t *testing.T) {
6855
// trigger soc upload on address1, check that only h1 is called
6956
g.Handle(socCh1)
7057

71-
waitHandlerCallback(t, &msgChan, 1)
72-
7358
ensureCalls(t, &h1Calls, 1)
7459
ensureCalls(t, &h2Calls, 0)
7560

@@ -78,8 +63,6 @@ func TestRegister(t *testing.T) {
7863

7964
g.Handle(socCh1)
8065

81-
waitHandlerCallback(t, &msgChan, 2)
82-
8366
ensureCalls(t, &h1Calls, 2)
8467
ensureCalls(t, &h2Calls, 0)
8568
ensureCalls(t, &h3Calls, 1)
@@ -88,16 +71,12 @@ func TestRegister(t *testing.T) {
8871

8972
g.Handle(socCh1)
9073

91-
waitHandlerCallback(t, &msgChan, 1)
92-
9374
ensureCalls(t, &h1Calls, 3)
9475
ensureCalls(t, &h2Calls, 0)
9576
ensureCalls(t, &h3Calls, 1)
9677

9778
g.Handle(socCh2)
9879

99-
waitHandlerCallback(t, &msgChan, 1)
100-
10180
ensureCalls(t, &h1Calls, 3)
10281
ensureCalls(t, &h2Calls, 1)
10382
ensureCalls(t, &h3Calls, 1)
@@ -110,15 +89,3 @@ func ensureCalls(t *testing.T, calls *int, exp int) {
11089
t.Fatalf("expected %d calls, found %d", exp, *calls)
11190
}
11291
}
113-
114-
func waitHandlerCallback(t *testing.T, msgChan *chan struct{}, count int) {
115-
t.Helper()
116-
117-
for range count {
118-
select {
119-
case <-*msgChan:
120-
case <-time.After(1 * time.Second):
121-
t.Fatal("reached timeout while waiting for handler message")
122-
}
123-
}
124-
}

0 commit comments

Comments
 (0)