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.2` on 2026-01-08T01:19:34Z · commit e0c3a88

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

chore(pipe): add logging stmts

Eric Bower 2026-01-05T14:46:42Z

refactor(pipe): create pipe monitors history table

Eric Bower 2026-01-07T20:00:47Z

→ fix(pipe): only update monitor if within window

Eric Bower 2026-01-07T21:52:08Z

chore(pipe): implement monitor history db interface

Eric Bower 2026-01-07T23:19:05Z

feat(pipe): record historical monitors

Eric Bower 2026-01-07T23:27:20Z

feat(pipe): monitor calculate uptime

Eric Bower 2026-01-07T23:27:20Z

feat(pipe): uptime cli command

Eric Bower 2026-01-08T00:48:50Z

chore(pipe): monitor history stubs

Eric Bower 2026-01-08T00:52:48Z

fix(pipe): uptime cmd

Eric Bower 2026-01-08T01:18:38Z

fix(pipe): only update monitor if within window

Eric Bower 2026-01-07T21:52:08Z

Semantic diff summary

1 added, 3 modified, 0 signature changed, 0 removed across 2 analyzed files

pkg/apps/pipe/cli.go

function_declaration updateMonitor modified

pkg/apps/pipe/ssh_test.go

type_declaration TestSSHServer modified
function_declaration NewTestSSHServer modified
function_declaration TestMonitor_FixedWindowNonSliding added

+30 -8 pkg/apps/pipe/cli.go #

......@@ -726,17 +726,39 @@ func (handler *CliHandler) updateMonitor(cmd *CliCmd, topic string) {
	}

	now := time.Now().UTC()
	if err := handler.DBPool.UpdatePipeMonitorLastPing(cmd.user.ID, topic, &now); err != nil {
		handler.Logger.Error("failed to update monitor last_ping", "err", err, "topic", topic)

	// Fixed window semantics: windows are discrete, non-overlapping time slots.
	// - last_ping: always updated to show most recent activity (user visibility)
	// - window_end: only advances when current time exceeds it (health scheduling)

	// If we're past the current window, advance to the window containing `now`
	newWindowEnd := *monitor.WindowEnd
	if !now.Before(*monitor.WindowEnd) {
		// Calculate which window period `now` falls into
		elapsed := now.Sub(*monitor.WindowEnd)
		periods := int(elapsed/monitor.WindowDur) + 1
		newWindowEnd = monitor.WindowEnd.Add(time.Duration(periods) * monitor.WindowDur)

		if err := handler.DBPool.UpsertPipeMonitor(cmd.user.ID, topic, monitor.WindowDur, &newWindowEnd); err != nil {
			handler.Logger.Error("failed to advance monitor window", "err", err, "topic", topic)
		}
		handler.Logger.Info("advanced monitor window",
			"topic", topic,
			"oldWindowEnd", monitor.WindowEnd.Format(time.RFC3339),
			"newWindowEnd", newWindowEnd.Format(time.RFC3339),
			"periodsMissed", periods-1,
		)
	}
	handler.Logger.Info("update monitor ping", "topic", topic)

	// Always reset the window to now + duration on ping
	newWindowEnd := now.Add(monitor.WindowDur)
	if err := handler.DBPool.UpsertPipeMonitor(cmd.user.ID, topic, monitor.WindowDur, &newWindowEnd); err != nil {
		handler.Logger.Error("failed to reset monitor window", "err", err, "topic", topic)
	// Always record the latest ping for user visibility
	if err := handler.DBPool.UpdatePipeMonitorLastPing(cmd.user.ID, topic, &now); err != nil {
		handler.Logger.Error("failed to update monitor last_ping", "err", err, "topic", topic)
	}
	handler.Logger.Info("updated monitor window", "topic", topic, "window", newWindowEnd.UTC().Format(time.RFC3339))
	handler.Logger.Info("recorded monitor ping",
		"topic", topic,
		"pingTime", now.Format(time.RFC3339),
		"windowEnd", newWindowEnd.Format(time.RFC3339),
	)
}

const monitorThrottleInterval = 15 * time.Second

+106 -6 pkg/apps/pipe/ssh_test.go #

......@@ -162,9 +162,10 @@ func (t *TestDB) FindPipeMonitorsByUser(userID string) ([]*db.PipeMonitor, error
}

type TestSSHServer struct {
	Cfg    *shared.ConfigSite
	DBPool *TestDB
	Cancel context.CancelFunc
	Cfg         *shared.ConfigSite
	DBPool      *TestDB
	PipeHandler *CliHandler
	Cancel      context.CancelFunc
}

func NewTestSSHServer(t *testing.T) *TestSSHServer {

......@@ -243,9 +244,10 @@ func NewTestSSHServer(t *testing.T) *TestSSHServer {
	time.Sleep(100 * time.Millisecond)

	return &TestSSHServer{
		Cfg:    cfg,
		DBPool: dbpool,
		Cancel: cancel,
		Cfg:         cfg,
		DBPool:      dbpool,
		PipeHandler: handler,
		Cancel:      cancel,
	}
}

......@@ -2093,3 +2095,101 @@ func TestStatus_HealthyImmediatelyAfterPing(t *testing.T) {
		t.Errorf("status should show healthy, got: %s", output)
	}
}

// TestMonitor_FixedWindowNonSliding verifies that pings within the same window
// do not slide the window forward. This is a regression test for a bug where
// each ping reset window_end to now+dur, creating a sliding window that never fails.
//
// Expected behavior:
//   - last_ping: always updated to show most recent activity (user visibility).
//   - window_end: only advances when current time exceeds it (health scheduling).
func TestMonitor_FixedWindowNonSliding(t *testing.T) {
	server := NewTestSSHServer(t)
	defer server.Shutdown()

	user := GenerateUser("alice")
	RegisterUserWithServer(server, user)

	client, err := user.NewClient()
	if err != nil {
		t.Fatalf("failed to connect: %v", err)
	}
	defer func() { _ = client.Close() }()

	// Create a monitor with 1 hour window
	_, err = user.RunCommand(client, "monitor fixed-window-test 1h")
	if err != nil {
		t.Logf("create command completed: %v", err)
	}

	// Get the initial window_end
	monitor, err := server.DBPool.FindPipeMonitorByTopic("alice-id", "alice/fixed-window-test")
	if err != nil {
		t.Fatalf("monitor should exist: %v", err)
	}
	initialWindowEnd := *monitor.WindowEnd

	// Simulate a ping by calling updateMonitor directly
	handler := server.PipeHandler

	// Create a mock CliCmd
	mockUser := &db.User{ID: "alice-id", Name: "alice"}
	cmd := &CliCmd{
		userName: "alice",
		user:     mockUser,
	}

	// First ping - should record last_ping but NOT change window_end
	handler.updateMonitor(cmd, "alice/fixed-window-test")

	monitor, err = server.DBPool.FindPipeMonitorByTopic("alice-id", "alice/fixed-window-test")
	if err != nil {
		t.Fatalf("monitor should exist after first ping: %v", err)
	}

	if monitor.LastPing == nil {
		t.Fatalf("last_ping should be set after first ping")
	}
	firstPingTime := *monitor.LastPing
	windowEndAfterFirstPing := *monitor.WindowEnd

	// BUG CHECK: With the bug, window_end would have slid forward to now+1h
	// With the fix, window_end should remain at the original scheduled time
	if !windowEndAfterFirstPing.Equal(initialWindowEnd) {
		t.Errorf("BUG DETECTED: window_end should NOT change after first ping within window\n"+
			"initial window_end: %v\n"+
			"window_end after ping: %v\n"+
			"Window slid forward by: %v",
			initialWindowEnd.Format(time.RFC3339),
			windowEndAfterFirstPing.Format(time.RFC3339),
			windowEndAfterFirstPing.Sub(initialWindowEnd))
	}

	// Second ping - last_ping SHOULD be updated (for user visibility)
	// but window_end should NOT change
	time.Sleep(10 * time.Millisecond) // Small delay to get different timestamp
	handler.updateMonitor(cmd, "alice/fixed-window-test")

	monitor, err = server.DBPool.FindPipeMonitorByTopic("alice-id", "alice/fixed-window-test")
	if err != nil {
		t.Fatalf("monitor should exist after second ping: %v", err)
	}

	// last_ping SHOULD be updated to show most recent activity
	if monitor.LastPing.Equal(firstPingTime) {
		t.Errorf("last_ping SHOULD be updated for user visibility\n"+
			"first ping time: %v\n"+
			"last_ping after second call: %v",
			firstPingTime.Format(time.RFC3339Nano),
			monitor.LastPing.Format(time.RFC3339Nano))
	}

	// But window_end should still be the original value (not sliding)
	if !monitor.WindowEnd.Equal(initialWindowEnd) {
		t.Errorf("BUG DETECTED: window_end should remain at original value\n"+
			"initial: %v\n"+
			"current: %v",
			initialWindowEnd.Format(time.RFC3339),
			monitor.WindowEnd.Format(time.RFC3339))
	}
}

...	...		@@ -726,17 +726,39 @@ func (handler CliHandler) updateMonitor(cmd CliCmd, topic string) {
726	726		}
727	727
728	728		now := time.Now().UTC()
729		-	if err := handler.DBPool.UpdatePipeMonitorLastPing(cmd.user.ID, topic, &now); err != nil {
730		-	handler.Logger.Error("failed to update monitor last_ping", "err", err, "topic", topic)
	729	+
	730	+	// Fixed window semantics: windows are discrete, non-overlapping time slots.
	731	+	// - last_ping: always updated to show most recent activity (user visibility)
	732	+	// - window_end: only advances when current time exceeds it (health scheduling)
	733	+
	734	+	// If we're past the current window, advance to the window containing `now`
	735	+	newWindowEnd := *monitor.WindowEnd
	736	+	if !now.Before(*monitor.WindowEnd) {
	737	+	// Calculate which window period `now` falls into
	738	+	elapsed := now.Sub(*monitor.WindowEnd)
	739	+	periods := int(elapsed/monitor.WindowDur) + 1
	740	+	newWindowEnd = monitor.WindowEnd.Add(time.Duration(periods) * monitor.WindowDur)
	741	+
	742	+	if err := handler.DBPool.UpsertPipeMonitor(cmd.user.ID, topic, monitor.WindowDur, &newWindowEnd); err != nil {
	743	+	handler.Logger.Error("failed to advance monitor window", "err", err, "topic", topic)
	744	+	}
	745	+	handler.Logger.Info("advanced monitor window",
	746	+	"topic", topic,
	747	+	"oldWindowEnd", monitor.WindowEnd.Format(time.RFC3339),
	748	+	"newWindowEnd", newWindowEnd.Format(time.RFC3339),
	749	+	"periodsMissed", periods-1,
	750	+	)
731	751		}
732		-	handler.Logger.Info("update monitor ping", "topic", topic)
733	752
734		-	// Always reset the window to now + duration on ping
735		-	newWindowEnd := now.Add(monitor.WindowDur)
736		-	if err := handler.DBPool.UpsertPipeMonitor(cmd.user.ID, topic, monitor.WindowDur, &newWindowEnd); err != nil {
737		-	handler.Logger.Error("failed to reset monitor window", "err", err, "topic", topic)
	753	+	// Always record the latest ping for user visibility
	754	+	if err := handler.DBPool.UpdatePipeMonitorLastPing(cmd.user.ID, topic, &now); err != nil {
	755	+	handler.Logger.Error("failed to update monitor last_ping", "err", err, "topic", topic)
738	756		}
739		-	handler.Logger.Info("updated monitor window", "topic", topic, "window", newWindowEnd.UTC().Format(time.RFC3339))
	757	+	handler.Logger.Info("recorded monitor ping",
	758	+	"topic", topic,
	759	+	"pingTime", now.Format(time.RFC3339),
	760	+	"windowEnd", newWindowEnd.Format(time.RFC3339),
	761	+	)
740	762		}
741	763
742	764		const monitorThrottleInterval = 15 * time.Second

...	...		@@ -243,9 +244,10 @@ func NewTestSSHServer(t testing.T) TestSSHServer {
243	244		time.Sleep(100 * time.Millisecond)
244	245
245	246		return &TestSSHServer{
246		-	Cfg: cfg,
247		-	DBPool: dbpool,
248		-	Cancel: cancel,
	247	+	Cfg: cfg,
	248	+	DBPool: dbpool,
	249	+	PipeHandler: handler,
	250	+	Cancel: cancel,
249	251		}
250	252		}
251	253

...	...		@@ -2093,3 +2095,101 @@ func TestStatus_HealthyImmediatelyAfterPing(t *testing.T) {
2093	2095		t.Errorf("status should show healthy, got: %s", output)
2094	2096		}
2095	2097		}
	2098	+
	2099	+	// TestMonitor_FixedWindowNonSliding verifies that pings within the same window
	2100	+	// do not slide the window forward. This is a regression test for a bug where
	2101	+	// each ping reset window_end to now+dur, creating a sliding window that never fails.
	2102	+	//
	2103	+	// Expected behavior:
	2104	+	// - last_ping: always updated to show most recent activity (user visibility).
	2105	+	// - window_end: only advances when current time exceeds it (health scheduling).
	2106	+	func TestMonitor_FixedWindowNonSliding(t *testing.T) {
	2107	+	server := NewTestSSHServer(t)
	2108	+	defer server.Shutdown()
	2109	+
	2110	+	user := GenerateUser("alice")
	2111	+	RegisterUserWithServer(server, user)
	2112	+
	2113	+	client, err := user.NewClient()
	2114	+	if err != nil {
	2115	+	t.Fatalf("failed to connect: %v", err)
	2116	+	}
	2117	+	defer func() { _ = client.Close() }()
	2118	+
	2119	+	// Create a monitor with 1 hour window
	2120	+	_, err = user.RunCommand(client, "monitor fixed-window-test 1h")
	2121	+	if err != nil {
	2122	+	t.Logf("create command completed: %v", err)
	2123	+	}
	2124	+
	2125	+	// Get the initial window_end
	2126	+	monitor, err := server.DBPool.FindPipeMonitorByTopic("alice-id", "alice/fixed-window-test")
	2127	+	if err != nil {
	2128	+	t.Fatalf("monitor should exist: %v", err)
	2129	+	}
	2130	+	initialWindowEnd := *monitor.WindowEnd
	2131	+
	2132	+	// Simulate a ping by calling updateMonitor directly
	2133	+	handler := server.PipeHandler
	2134	+
	2135	+	// Create a mock CliCmd
	2136	+	mockUser := &db.User{ID: "alice-id", Name: "alice"}
	2137	+	cmd := &CliCmd{
	2138	+	userName: "alice",
	2139	+	user: mockUser,
	2140	+	}
	2141	+
	2142	+	// First ping - should record last_ping but NOT change window_end
	2143	+	handler.updateMonitor(cmd, "alice/fixed-window-test")
	2144	+
	2145	+	monitor, err = server.DBPool.FindPipeMonitorByTopic("alice-id", "alice/fixed-window-test")
	2146	+	if err != nil {
	2147	+	t.Fatalf("monitor should exist after first ping: %v", err)
	2148	+	}
	2149	+
	2150	+	if monitor.LastPing == nil {
	2151	+	t.Fatalf("last_ping should be set after first ping")
	2152	+	}
	2153	+	firstPingTime := *monitor.LastPing
	2154	+	windowEndAfterFirstPing := *monitor.WindowEnd
	2155	+
	2156	+	// BUG CHECK: With the bug, window_end would have slid forward to now+1h
	2157	+	// With the fix, window_end should remain at the original scheduled time
	2158	+	if !windowEndAfterFirstPing.Equal(initialWindowEnd) {
	2159	+	t.Errorf("BUG DETECTED: window_end should NOT change after first ping within window\n"+
	2160	+	"initial window_end: %v\n"+
	2161	+	"window_end after ping: %v\n"+
	2162	+	"Window slid forward by: %v",
	2163	+	initialWindowEnd.Format(time.RFC3339),
	2164	+	windowEndAfterFirstPing.Format(time.RFC3339),
	2165	+	windowEndAfterFirstPing.Sub(initialWindowEnd))
	2166	+	}
	2167	+
	2168	+	// Second ping - last_ping SHOULD be updated (for user visibility)
	2169	+	// but window_end should NOT change
	2170	+	time.Sleep(10 * time.Millisecond) // Small delay to get different timestamp
	2171	+	handler.updateMonitor(cmd, "alice/fixed-window-test")
	2172	+
	2173	+	monitor, err = server.DBPool.FindPipeMonitorByTopic("alice-id", "alice/fixed-window-test")
	2174	+	if err != nil {
	2175	+	t.Fatalf("monitor should exist after second ping: %v", err)
	2176	+	}
	2177	+
	2178	+	// last_ping SHOULD be updated to show most recent activity
	2179	+	if monitor.LastPing.Equal(firstPingTime) {
	2180	+	t.Errorf("last_ping SHOULD be updated for user visibility\n"+
	2181	+	"first ping time: %v\n"+
	2182	+	"last_ping after second call: %v",
	2183	+	firstPingTime.Format(time.RFC3339Nano),
	2184	+	monitor.LastPing.Format(time.RFC3339Nano))
	2185	+	}
	2186	+
	2187	+	// But window_end should still be the original value (not sliding)
	2188	+	if !monitor.WindowEnd.Equal(initialWindowEnd) {
	2189	+	t.Errorf("BUG DETECTED: window_end should remain at original value\n"+
	2190	+	"initial: %v\n"+
	2191	+	"current: %v",
	2192	+	initialWindowEnd.Format(time.RFC3339),
	2193	+	monitor.WindowEnd.Format(time.RFC3339))
	2194	+	}
	2195	+	}

...	...		@@ -162,9 +162,10 @@ func (t TestDB) FindPipeMonitorsByUser(userID string) ([]db.PipeMonitor, error
162	162		}
163	163
164	164		type TestSSHServer struct {
165		-	Cfg *shared.ConfigSite
166		-	DBPool *TestDB
167		-	Cancel context.CancelFunc
	165	+	Cfg *shared.ConfigSite
	166	+	DBPool *TestDB
	167	+	PipeHandler *CliHandler
	168	+	Cancel context.CancelFunc
168	169		}
169	170
170	171		func NewTestSSHServer(t testing.T) TestSSHServer {

pico

Patchset 103.2 on 2026-01-08T01:19:34Z · commit e0c3a88

Patchset `103.2` on 2026-01-08T01:19:34Z · commit e0c3a88