pico

chore(pubsub): add more tests open rss
created pr with 105.1 on 2026-01-25T17:03:08Z · by c8ef7d19
added 105.2 on 2026-01-27T04:11:48Z · by c8ef7d19
1: 603ca6e = 1: 603ca6e chore(pubsub): add more tests
2: 501c042 ! 2: 17e00b2 feat(pubsub): round robin
added 105.3 on 2026-01-29T01:03:33Z · by c8ef7d19
1: 603ca6e = 1: 603ca6e chore(pubsub): add more tests
2: 17e00b2 = 2: 17e00b2 feat(pubsub): round robin
-: ------- > 3: e3136bd fix(pubsub): check for eof before processing and skip empty byte reads
-: ------- > 4: 9a6d19e fix: rr
-: ------- > 5: 5b3f3a1 fix: sending 0 byte read
added 105.4 on 2026-02-01T17:16:43Z · by c8ef7d19
1: 603ca6e = 1: 603ca6e chore(pubsub): add more tests
2: 17e00b2 = 2: 17e00b2 feat(pubsub): round robin
3: e3136bd = 3: e3136bd fix(pubsub): check for eof before processing and skip empty byte reads
4: 9a6d19e = 4: 9a6d19e fix: rr
5: 5b3f3a1 = 5: 5b3f3a1 fix: sending 0 byte read
-: ------- > 6: 4fff471 refactor: fixes
-: ------- > 7: d0dfc85 chore: SetDispatch on Broker
changed status to accepted on 2026-02-23T02:02:07Z · by c8ef7d19
cmds
checkout latest patchset: 
ssh pr.pico.sh print 105 | git am -3
checkout any patchset in a patch request: 
ssh pr.pico.sh print 105.[rev] | git am -3
add changes to patch request: 
git format-patch main --stdout | ssh pr.pico.sh pr add 105
set PR to open (enables RSS notifications): 
ssh pr.pico.sh pr open 105
set PR to draft (stops RSS notifications): 
ssh pr.pico.sh pr draft 105

Patchset 105.2 on 2026-01-27T04:11:48Z · commit 603ca6e

→ chore(pubsub): add more tests
Eric Bower 2026-01-25T16:38:25Z
 feat(pubsub): round robin
