pico

created pr with 103.1 on 2026-01-05T14:52:02Z · by c8ef7d19

added 103.2 on 2026-01-08T01:19:34Z · by c8ef7d19


                      1:  66cf9b9 = 1:  66cf9b9 feat(pipe): add pipe_monitors table


                      2:  83b0c70 = 2:  83b0c70 chore(pipe): add new db methods


                      3:  43aa4a1 = 3:  43aa4a1 chore(pipe): add db impl


                      4:  a75186b = 4:  a75186b chore(pipe): add tests for monitoring


                      5:  9c7fcda = 5:  9c7fcda feat(pipe): add monitor cli


                      6:  2b86dec = 6:  2b86dec feat(pipe): status and rss commands


                      7:  8df2aa2 = 7:  8df2aa2 feat(pipe): monitor pub and pipe cmd


                      8:  11a5bce = 8:  11a5bce chore(pipe): monitor help text


                      9:  670350c = 9:  670350c refactor(pipe): monitor pipes on throttle interval


                      10:  ade2b83 = 10:  ade2b83 fix: window and ping fixes


                      11:  68f5cd1 = 11:  68f5cd1 chore(pipe): add tests


                      -:  ------- > 12:  74fb40f chore(pipe): add logging stmts


                      -:  ------- > 13:  c422818 refactor(pipe): create pipe monitors history table


                      -:  ------- > 14:  e0c3a88 fix(pipe): only update monitor if within window


                      -:  ------- > 15:  9a8fbde chore(pipe): implement monitor history db interface


                      -:  ------- > 16:  f168507 feat(pipe): record historical monitors


                      -:  ------- > 17:  16df77f feat(pipe): monitor calculate uptime


                      -:  ------- > 18:  10ffdf1 feat(pipe): uptime cli command


                      -:  ------- > 19:  8231ac5 chore(pipe): monitor history stubs


                      -:  ------- > 20:  65a63a9 fix(pipe): uptime cmd

added 103.3 on 2026-02-01T17:16:24Z · by c8ef7d19


                      3:  43aa4a1 ! 1:  603ca6e chore(pubsub): add more tests


                      1:  66cf9b9 < -:  ------- feat(pipe): add pipe_monitors table


                      2:  83b0c70 < -:  ------- chore(pipe): add new db methods


                      4:  a75186b ! 2:  17e00b2 feat(pubsub): round robin


                      13:  c422818 ! 3:  e3136bd fix(pubsub): check for eof before processing and skip empty byte reads


                      -:  ------- > 4:  9a6d19e fix: rr


                      5:  9c7fcda < -:  ------- feat(pipe): add monitor cli


                      16:  f168507 ! 5:  5b3f3a1 fix: sending 0 byte read


                      6:  2b86dec < -:  ------- feat(pipe): status and rss commands


                      11:  68f5cd1 ! 6:  4fff471 refactor: fixes


                      7:  8df2aa2 < -:  ------- feat(pipe): monitor pub and pipe cmd


                      14:  e0c3a88 ! 7:  d0dfc85 chore: SetDispatch on Broker


                      8:  11a5bce < -:  ------- chore(pipe): monitor help text


                      9:  670350c < -:  ------- refactor(pipe): monitor pipes on throttle interval


                      10:  ade2b83 < -:  ------- fix: window and ping fixes


                      12:  74fb40f < -:  ------- chore(pipe): add logging stmts


                      15:  9a8fbde < -:  ------- chore(pipe): implement monitor history db interface


                      17:  16df77f < -:  ------- feat(pipe): monitor calculate uptime


                      18:  10ffdf1 < -:  ------- feat(pipe): uptime cli command


                      19:  8231ac5 < -:  ------- chore(pipe): monitor history stubs


                      20:  65a63a9 < -:  ------- fix(pipe): uptime cmd

cmds

checkout latest patchset:

ssh pr.pico.sh print 103 | git am -3

checkout any patchset in a patch request:

ssh pr.pico.sh print 103.[rev] | git am -3

add changes to patch request:

git format-patch main --stdout | ssh pr.pico.sh pr add 103

Patchset `103.3` on 2026-02-01T17:16:24Z · commit d0dfc85

chore(pubsub): add more tests

