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 17e00b2

chore(pubsub): add more tests
Eric Bower 2026-01-25T16:38:25Z
 → feat(pubsub): round robin
Eric Bower 2026-01-25T16:38:25Z
+12 -3 pkg/apps/pipe/cli.go #
......@@ -600,6 +600,7 @@ func (handler *CliHandler) pub(cmd *CliCmd, topic string, clientID string) error
600600 block := pubCmd.Bool("b", true, "Block writes until a subscriber is available")
601601 timeout := pubCmd.Duration("t", 30*24*time.Hour, "Timeout as a Go duration to block for a subscriber to be available. Valid time units are 'ns', 'us' (or 'µs'), 'ms', 's', 'm', 'h'. Default is 30 days.")
602602 clean := pubCmd.Bool("c", false, "Don't send status messages")
603+ broker := pubCmd.String("bk", "multicast", "Type of broker (e.g. multicast, round_robin)")
603604
604605 if !flagCheck(pubCmd, topic, cmd.args) {
605606 return fmt.Errorf("invalid cmd args")
......@@ -619,6 +620,7 @@ func (handler *CliHandler) pub(cmd *CliCmd, topic string, clientID string) error
619620 "topic", topic,
620621 "access", *access,
621622 "clean", *clean,
623+ "broker", *broker,
622624 )
623625
624626 var accessList []string
......@@ -795,13 +797,20 @@ func (handler *CliHandler) pub(cmd *CliCmd, topic string, clientID string) error
795797
796798 throttledRW := newThrottledMonitorRW(rw, handler, cmd, name)
797799
800+ var bk psub.MessageDispatcher
801+ bk = &psub.MulticastDispatcher{}
802+ if *broker == "round_robin" {
803+ fmt.Println("BROKER ROUND ROBIN")
804+ bk = &psub.RoundRobinDispatcher{}
805+ }
806+ channel := psub.NewChannel(name)
807+ channel.Dispatcher = bk
808+
798809 err := handler.PubSub.Pub(
799810 cmd.pipeCtx,
800811 clientID,
801812 throttledRW,
802- []*psub.Channel{
803- psub.NewChannel(name),
804- },
813+ []*psub.Channel{channel},
805814 *block,
806815 )
807816
+5 -0 pkg/pubsub/broker.go #
......@@ -196,6 +196,11 @@ func (b *BaseBroker) Connect(client *Client, channels []*Channel) (error, error)
196196
197197 func (b *BaseBroker) ensureChannel(channel *Channel) *Channel {
198198 dataChannel, _ := b.Channels.LoadOrStore(channel.Topic, channel)
199+ // Allow overwriting the dispatcher
200+ if channel.Dispatcher != nil && dataChannel.Dispatcher == nil {
201+ dataChannel.Dispatcher = channel.Dispatcher
202+ }
203+
199204 dataChannel.Handle()
200205 return dataChannel
201206 }
+14 -22 pkg/pubsub/channel.go #
......@@ -57,6 +57,7 @@ type Channel struct {
5757 Clients *syncmap.Map[string, *Client]
5858 handleOnce sync.Once
5959 cleanupOnce sync.Once
60+ Dispatcher MessageDispatcher
6061 }
6162
6263 func (c *Channel) GetClients() iter.Seq2[string, *Client] {
......@@ -83,30 +84,21 @@ func (c *Channel) Handle() {
8384 case <-c.Done:
8485 return
8586 case data, ok := <-c.Data:
86- var wg sync.WaitGroup
87- for _, client := range c.GetClients() {
88- if client.Direction == ChannelDirectionInput || (client.ID == data.ClientID && !client.Replay) {
89- continue
87+ if !ok {
88+ // Channel is closing, close all client data channels
89+ for _, client := range c.GetClients() {
90+ client.onceData.Do(func() {
91+ close(client.Data)
92+ })
9093 }
91-
92- wg.Add(1)
93- go func() {
94- defer wg.Done()
95- if !ok {
96- client.onceData.Do(func() {
97- close(client.Data)
98- })
99- return
100- }
101-
102- select {
103- case client.Data <- data:
104- case <-client.Done:
105- case <-c.Done:
106- }
107- }()
94+ return
10895 }
109- wg.Wait()
96+
97+ // Collect eligible subscribers
98+ subscribers := dispatcherForGetClients(c.GetClients(), data)
99+
100+ // Dispatch message using the configured dispatcher
101+ _ = c.Dispatcher.Dispatch(data, subscribers, c.Done)
110102 }
111103 }
112104 }()
+26 -0 pkg/pubsub/dispatcher.go #
......@@ -0,0 +1,26 @@
1+package pubsub
2+
3+import (
4+ "iter"
5+)
6+
7+// MessageDispatcher defines how messages are dispatched to subscribers.
8+type MessageDispatcher interface {
9+ // Dispatch sends a message to the appropriate subscriber(s).
10+ // It receives the message, all subscribers, and the channel's sync primitives.
11+ Dispatch(msg ChannelMessage, subscribers []*Client, channelDone chan struct{}) error
12+}
13+
14+// dispatcherForGetClients collects eligible clients for dispatching.
15+// Returns clients that should receive messages (output direction, not the sender unless replay).
16+func dispatcherForGetClients(getClients iter.Seq2[string, *Client], msg ChannelMessage) []*Client {
17+ subscribers := make([]*Client, 0)
18+ for _, client := range getClients {
19+ // Skip input-only clients and senders (unless replay is enabled)
20+ if client.Direction == ChannelDirectionInput || (client.ID == msg.ClientID && !client.Replay) {
21+ continue
22+ }
23+ subscribers = append(subscribers, client)
24+ }
25+ return subscribers
26+}
+21 -0 pkg/pubsub/multicast.go #
......@@ -6,6 +6,7 @@ import (
66 "io"
77 "iter"
88 "log/slog"
9+ "sync"
910
1011 "github.com/antoniomika/syncmap"
1112 )
......@@ -81,4 +82,24 @@ func (p *Multicast) Sub(ctx context.Context, ID string, rw io.ReadWriter, channe
8182 return errors.Join(p.connect(ctx, ID, rw, channels, ChannelDirectionOutput, false, false, keepAlive))
8283 }
8384
85+// MulticastDispatcher sends each message to all eligible subscribers.
86+type MulticastDispatcher struct{}
87+
88+func (d *MulticastDispatcher) Dispatch(msg ChannelMessage, subscribers []*Client, channelDone chan struct{}) error {
89+ var wg sync.WaitGroup
90+ for _, client := range subscribers {
91+ wg.Add(1)
92+ go func(cl *Client) {
93+ defer wg.Done()
94+ select {
95+ case cl.Data <- msg:
96+ case <-cl.Done:
97+ case <-channelDone:
98+ }
99+ }(client)
100+ }
101+ wg.Wait()
102+ return nil
103+}
104+
84105 var _ PubSub = (*Multicast)(nil)
+51 -0 pkg/pubsub/roundrobin.go #
......@@ -0,0 +1,51 @@
1+package pubsub
2+
3+import (
4+ "slices"
5+ "strings"
6+ "sync"
7+)
8+
9+/*
10+RoundRobin is a load-balancing broker that distributes published messages
11+to subscribers using a round-robin algorithm.
12+
13+Unlike Multicast which sends each message to all subscribers, RoundRobin
14+sends each message to exactly one subscriber, rotating through the available
15+subscribers for each published message. This provides load balancing for
16+message processing.
17+
18+It maintains independent round-robin state per channel/topic.
19+*/
20+type RoundRobinDispatcher struct {
21+ index uint32
22+ mu sync.Mutex
23+}
24+
25+func (d *RoundRobinDispatcher) Dispatch(msg ChannelMessage, subscribers []*Client, channelDone chan struct{}) error {
26+ // If no subscribers, nothing to dispatch
27+ // BlockWrite behavior at publish time ensures subscribers are present when needed
28+ if len(subscribers) == 0 {
29+ return nil
30+ }
31+
32+ slices.SortFunc(subscribers, func(a, b *Client) int {
33+ return strings.Compare(a.ID, b.ID)
34+ })
35+
36+ // Select the next subscriber in round-robin order
37+ d.mu.Lock()
38+ selectedIdx := int(d.index % uint32(len(subscribers)))
39+ d.index++
40+ d.mu.Unlock()
41+
42+ selectedClient := subscribers[selectedIdx]
43+
44+ select {
45+ case selectedClient.Data <- msg:
46+ case <-selectedClient.Done:
47+ case <-channelDone:
48+ }
49+
50+ return nil
51+}
+354 -0 pkg/pubsub/roundrobin_test.go #
......@@ -0,0 +1,354 @@
1+package pubsub
2+
3+import (
4+ "bytes"
5+ "context"
6+ "fmt"
7+ "log/slog"
8+ "sync"
9+ "testing"
10+ "time"
11+)
12+
13+func TestRoundRobinSingleSub(t *testing.T) {
14+ // Single publisher, single subscriber
15+ // Should work like normal pub/sub
16+ orderActual := ""
17+ orderExpected := "sub-pub-"
18+ actual := new(Buffer)
19+ expected := "some test data"
20+ name := "test-channel"
21+ syncer := make(chan int)
22+
23+ rr := NewRoundRobin(slog.Default())
24+
25+ var wg sync.WaitGroup
26+ wg.Add(2)
27+
28+ channel := NewChannel(name)
29+
30+ go func() {
31+ orderActual += "sub-"
32+ syncer <- 0
33+ fmt.Println(rr.Sub(context.TODO(), "1", actual, []*Channel{channel}, false))
34+ wg.Done()
35+ }()
36+
37+ <-syncer
38+
39+ go func() {
40+ orderActual += "pub-"
41+ fmt.Println(rr.Pub(context.TODO(), "2", &Buffer{b: *bytes.NewBufferString(expected)}, []*Channel{channel}, true))
42+ wg.Done()
43+ }()
44+
45+ wg.Wait()
46+
47+ if orderActual != orderExpected {
48+ t.Fatalf("\norderActual:(%s)\norderExpected:(%s)", orderActual, orderExpected)
49+ }
50+ if actual.String() != expected {
51+ t.Fatalf("\nactual:(%s)\nexpected:(%s)", actual.String(), expected)
52+ }
53+}
54+
55+func TestRoundRobinMultipleSubs(t *testing.T) {
56+ // Single publisher, multiple subscribers
57+ // Verify round-robin distributes across subscribers
58+ name := "test-channel"
59+
60+ rr := NewRoundRobin(slog.Default())
61+
62+ buffers := []*Buffer{new(Buffer), new(Buffer), new(Buffer)}
63+ channel := NewChannel(name)
64+
65+ var wg sync.WaitGroup
66+
67+ // Subscribe three clients sequentially with sync point
68+ syncer := make(chan int, 3)
69+ for i := range buffers {
70+ idx := i
71+ wg.Add(1)
72+ go func() {
73+ defer wg.Done()
74+ clientID := fmt.Sprintf("sub-%d", idx)
75+ _ = rr.Sub(context.TODO(), clientID, buffers[idx], []*Channel{channel}, false)
76+ }()
77+ syncer <- i
78+ }
79+
80+ // Wait for all subscribers to connect
81+ for i := 0; i < 3; i++ {
82+ <-syncer
83+ }
84+ time.Sleep(200 * time.Millisecond)
85+
86+ // Publish many messages to verify distribution
87+ numMsgs := 9
88+ for i := 0; i < numMsgs; i++ {
89+ wg.Add(1)
90+ idx := i
91+ go func() {
92+ defer wg.Done()
93+ msg := fmt.Sprintf("msg%d\n", idx)
94+ _ = rr.Pub(context.TODO(), "pub", &Buffer{b: *bytes.NewBufferString(msg)}, []*Channel{channel}, false)
95+ }()
96+ }
97+
98+ wg.Wait()
99+
100+ // Verify that messages were distributed across all subscribers
101+ for i, buf := range buffers {
102+ content := buf.String()
103+ t.Logf("sub-%d received %d bytes", i, len(content))
104+ if len(content) == 0 {
105+ t.Logf("WARNING: sub-%d received no messages", i)
106+ }
107+ }
108+
109+ // At least one subscriber should have received messages
110+ totalLen := 0
111+ for _, buf := range buffers {
112+ totalLen += len(buf.String())
113+ }
114+ if totalLen == 0 {
115+ t.Fatal("No messages were delivered to any subscriber")
116+ }
117+}
118+
119+func TestRoundRobinDistribution(t *testing.T) {
120+ // Verify that messages are distributed evenly across subscribers
121+ expected := "msg"
122+ name := "test-channel"
123+ numSubs := 3
124+ numMessages := 9
125+
126+ rr := NewRoundRobin(slog.Default())
127+
128+ buffers := make([]*Buffer, numSubs)
129+ for i := 0; i < numSubs; i++ {
130+ buffers[i] = new(Buffer)
131+ }
132+ channels := []*Channel{NewChannel(name)}
133+
134+ var wg sync.WaitGroup
135+
136+ // Subscribe clients
137+ for i := 0; i < numSubs; i++ {
138+ wg.Add(1)
139+ idx := i
140+ go func() {
141+ defer wg.Done()
142+ clientID := fmt.Sprintf("sub-%d", idx)
143+ _ = rr.Sub(context.TODO(), clientID, buffers[idx], channels, false)
144+ }()
145+ }
146+
147+ time.Sleep(100 * time.Millisecond)
148+
149+ // Publish multiple messages
150+ for i := 0; i < numMessages; i++ {
151+ wg.Add(1)
152+ msgIdx := i
153+ go func() {
154+ defer wg.Done()
155+ msg := fmt.Sprintf("%s%d\n", expected, msgIdx)
156+ _ = rr.Pub(context.TODO(), "pub", &Buffer{b: *bytes.NewBufferString(msg)}, channels, false)
157+ }()
158+ }
159+
160+ wg.Wait()
161+
162+ // Count messages per subscriber
163+ msgCounts := make(map[int]int)
164+ for i, buf := range buffers {
165+ // Count occurrences of "msg" in the buffer
166+ content := buf.String()
167+ count := 0
168+ for j := 0; j < numMessages; j++ {
169+ marker := fmt.Sprintf("msg%d", j)
170+ if bytes.Contains([]byte(content), []byte(marker)) {
171+ count++
172+ }
173+ }
174+ msgCounts[i] = count
175+ t.Logf("sub-%d received %d messages", i, count)
176+ }
177+
178+ // Verify relatively even distribution (within 1 message difference due to concurrency)
179+ minCount := msgCounts[0]
180+ maxCount := msgCounts[0]
181+ for i := 1; i < numSubs; i++ {
182+ if msgCounts[i] < minCount {
183+ minCount = msgCounts[i]
184+ }
185+ if msgCounts[i] > maxCount {
186+ maxCount = msgCounts[i]
187+ }
188+ }
189+
190+ if maxCount-minCount > 2 {
191+ t.Fatalf("Uneven distribution: min=%d, max=%d, difference=%d", minCount, maxCount, maxCount-minCount)
192+ }
193+}
194+
195+func TestRoundRobinSubscriberJoinLeave(t *testing.T) {
196+ // Test behavior when subscribers join and leave mid-stream
197+ // Verify the broker gracefully handles subscriber changes
198+ name := "test-channel"
199+
200+ rr := NewRoundRobin(slog.Default())
201+ channel := NewChannel(name)
202+
203+ buf1 := new(Buffer)
204+ buf2 := new(Buffer)
205+ buf3 := new(Buffer)
206+
207+ var wg sync.WaitGroup
208+
209+ ctx1, cancel1 := context.WithCancel(context.Background())
210+ ctx2, cancel2 := context.WithCancel(context.Background())
211+
212+ // Start with 2 subscribers
213+ wg.Add(2)
214+ go func() {
215+ defer wg.Done()
216+ _ = rr.Sub(ctx1, "sub-1", buf1, []*Channel{channel}, false)
217+ }()
218+ go func() {
219+ defer wg.Done()
220+ _ = rr.Sub(ctx2, "sub-2", buf2, []*Channel{channel}, false)
221+ }()
222+
223+ time.Sleep(200 * time.Millisecond)
224+
225+ // Publish some messages with 2 subscribers
226+ for i := 0; i < 2; i++ {
227+ wg.Add(1)
228+ idx := i
229+ go func() {
230+ defer wg.Done()
231+ msg := fmt.Sprintf("msg%d\n", idx)
232+ _ = rr.Pub(context.TODO(), "pub", &Buffer{b: *bytes.NewBufferString(msg)}, []*Channel{channel}, false)
233+ }()
234+ }
235+ time.Sleep(100 * time.Millisecond)
236+
237+ // Remove sub-1
238+ cancel1()
239+ time.Sleep(200 * time.Millisecond)
240+
241+ // Add sub-3
242+ ctx3, cancel3 := context.WithCancel(context.Background())
243+ wg.Add(1)
244+ go func() {
245+ defer wg.Done()
246+ _ = rr.Sub(ctx3, "sub-3", buf3, []*Channel{channel}, false)
247+ }()
248+
249+ time.Sleep(200 * time.Millisecond)
250+
251+ // Publish more messages with different subscriber set
252+ for i := 2; i < 4; i++ {
253+ wg.Add(1)
254+ idx := i
255+ go func() {
256+ defer wg.Done()
257+ msg := fmt.Sprintf("msg%d\n", idx)
258+ _ = rr.Pub(context.TODO(), "pub", &Buffer{b: *bytes.NewBufferString(msg)}, []*Channel{channel}, false)
259+ }()
260+ }
261+
262+ wg.Wait()
263+ cancel2()
264+ cancel3()
265+
266+ t.Logf("sub-1: %d bytes", len(buf1.String()))
267+ t.Logf("sub-2: %d bytes", len(buf2.String()))
268+ t.Logf("sub-3: %d bytes", len(buf3.String()))
269+
270+ // Verify that messages were delivered (exact distribution depends on timing)
271+ totalLen := len(buf1.String()) + len(buf2.String()) + len(buf3.String())
272+ if totalLen == 0 {
273+ t.Fatal("No messages were delivered after subscriber changes")
274+ }
275+}
276+
277+func TestRoundRobinMultipleChannels(t *testing.T) {
278+ // Test that each channel maintains independent round-robin state
279+ rr := NewRoundRobin(slog.Default())
280+
281+ ch1 := NewChannel("topic-1")
282+ ch2 := NewChannel("topic-2")
283+
284+ buf1ch1 := new(Buffer)
285+ buf2ch1 := new(Buffer)
286+ buf1ch2 := new(Buffer)
287+ buf2ch2 := new(Buffer)
288+
289+ var wg sync.WaitGroup
290+
291+ // Subscribe to channel 1
292+ wg.Add(2)
293+ go func() {
294+ defer wg.Done()
295+ _ = rr.Sub(context.TODO(), "sub-1-ch1", buf1ch1, []*Channel{ch1}, false)
296+ }()
297+ go func() {
298+ defer wg.Done()
299+ _ = rr.Sub(context.TODO(), "sub-2-ch1", buf2ch1, []*Channel{ch1}, false)
300+ }()
301+
302+ // Subscribe to channel 2
303+ wg.Add(2)
304+ go func() {
305+ defer wg.Done()
306+ _ = rr.Sub(context.TODO(), "sub-1-ch2", buf1ch2, []*Channel{ch2}, false)
307+ }()
308+ go func() {
309+ defer wg.Done()
310+ _ = rr.Sub(context.TODO(), "sub-2-ch2", buf2ch2, []*Channel{ch2}, false)
311+ }()
312+
313+ time.Sleep(100 * time.Millisecond)
314+
315+ // Publish to channel 1
316+ wg.Add(2)
317+ for i := 0; i < 2; i++ {
318+ idx := i
319+ go func() {
320+ defer wg.Done()
321+ msg := fmt.Sprintf("ch1-msg%d\n", idx)
322+ _ = rr.Pub(context.TODO(), "pub-1", &Buffer{b: *bytes.NewBufferString(msg)}, []*Channel{ch1}, false)
323+ }()
324+ }
325+
326+ // Publish to channel 2
327+ wg.Add(2)
328+ for i := 0; i < 2; i++ {
329+ idx := i
330+ go func() {
331+ defer wg.Done()
332+ msg := fmt.Sprintf("ch2-msg%d\n", idx)
333+ _ = rr.Pub(context.TODO(), "pub-2", &Buffer{b: *bytes.NewBufferString(msg)}, []*Channel{ch2}, false)
334+ }()
335+ }
336+
337+ wg.Wait()
338+
339+ t.Logf("ch1-buf1: %s", buf1ch1.String())
340+ t.Logf("ch1-buf2: %s", buf2ch1.String())
341+ t.Logf("ch2-buf1: %s", buf1ch2.String())
342+ t.Logf("ch2-buf2: %s", buf2ch2.String())
343+
344+ // Both channels should have distributed messages independently
345+ ch1Total := len(buf1ch1.String()) + len(buf2ch1.String())
346+ ch2Total := len(buf1ch2.String()) + len(buf2ch2.String())
347+
348+ if ch1Total == 0 {
349+ t.Fatal("Channel 1 should have received messages")
350+ }
351+ if ch2Total == 0 {
352+ t.Fatal("Channel 2 should have received messages")
353+ }
354+}
Back to top