Eric Bower 2026-01-25T16:38:25Z
+408 -0 pkg/pubsub/regression_test.go #
......@@ -0,0 +1,408 @@
1+package pubsub
2+
3+import (
4+ "bytes"
5+ "context"
6+ "fmt"
7+ "log/slog"
8+ "sync"
9+ "sync/atomic"
10+ "testing"
11+ "time"
12+)
13+
14+// TestChannelMessageOrdering verifies that messages are delivered without panics or corruption.
15+// This applies to both multicast and round-robin dispatchers.
16+func TestChannelMessageOrdering(t *testing.T) {
17+ name := "order-test"
18+ numMessages := 3
19+
20+ // Test with Multicast
21+ t.Run("Multicast", func(t *testing.T) {
22+ cast := NewMulticast(slog.Default())
23+ buf := new(Buffer)
24+ channel := NewChannel(name)
25+
26+ var wg sync.WaitGroup
27+ syncer := make(chan int)
28+
29+ // Subscribe
30+ wg.Add(1)
31+ go func() {
32+ defer wg.Done()
33+ syncer <- 0
34+ _ = cast.Sub(context.TODO(), "sub", buf, []*Channel{channel}, false)
35+ }()
36+
37+ <-syncer
38+
39+ // Publish messages
40+ for i := 0; i < numMessages; i++ {
41+ wg.Add(1)
42+ idx := i
43+ go func() {
44+ defer wg.Done()
45+ msg := fmt.Sprintf("msg%d\n", idx)
46+ _ = cast.Pub(context.TODO(), "pub", &Buffer{b: *bytes.NewBufferString(msg)}, []*Channel{channel}, false)
47+ }()
48+ }
49+
50+ wg.Wait()
51+
52+ // Verify at least some messages were received
53+ content := buf.String()
54+ if len(content) == 0 {
55+ t.Error("Multicast: no messages received")
56+ }
57+ })
58+
59+ // Test with RoundRobin
60+ t.Run("RoundRobin", func(t *testing.T) {
61+ rr := NewRoundRobin(slog.Default())
62+ buf := new(Buffer)
63+ channel := NewChannel(name)
64+
65+ var wg sync.WaitGroup
66+ syncer := make(chan int)
67+
68+ // Subscribe
69+ wg.Add(1)
70+ go func() {
71+ defer wg.Done()
72+ syncer <- 0
73+ _ = rr.Sub(context.TODO(), "sub", buf, []*Channel{channel}, false)
74+ }()
75+
76+ <-syncer
77+
78+ // Publish messages
79+ for i := 0; i < numMessages; i++ {
80+ wg.Add(1)
81+ idx := i
82+ go func() {
83+ defer wg.Done()
84+ msg := fmt.Sprintf("msg%d\n", idx)
85+ _ = rr.Pub(context.TODO(), "pub", &Buffer{b: *bytes.NewBufferString(msg)}, []*Channel{channel}, false)
86+ }()
87+ }
88+
89+ wg.Wait()
90+
91+ // Verify at least some messages were received
92+ content := buf.String()
93+ if len(content) == 0 {
94+ t.Error("RoundRobin: no messages received")
95+ }
96+ })
97+}
98+
99+// TestDispatcherClientDirection verifies that both dispatchers respect client direction.
100+// Publishers should not receive messages they publish.
101+func TestDispatcherClientDirection(t *testing.T) {
102+ name := "direction-test"
103+
104+ t.Run("Multicast", func(t *testing.T) {
105+ cast := NewMulticast(slog.Default())
106+ pubBuf := new(Buffer)
107+ subBuf := new(Buffer)
108+ channel := NewChannel(name)
109+
110+ var wg sync.WaitGroup
111+
112+ // Publisher (input only)
113+ wg.Add(1)
114+ go func() {
115+ defer wg.Done()
116+ _ = cast.Pub(context.TODO(), "pub", &Buffer{b: *bytes.NewBufferString("test")}, []*Channel{channel}, false)
117+ }()
118+
119+ // Subscriber (output only)
120+ wg.Add(1)
121+ go func() {
122+ defer wg.Done()
123+ _ = cast.Sub(context.TODO(), "sub", subBuf, []*Channel{channel}, false)
124+ }()
125+
126+ wg.Wait()
127+
128+ // Publisher should not receive the message
129+ if pubBuf.String() != "" {
130+ t.Errorf("Publisher received message: %q", pubBuf.String())
131+ }
132+
133+ // Subscriber should receive it
134+ if subBuf.String() != "test" {
135+ t.Errorf("Subscriber should have received message, got: %q", subBuf.String())
136+ }
137+ })
138+
139+ t.Run("RoundRobin", func(t *testing.T) {
140+ rr := NewRoundRobin(slog.Default())
141+ pubBuf := new(Buffer)
142+ subBuf := new(Buffer)
143+ channel := NewChannel(name)
144+
145+ var wg sync.WaitGroup
146+
147+ // Publisher (input only)
148+ wg.Add(1)
149+ go func() {
150+ defer wg.Done()
151+ _ = rr.Pub(context.TODO(), "pub", &Buffer{b: *bytes.NewBufferString("test")}, []*Channel{channel}, false)
152+ }()
153+
154+ // Subscriber (output only)
155+ wg.Add(1)
156+ go func() {
157+ defer wg.Done()
158+ _ = rr.Sub(context.TODO(), "sub", subBuf, []*Channel{channel}, false)
159+ }()
160+
161+ wg.Wait()
162+
163+ // Publisher should not receive the message
164+ if pubBuf.String() != "" {
165+ t.Errorf("Publisher received message: %q", pubBuf.String())
166+ }
167+
168+ // Subscriber should receive it
169+ if subBuf.String() != "test" {
170+ t.Errorf("Subscriber should have received message, got: %q", subBuf.String())
171+ }
172+ })
173+}
174+
175+// TestChannelConcurrentPublishes verifies that concurrent publishes don't cause races or data loss.
176+func TestChannelConcurrentPublishes(t *testing.T) {
177+ name := "concurrent-test"
178+ numPublishers := 10
179+ msgsPerPublisher := 5
180+ numSubscribers := 3
181+
182+ t.Run("Multicast", func(t *testing.T) {
183+ cast := NewMulticast(slog.Default())
184+ buffers := make([]*Buffer, numSubscribers)
185+ for i := range buffers {
186+ buffers[i] = new(Buffer)
187+ }
188+ channel := NewChannel(name)
189+
190+ var wg sync.WaitGroup
191+
192+ // Subscribe
193+ for i := range buffers {
194+ wg.Add(1)
195+ idx := i
196+ go func() {
197+ defer wg.Done()
198+ _ = cast.Sub(context.TODO(), fmt.Sprintf("sub-%d", idx), buffers[idx], []*Channel{channel}, false)
199+ }()
200+ }
201+ time.Sleep(100 * time.Millisecond)
202+
203+ // Concurrent publishers
204+ pubCount := int32(0)
205+ for p := 0; p < numPublishers; p++ {
206+ pubID := p
207+ for m := 0; m < msgsPerPublisher; m++ {
208+ wg.Add(1)
209+ msgNum := m
210+ go func() {
211+ defer wg.Done()
212+ msg := fmt.Sprintf("pub%d-msg%d\n", pubID, msgNum)
213+ _ = cast.Pub(context.TODO(), fmt.Sprintf("pub-%d", pubID), &Buffer{b: *bytes.NewBufferString(msg)}, []*Channel{channel}, false)
214+ atomic.AddInt32(&pubCount, 1)
215+ }()
216+ }
217+ }
218+
219+ wg.Wait()
220+
221+ // Verify all messages delivered to all subscribers
222+ totalExpectedMessages := numPublishers * msgsPerPublisher
223+ for i, buf := range buffers {
224+ messageCount := bytes.Count([]byte(buf.String()), []byte("\n"))
225+ if messageCount != totalExpectedMessages {
226+ t.Errorf("Subscriber %d: expected %d messages, got %d", i, totalExpectedMessages, messageCount)
227+ }
228+ }
229+
230+ // Verify all publishes completed
231+ if pubCount != int32(totalExpectedMessages) {
232+ t.Errorf("Expected %d publishes to complete, got %d", totalExpectedMessages, pubCount)
233+ }
234+ })
235+
236+ t.Run("RoundRobin", func(t *testing.T) {
237+ rr := NewRoundRobin(slog.Default())
238+ buffers := make([]*Buffer, numSubscribers)
239+ for i := range buffers {
240+ buffers[i] = new(Buffer)
241+ }
242+ channel := NewChannel(name)
243+
244+ var wg sync.WaitGroup
245+
246+ // Subscribe
247+ for i := range buffers {
248+ wg.Add(1)
249+ idx := i
250+ go func() {
251+ defer wg.Done()
252+ _ = rr.Sub(context.TODO(), fmt.Sprintf("sub-%d", idx), buffers[idx], []*Channel{channel}, false)
253+ }()
254+ }
255+ time.Sleep(100 * time.Millisecond)
256+
257+ // Concurrent publishers
258+ pubCount := int32(0)
259+ for p := 0; p < numPublishers; p++ {
260+ pubID := p
261+ for m := 0; m < msgsPerPublisher; m++ {
262+ wg.Add(1)
263+ msgNum := m
264+ go func() {
265+ defer wg.Done()
266+ msg := fmt.Sprintf("pub%d-msg%d\n", pubID, msgNum)
267+ _ = rr.Pub(context.TODO(), fmt.Sprintf("pub-%d", pubID), &Buffer{b: *bytes.NewBufferString(msg)}, []*Channel{channel}, false)
268+ atomic.AddInt32(&pubCount, 1)
269+ }()
270+ }
271+ }
272+
273+ wg.Wait()
274+
275+ // Verify all messages distributed (one to each subscriber per round-robin cycle)
276+ // Allow for some timing variance - expect at least 90% of messages
277+ totalExpectedMessages := numPublishers * msgsPerPublisher
278+ totalDelivered := int32(0)
279+ for _, buf := range buffers {
280+ messageCount := bytes.Count([]byte(buf.String()), []byte("\n"))
281+ totalDelivered += int32(messageCount)
282+ }
283+ minExpected := int32(float32(totalExpectedMessages) * 0.9)
284+ if totalDelivered < minExpected {
285+ t.Errorf("Expected at least %d messages, got %d", minExpected, totalDelivered)
286+ }
287+
288+ // Verify all publishes completed
289+ if pubCount != int32(totalExpectedMessages) {
290+ t.Errorf("Expected %d publishes to complete, got %d", totalExpectedMessages, pubCount)
291+ }
292+ })
293+}
294+
295+// TestDispatcherEmptySubscribers verifies that dispatchers handle empty subscriber set without panic.
296+func TestDispatcherEmptySubscribers(t *testing.T) {
297+ name := "empty-subs-test"
298+
299+ t.Run("Multicast", func(t *testing.T) {
300+ cast := NewMulticast(slog.Default())
301+ channel := NewChannel(name)
302+
303+ var wg sync.WaitGroup
304+
305+ // Publish with no subscribers (should not panic)
306+ wg.Add(1)
307+ go func() {
308+ defer wg.Done()
309+ defer func() {
310+ if r := recover(); r != nil {
311+ t.Errorf("Multicast panicked with no subscribers: %v", r)
312+ }
313+ }()
314+ _ = cast.Pub(context.TODO(), "pub", &Buffer{b: *bytes.NewBufferString("test")}, []*Channel{channel}, false)
315+ }()
316+
317+ wg.Wait()
318+ t.Log("Multicast handled empty subscribers correctly")
319+ })
320+
321+ t.Run("RoundRobin", func(t *testing.T) {
322+ rr := NewRoundRobin(slog.Default())
323+ channel := NewChannel(name)
324+
325+ var wg sync.WaitGroup
326+
327+ // Publish with no subscribers (should not panic)
328+ wg.Add(1)
329+ go func() {
330+ defer wg.Done()
331+ defer func() {
332+ if r := recover(); r != nil {
333+ t.Errorf("RoundRobin panicked with no subscribers: %v", r)
334+ }
335+ }()
336+ _ = rr.Pub(context.TODO(), "pub", &Buffer{b: *bytes.NewBufferString("test")}, []*Channel{channel}, false)
337+ }()
338+
339+ wg.Wait()
340+ t.Log("RoundRobin handled empty subscribers correctly")
341+ })
342+}
343+
344+// TestDispatcherSingleSubscriber verifies that both dispatchers work correctly with one subscriber.
345+func TestDispatcherSingleSubscriber(t *testing.T) {
346+ name := "single-sub-test"
347+ message := "single-sub-message"
348+
349+ t.Run("Multicast", func(t *testing.T) {
350+ cast := NewMulticast(slog.Default())
351+ buf := new(Buffer)
352+ channel := NewChannel(name)
353+
354+ var wg sync.WaitGroup
355+
356+ // Subscribe
357+ wg.Add(1)
358+ go func() {
359+ defer wg.Done()
360+ _ = cast.Sub(context.TODO(), "sub", buf, []*Channel{channel}, false)
361+ }()
362+
363+ time.Sleep(100 * time.Millisecond)
364+
365+ // Publish
366+ wg.Add(1)
367+ go func() {
368+ defer wg.Done()
369+ _ = cast.Pub(context.TODO(), "pub", &Buffer{b: *bytes.NewBufferString(message)}, []*Channel{channel}, false)
370+ }()
371+
372+ wg.Wait()
373+
374+ if buf.String() != message {
375+ t.Errorf("Multicast with single subscriber: expected %q, got %q", message, buf.String())
376+ }
377+ })
378+
379+ t.Run("RoundRobin", func(t *testing.T) {
380+ rr := NewRoundRobin(slog.Default())
381+ buf := new(Buffer)
382+ channel := NewChannel(name)
383+
384+ var wg sync.WaitGroup
385+
386+ // Subscribe
387+ wg.Add(1)
388+ go func() {
389+ defer wg.Done()
390+ _ = rr.Sub(context.TODO(), "sub", buf, []*Channel{channel}, false)
391+ }()
392+
393+ time.Sleep(100 * time.Millisecond)
394+
395+ // Publish
396+ wg.Add(1)
397+ go func() {
398+ defer wg.Done()
399+ _ = rr.Pub(context.TODO(), "pub", &Buffer{b: *bytes.NewBufferString(message)}, []*Channel{channel}, false)
400+ }()
401+
402+ wg.Wait()
403+
404+ if buf.String() != message {
405+ t.Errorf("RoundRobin with single subscriber: expected %q, got %q", message, buf.String())
406+ }
407+ })
408+}
Back to top