Eric Bower 2026-01-25T16:38:25Z

feat(pubsub): round robin

Eric Bower 2026-01-25T16:38:25Z

fix(pubsub): check for eof before processing and skip empty byte reads

Eric Bower 2026-01-28T01:42:34Z

fix: rr

Eric Bower 2026-01-28T01:50:04Z

fix: sending 0 byte read

Eric Bower 2026-01-28T02:50:25Z

refactor: fixes

Eric Bower 2026-01-29T02:44:01Z

→ chore: SetDispatch on Broker

Eric Bower 2026-02-01T16:37:59Z

chore: SetDispatch on Broker

Eric Bower 2026-02-01T16:37:59Z

Semantic diff summary

4 added, 6 modified, 0 signature changed, 4 removed across 5 analyzed files

pkg/apps/pipe/cli.go

function_declaration pub modified

pkg/pubsub/broker.go

chunk lines 5-11 modified
chunk lines 22-28 modified
method_declaration ensureChannel modified
method_declaration SetDispatcher added

pkg/pubsub/dispatcher.go

type_declaration MulticastDispatcher added
method_declaration Dispatch added
type_declaration RoundRobinDispatcher added

pkg/pubsub/multicast.go

chunk lines 6-11 modified
function_declaration GetSubs modified
method_declaration Dispatch removed
type_declaration MulticastDispatcher removed

pkg/pubsub/roundrobin.go

type_declaration RoundRobinDispatcher removed
method_declaration Dispatch removed

+1 -1 pkg/apps/pipe/cli.go #

