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.1` on 2026-01-05T14:52:02Z · commit 670350c

feat(pipe): add pipe_monitors table

Eric Bower 2025-12-26T21:02:39Z

chore(pipe): add new db methods

Eric Bower 2025-12-26T21:24:18Z

chore(pipe): add db impl

Eric Bower 2025-12-26T21:37:49Z

chore(pipe): add tests for monitoring

Eric Bower 2025-12-28T14:41:40Z

feat(pipe): add monitor cli

Eric Bower 2025-12-28T14:44:04Z

feat(pipe): status and rss commands

Eric Bower 2025-12-28T14:55:12Z

feat(pipe): monitor pub and pipe cmd

Eric Bower 2025-12-28T15:26:40Z

chore(pipe): monitor help text

Eric Bower 2025-12-28T15:40:57Z

→ refactor(pipe): monitor pipes on throttle interval

Eric Bower 2025-12-28T16:22:23Z

fix: window and ping fixes

Eric Bower 2026-01-03T02:26:40Z

chore(pipe): add tests

Eric Bower 2026-01-03T03:50:20Z

refactor(pipe): monitor pipes on throttle interval

Eric Bower 2025-12-28T16:22:23Z

Semantic diff summary

5 added, 3 modified, 0 signature changed, 0 removed across 1 analyzed file

pkg/apps/pipe/cli.go

chunk lines 9-15 modified
function_declaration pub modified
method_declaration Read added
method_declaration Write added
type_declaration throttledMonitorRW added
function_declaration newThrottledMonitorRW added
method_declaration throttledUpdate added
function_declaration pipe modified

+54 -2 pkg/apps/pipe/cli.go #

......@@ -9,6 +9,7 @@ import (
	"log/slog"
	"slices"
	"strings"
	"sync/atomic"
	"text/tabwriter"
	"time"

......@@ -673,10 +674,12 @@ func (handler *CliHandler) pub(cmd *CliCmd, topic string, clientID string) error
		_, _ = fmt.Fprintln(cmd.sesh, "sending msg ...")
	}

	throttledRW := newThrottledMonitorRW(rw, handler, cmd, name)

	err := handler.PubSub.Pub(
		cmd.pipeCtx,
		clientID,
		rw,
		throttledRW,
		[]*psub.Channel{
			psub.NewChannel(name),
		},

......@@ -720,6 +723,53 @@ func (handler *CliHandler) updateMonitor(cmd *CliCmd, topic string) {
	}
}

const monitorThrottleInterval = 15 * time.Second

type throttledMonitorRW struct {
	rw       io.ReadWriter
	handler  *CliHandler
	cmd      *CliCmd
	topic    string
	lastPing atomic.Int64 // Unix nanoseconds
}

func newThrottledMonitorRW(rw io.ReadWriter, handler *CliHandler, cmd *CliCmd, topic string) *throttledMonitorRW {
	return &throttledMonitorRW{
		rw:      rw,
		handler: handler,
		cmd:     cmd,
		topic:   topic,
	}
}

func (t *throttledMonitorRW) throttledUpdate() {
	now := time.Now().UnixNano()
	last := t.lastPing.Load()

	// First ping (last == 0) or interval elapsed
	if last == 0 || now-last >= int64(monitorThrottleInterval) {
		if t.lastPing.CompareAndSwap(last, now) {
			t.handler.updateMonitor(t.cmd, t.topic)
		}
	}
}

func (t *throttledMonitorRW) Read(p []byte) (int, error) {
	n, err := t.rw.Read(p)
	if n > 0 {
		t.throttledUpdate()
	}
	return n, err
}

func (t *throttledMonitorRW) Write(p []byte) (int, error) {
	n, err := t.rw.Write(p)
	if n > 0 {
		t.throttledUpdate()
	}
	return n, err
}

func (handler *CliHandler) sub(cmd *CliCmd, topic string, clientID string) error {
	subCmd := flagSet("sub", cmd.sesh)
	access := subCmd.String("a", "", "Comma separated list of pico usernames or ssh-key fingerprints to allow access to a topic")

......@@ -899,10 +949,12 @@ func (handler *CliHandler) pipe(cmd *CliCmd, topic string, clientID string) erro
		)
	}

	throttledRW := newThrottledMonitorRW(cmd.sesh, handler, cmd, name)

	readErr, writeErr := handler.PubSub.Pipe(
		cmd.pipeCtx,
		clientID,
		cmd.sesh,
		throttledRW,
		[]*psub.Channel{
			psub.NewChannel(name),
		},

...	...		@@ -673,10 +674,12 @@ func (handler CliHandler) pub(cmd CliCmd, topic string, clientID string) error
673	674		_, _ = fmt.Fprintln(cmd.sesh, "sending msg ...")
674	675		}
675	676
	677	+	throttledRW := newThrottledMonitorRW(rw, handler, cmd, name)
	678	+
676	679		err := handler.PubSub.Pub(
677	680		cmd.pipeCtx,
678	681		clientID,
679		-	rw,
	682	+	throttledRW,
680	683		[]*psub.Channel{
681	684		psub.NewChannel(name),
682	685		},

...	...		@@ -720,6 +723,53 @@ func (handler CliHandler) updateMonitor(cmd CliCmd, topic string) {
720	723		}
721	724		}
722	725
	726	+	const monitorThrottleInterval = 15 * time.Second
	727	+
	728	+	type throttledMonitorRW struct {
	729	+	rw io.ReadWriter
	730	+	handler *CliHandler
	731	+	cmd *CliCmd
	732	+	topic string
	733	+	lastPing atomic.Int64 // Unix nanoseconds
	734	+	}
	735	+
	736	+	func newThrottledMonitorRW(rw io.ReadWriter, handler CliHandler, cmd CliCmd, topic string) *throttledMonitorRW {
	737	+	return &throttledMonitorRW{
	738	+	rw: rw,
	739	+	handler: handler,
	740	+	cmd: cmd,
	741	+	topic: topic,
	742	+	}
	743	+	}
	744	+
	745	+	func (t *throttledMonitorRW) throttledUpdate() {
	746	+	now := time.Now().UnixNano()
	747	+	last := t.lastPing.Load()
	748	+
	749	+	// First ping (last == 0) or interval elapsed
	750	+	if last == 0 \|\| now-last >= int64(monitorThrottleInterval) {
	751	+	if t.lastPing.CompareAndSwap(last, now) {
	752	+	t.handler.updateMonitor(t.cmd, t.topic)
	753	+	}
	754	+	}
	755	+	}
	756	+
	757	+	func (t *throttledMonitorRW) Read(p []byte) (int, error) {
	758	+	n, err := t.rw.Read(p)
	759	+	if n > 0 {
	760	+	t.throttledUpdate()
	761	+	}
	762	+	return n, err
	763	+	}
	764	+
	765	+	func (t *throttledMonitorRW) Write(p []byte) (int, error) {
	766	+	n, err := t.rw.Write(p)
	767	+	if n > 0 {
	768	+	t.throttledUpdate()
	769	+	}
	770	+	return n, err
	771	+	}
	772	+
723	773		func (handler CliHandler) sub(cmd CliCmd, topic string, clientID string) error {
724	774		subCmd := flagSet("sub", cmd.sesh)
725	775		access := subCmd.String("a", "", "Comma separated list of pico usernames or ssh-key fingerprints to allow access to a topic")

...	...		@@ -899,10 +949,12 @@ func (handler CliHandler) pipe(cmd CliCmd, topic string, clientID string) erro
899	949		)
900	950		}
901	951
	952	+	throttledRW := newThrottledMonitorRW(cmd.sesh, handler, cmd, name)
	953	+
902	954		readErr, writeErr := handler.PubSub.Pipe(
903	955		cmd.pipeCtx,
904	956		clientID,
905		-	cmd.sesh,
	957	+	throttledRW,
906	958		[]*psub.Channel{
907	959		psub.NewChannel(name),
908	960		},

...	...		@@ -9,6 +9,7 @@ import (
9	9		"log/slog"
10	10		"slices"
11	11		"strings"
	12	+	"sync/atomic"
12	13		"text/tabwriter"
13	14		"time"
14	15

pico

Patchset 103.1 on 2026-01-05T14:52:02Z · commit 670350c

Patchset `103.1` on 2026-01-05T14:52:02Z · commit 670350c