......@@ -803,7 +803,7 @@ func (handler *CliHandler) pub(cmd *CliCmd, topic string, clientID string) error
		bk = &psub.RoundRobinDispatcher{}
	}
	channel := psub.NewChannel(name)
	channel.Dispatcher = bk
	_ = handler.PubSub.SetDispatcher(bk, []*psub.Channel{channel})

	err := handler.PubSub.Pub(
		cmd.pipeCtx,

+12 -5 pkg/pubsub/broker.go #

......@@ -5,6 +5,7 @@ import (
	"io"
	"iter"
	"log/slog"
	"reflect"
	"sync"
	"time"

......@@ -21,6 +22,7 @@ type Broker interface {
	GetChannels() iter.Seq2[string, *Channel]
	GetClients() iter.Seq2[string, *Client]
	Connect(*Client, []*Channel) (error, error)
	SetDispatcher(dispatcher MessageDispatcher, channels []*Channel) error
}

type BaseBroker struct {

......@@ -197,13 +199,18 @@ func (b *BaseBroker) Connect(client *Client, channels []*Channel) (error, error)

func (b *BaseBroker) ensureChannel(channel *Channel) *Channel {
	dataChannel, _ := b.Channels.LoadOrStore(channel.Topic, channel)
	// Allow overwriting the dispatcher
	if channel.Dispatcher != nil && dataChannel.Dispatcher == nil {
		dataChannel.Dispatcher = channel.Dispatcher
	}

	dataChannel.Handle()
	return dataChannel
}

func (b *BaseBroker) SetDispatcher(dispatcher MessageDispatcher, channels []*Channel) error {
	for _, channel := range channels {
		dataChannel := b.ensureChannel(channel)
		if reflect.TypeOf(dataChannel.Dispatcher) != reflect.TypeOf(dispatcher) {
			dataChannel.Dispatcher = dispatcher
		}
	}
	return nil
}

var _ Broker = (*BaseBroker)(nil)

+70 -0 pkg/pubsub/dispatcher.go #

......@@ -1,8 +1,78 @@
package pubsub

import (
	"slices"
	"strings"
	"sync"
)

// MessageDispatcher defines how messages are dispatched to subscribers.
type MessageDispatcher interface {
	// Dispatch sends a message to the appropriate subscriber(s).
	// It receives the message, all subscribers, and the channel's sync primitives.
	Dispatch(msg ChannelMessage, subscribers []*Client, channelDone chan struct{}) error
}

// MulticastDispatcher sends each message to all eligible subscribers.
type MulticastDispatcher struct{}

func (d *MulticastDispatcher) Dispatch(msg ChannelMessage, subscribers []*Client, channelDone chan struct{}) error {
	var wg sync.WaitGroup
	for _, client := range subscribers {
		wg.Add(1)
		go func(cl *Client) {
			defer wg.Done()
			select {
			case cl.Data <- msg:
			case <-cl.Done:
			case <-channelDone:
			}
		}(client)
	}
	wg.Wait()
	return nil
}

/*
RoundRobin is a load-balancing broker that distributes published messages
to subscribers using a round-robin algorithm.

Unlike Multicast which sends each message to all subscribers, RoundRobin
sends each message to exactly one subscriber, rotating through the available
subscribers for each published message. This provides load balancing for
message processing.

It maintains independent round-robin state per channel/topic.
*/
type RoundRobinDispatcher struct {
	index uint32
	mu    sync.Mutex
}

func (d *RoundRobinDispatcher) Dispatch(msg ChannelMessage, subscribers []*Client, channelDone chan struct{}) error {
	// If no subscribers, nothing to dispatch
	// BlockWrite behavior at publish time ensures subscribers are present when needed
	if len(subscribers) == 0 {
		return nil
	}

	slices.SortFunc(subscribers, func(a, b *Client) int {
		return strings.Compare(a.ID, b.ID)
	})

	// Select the next subscriber in round-robin order
	d.mu.Lock()
	selectedIdx := int(d.index % uint32(len(subscribers)))
	d.index++
	d.mu.Unlock()

	selectedClient := subscribers[selectedIdx]

	select {
	case selectedClient.Data <- msg:
	case <-selectedClient.Done:
	case <-channelDone:
	}

	return nil
}

+4 -23 pkg/pubsub/multicast.go #

......@@ -6,7 +6,6 @@ import (
	"io"
	"iter"
	"log/slog"
	"sync"

	"github.com/antoniomika/syncmap"
)

......@@ -62,9 +61,11 @@ func (p *Multicast) GetSubs() iter.Seq2[string, *Client] {
func (p *Multicast) connect(ctx context.Context, ID string, rw io.ReadWriter, channels []*Channel, direction ChannelDirection, blockWrite bool, replay, keepAlive bool, dispatcher MessageDispatcher) (error, error) {
	client := NewClient(ID, rw, direction, blockWrite, replay, keepAlive)

	// Set dispatcher on all channels
	// Set dispatcher on all channels (only if not already set)
	for _, ch := range channels {
		ch.Dispatcher = dispatcher
		if ch.Dispatcher == nil {
			ch.Dispatcher = dispatcher
		}
	}

	go func() {

......@@ -87,24 +88,4 @@ func (p *Multicast) Sub(ctx context.Context, ID string, rw io.ReadWriter, channe
	return errors.Join(p.connect(ctx, ID, rw, channels, ChannelDirectionOutput, false, false, keepAlive, &MulticastDispatcher{}))
}

// MulticastDispatcher sends each message to all eligible subscribers.
type MulticastDispatcher struct{}

func (d *MulticastDispatcher) Dispatch(msg ChannelMessage, subscribers []*Client, channelDone chan struct{}) error {
	var wg sync.WaitGroup
	for _, client := range subscribers {
		wg.Add(1)
		go func(cl *Client) {
			defer wg.Done()
			select {
			case cl.Data <- msg:
			case <-cl.Done:
			case <-channelDone:
			}
		}(client)
	}
	wg.Wait()
	return nil
}

var _ PubSub = (*Multicast)(nil)

+0 -51 pkg/pubsub/roundrobin.go #

......@@ -1,51 +0,0 @@
package pubsub

import (
	"slices"
	"strings"
	"sync"
)

/*
RoundRobin is a load-balancing broker that distributes published messages
to subscribers using a round-robin algorithm.

Unlike Multicast which sends each message to all subscribers, RoundRobin
sends each message to exactly one subscriber, rotating through the available
subscribers for each published message. This provides load balancing for
message processing.

It maintains independent round-robin state per channel/topic.
*/
type RoundRobinDispatcher struct {
	index uint32
	mu    sync.Mutex
}

func (d *RoundRobinDispatcher) Dispatch(msg ChannelMessage, subscribers []*Client, channelDone chan struct{}) error {
	// If no subscribers, nothing to dispatch
	// BlockWrite behavior at publish time ensures subscribers are present when needed
	if len(subscribers) == 0 {
		return nil
	}

	slices.SortFunc(subscribers, func(a, b *Client) int {
		return strings.Compare(a.ID, b.ID)
	})

	// Select the next subscriber in round-robin order
	d.mu.Lock()
	selectedIdx := int(d.index % uint32(len(subscribers)))
	d.index++
	d.mu.Unlock()

	selectedClient := subscribers[selectedIdx]

	select {
	case selectedClient.Data <- msg:
	case <-selectedClient.Done:
	case <-channelDone:
	}

	return nil
}

...	...		@@ -803,7 +803,7 @@ func (handler CliHandler) pub(cmd CliCmd, topic string, clientID string) error
803	803		bk = &psub.RoundRobinDispatcher{}
804	804		}
805	805		channel := psub.NewChannel(name)
806		-	channel.Dispatcher = bk
	806	+	_ = handler.PubSub.SetDispatcher(bk, []*psub.Channel{channel})
807	807
808	808		err := handler.PubSub.Pub(
809	809		cmd.pipeCtx,

...	...		@@ -197,13 +199,18 @@ func (b BaseBroker) Connect(client Client, channels []*Channel) (error, error)
197	199
198	200		func (b BaseBroker) ensureChannel(channel Channel) *Channel {
199	201		dataChannel, _ := b.Channels.LoadOrStore(channel.Topic, channel)
200		-	// Allow overwriting the dispatcher
201		-	if channel.Dispatcher != nil && dataChannel.Dispatcher == nil {
202		-	dataChannel.Dispatcher = channel.Dispatcher
203		-	}
204		-
205	202		dataChannel.Handle()
206	203		return dataChannel
207	204		}
208	205
	206	+	func (b BaseBroker) SetDispatcher(dispatcher MessageDispatcher, channels []Channel) error {
	207	+	for _, channel := range channels {
	208	+	dataChannel := b.ensureChannel(channel)
	209	+	if reflect.TypeOf(dataChannel.Dispatcher) != reflect.TypeOf(dispatcher) {
	210	+	dataChannel.Dispatcher = dispatcher
	211	+	}
	212	+	}
	213	+	return nil
	214	+	}
	215	+
209	216		var _ Broker = (*BaseBroker)(nil)

...	...		@@ -5,6 +5,7 @@ import (
5	5		"io"
6	6		"iter"
7	7		"log/slog"
	8	+	"reflect"
8	9		"sync"
9	10		"time"
10	11

...	...		@@ -21,6 +22,7 @@ type Broker interface {
21	22		GetChannels() iter.Seq2[string, *Channel]
22	23		GetClients() iter.Seq2[string, *Client]
23	24		Connect(Client, []Channel) (error, error)
	25	+	SetDispatcher(dispatcher MessageDispatcher, channels []*Channel) error
24	26		}
25	27
26	28		type BaseBroker struct {

...	...		@@ -1,8 +1,78 @@
1	1		package pubsub
2	2
	3	+	import (
	4	+	"slices"
	5	+	"strings"
	6	+	"sync"
	7	+	)
	8	+
3	9		// MessageDispatcher defines how messages are dispatched to subscribers.
4	10		type MessageDispatcher interface {
5	11		// Dispatch sends a message to the appropriate subscriber(s).
6	12		// It receives the message, all subscribers, and the channel's sync primitives.
7	13		Dispatch(msg ChannelMessage, subscribers []*Client, channelDone chan struct{}) error
8	14		}
	15	+
	16	+	// MulticastDispatcher sends each message to all eligible subscribers.
	17	+	type MulticastDispatcher struct{}
	18	+
	19	+	func (d MulticastDispatcher) Dispatch(msg ChannelMessage, subscribers []Client, channelDone chan struct{}) error {
	20	+	var wg sync.WaitGroup
	21	+	for _, client := range subscribers {
	22	+	wg.Add(1)
	23	+	go func(cl *Client) {
	24	+	defer wg.Done()
	25	+	select {
	26	+	case cl.Data <- msg:
	27	+	case <-cl.Done:
	28	+	case <-channelDone:
	29	+	}
	30	+	}(client)
	31	+	}
	32	+	wg.Wait()
	33	+	return nil
	34	+	}
	35	+
	36	+	/*
	37	+	RoundRobin is a load-balancing broker that distributes published messages
	38	+	to subscribers using a round-robin algorithm.
	39	+
	40	+	Unlike Multicast which sends each message to all subscribers, RoundRobin
	41	+	sends each message to exactly one subscriber, rotating through the available
	42	+	subscribers for each published message. This provides load balancing for
	43	+	message processing.
	44	+
	45	+	It maintains independent round-robin state per channel/topic.
	46	+	*/
	47	+	type RoundRobinDispatcher struct {
	48	+	index uint32
	49	+	mu sync.Mutex
	50	+	}
	51	+
	52	+	func (d RoundRobinDispatcher) Dispatch(msg ChannelMessage, subscribers []Client, channelDone chan struct{}) error {
	53	+	// If no subscribers, nothing to dispatch
	54	+	// BlockWrite behavior at publish time ensures subscribers are present when needed
	55	+	if len(subscribers) == 0 {
	56	+	return nil
	57	+	}
	58	+
	59	+	slices.SortFunc(subscribers, func(a, b *Client) int {
	60	+	return strings.Compare(a.ID, b.ID)
	61	+	})
	62	+
	63	+	// Select the next subscriber in round-robin order
	64	+	d.mu.Lock()
	65	+	selectedIdx := int(d.index % uint32(len(subscribers)))
	66	+	d.index++
	67	+	d.mu.Unlock()
	68	+
	69	+	selectedClient := subscribers[selectedIdx]
	70	+
	71	+	select {
	72	+	case selectedClient.Data <- msg:
	73	+	case <-selectedClient.Done:
	74	+	case <-channelDone:
	75	+	}
	76	+
	77	+	return nil
	78	+	}

...	...		@@ -6,7 +6,6 @@ import (
6	6		"io"
7	7		"iter"
8	8		"log/slog"
9		-	"sync"
10	9
11	10		"github.com/antoniomika/syncmap"
12	11		)

...	...		@@ -62,9 +61,11 @@ func (p Multicast) GetSubs() iter.Seq2[string, Client] {
62	61		func (p Multicast) connect(ctx context.Context, ID string, rw io.ReadWriter, channels []Channel, direction ChannelDirection, blockWrite bool, replay, keepAlive bool, dispatcher MessageDispatcher) (error, error) {
63	62		client := NewClient(ID, rw, direction, blockWrite, replay, keepAlive)
64	63
65		-	// Set dispatcher on all channels
	64	+	// Set dispatcher on all channels (only if not already set)
66	65		for _, ch := range channels {
67		-	ch.Dispatcher = dispatcher
	66	+	if ch.Dispatcher == nil {
	67	+	ch.Dispatcher = dispatcher
	68	+	}
68	69		}
69	70
70	71		go func() {

...	...		@@ -87,24 +88,4 @@ func (p *Multicast) Sub(ctx context.Context, ID string, rw io.ReadWriter, channe
87	88		return errors.Join(p.connect(ctx, ID, rw, channels, ChannelDirectionOutput, false, false, keepAlive, &MulticastDispatcher{}))
88	89		}
89	90
90		-	// MulticastDispatcher sends each message to all eligible subscribers.
91		-	type MulticastDispatcher struct{}
92		-
93		-	func (d MulticastDispatcher) Dispatch(msg ChannelMessage, subscribers []Client, channelDone chan struct{}) error {
94		-	var wg sync.WaitGroup
95		-	for _, client := range subscribers {
96		-	wg.Add(1)
97		-	go func(cl *Client) {
98		-	defer wg.Done()
99		-	select {
100		-	case cl.Data <- msg:
101		-	case <-cl.Done:
102		-	case <-channelDone:
103		-	}
104		-	}(client)
105		-	}
106		-	wg.Wait()
107		-	return nil
108		-	}
109		-
110	91		var _ PubSub = (*Multicast)(nil)

...	...		@@ -1,51 +0,0 @@
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		-	}

pico

Patchset 103.3 on 2026-02-01T17:16:24Z · commit d0dfc85

Patchset `103.3` on 2026-02-01T17:16:24Z · commit d0dfc85