pico

feat(pipe): add pipe_monitors table rss
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 68f5cd1

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
+172 -0 pkg/apps/pipe/ssh_test.go #
......@@ -1921,3 +1921,175 @@ func TestPub_UpdatesMonitorLastPing(t *testing.T) {
19211921 t.Errorf("last_ping should be recent, got: %v", monitor.LastPing)
19221922 }
19231923 }
1924+
1925+// Tests for monitor status edge cases
1926+
1927+func TestStatus_PingAtExactWindowStart(t *testing.T) {
1928+ // Bug fix: Status() should use >= for windowStart comparison
1929+ // A ping exactly at windowStart should be healthy
1930+ now := time.Now().UTC()
1931+ windowEnd := now.Add(1 * time.Hour)
1932+ windowStart := windowEnd.Add(-1 * time.Hour) // equals now
1933+
1934+ monitor := &db.PipeMonitor{
1935+ LastPing: &windowStart, // ping exactly at window start
1936+ WindowEnd: &windowEnd,
1937+ WindowDur: 1 * time.Hour,
1938+ }
1939+
1940+ err := monitor.Status()
1941+ if err != nil {
1942+ t.Errorf("ping at exact window start should be healthy, got: %v", err)
1943+ }
1944+}
1945+
1946+func TestStatus_WindowExpired(t *testing.T) {
1947+ // Bug fix: Status() should check if current time is past windowEnd
1948+ now := time.Now().UTC()
1949+ windowEnd := now.Add(-1 * time.Minute) // window ended 1 minute ago
1950+ lastPing := now.Add(-30 * time.Second) // ping was 30 seconds ago
1951+
1952+ monitor := &db.PipeMonitor{
1953+ LastPing: &lastPing,
1954+ WindowEnd: &windowEnd,
1955+ WindowDur: 1 * time.Hour,
1956+ }
1957+
1958+ err := monitor.Status()
1959+ if err == nil {
1960+ t.Error("expired window should be unhealthy")
1961+ }
1962+ if !strings.Contains(err.Error(), "window expired") {
1963+ t.Errorf("error should mention window expired, got: %v", err)
1964+ }
1965+}
1966+
1967+func TestStatus_PingResetsWindow(t *testing.T) {
1968+ // Bug fix: Every ping should reset window to now + duration
1969+ server := NewTestSSHServer(t)
1970+ defer server.Shutdown()
1971+
1972+ user := GenerateUser("alice")
1973+ RegisterUserWithServer(server, user)
1974+
1975+ // Create a monitor with an expired window
1976+ expiredWindowEnd := time.Now().UTC().Add(-10 * time.Minute)
1977+ _ = server.DBPool.UpsertPipeMonitor("alice-id", "alice/reset-test", 5*time.Minute, &expiredWindowEnd)
1978+
1979+ client, err := user.NewClient()
1980+ if err != nil {
1981+ t.Fatalf("failed to connect: %v", err)
1982+ }
1983+ defer func() { _ = client.Close() }()
1984+
1985+ // Start a subscriber first so pub doesn't block
1986+ subClient, err := user.NewClient()
1987+ if err != nil {
1988+ t.Fatalf("failed to connect subscriber: %v", err)
1989+ }
1990+ defer func() { _ = subClient.Close() }()
1991+
1992+ subSession, err := subClient.NewSession()
1993+ if err != nil {
1994+ t.Fatalf("failed to create sub session: %v", err)
1995+ }
1996+ defer func() { _ = subSession.Close() }()
1997+
1998+ if err := subSession.Start("sub reset-test -c"); err != nil {
1999+ t.Fatalf("failed to start sub: %v", err)
2000+ }
2001+
2002+ time.Sleep(100 * time.Millisecond)
2003+
2004+ // Pub to trigger monitor update
2005+ _, err = user.RunCommandWithStdin(client, "pub reset-test -c", "ping")
2006+ if err != nil {
2007+ t.Logf("pub command completed: %v", err)
2008+ }
2009+
2010+ // Check that window was reset
2011+ monitor, err := server.DBPool.FindPipeMonitorByTopic("alice-id", "alice/reset-test")
2012+ if err != nil {
2013+ t.Fatalf("monitor should exist: %v", err)
2014+ }
2015+
2016+ if monitor.WindowEnd == nil {
2017+ t.Fatal("window_end should be set")
2018+ }
2019+
2020+ // Window end should now be in the future
2021+ if !monitor.WindowEnd.After(time.Now().UTC()) {
2022+ t.Errorf("window_end should be in the future after ping, got: %v", monitor.WindowEnd)
2023+ }
2024+}
2025+
2026+func TestStatus_HealthyImmediatelyAfterPing(t *testing.T) {
2027+ // Bug fix: After a ping, status should immediately show healthy
2028+ server := NewTestSSHServer(t)
2029+ defer server.Shutdown()
2030+
2031+ user := GenerateUser("alice")
2032+ RegisterUserWithServer(server, user)
2033+
2034+ client, err := user.NewClient()
2035+ if err != nil {
2036+ t.Fatalf("failed to connect: %v", err)
2037+ }
2038+ defer func() { _ = client.Close() }()
2039+
2040+ // Create monitor
2041+ _, err = user.RunCommand(client, "monitor health-test 5m")
2042+ if err != nil {
2043+ t.Fatalf("failed to create monitor: %v", err)
2044+ }
2045+
2046+ // Start subscriber
2047+ subClient, err := user.NewClient()
2048+ if err != nil {
2049+ t.Fatalf("failed to connect subscriber: %v", err)
2050+ }
2051+ defer func() { _ = subClient.Close() }()
2052+
2053+ subSession, err := subClient.NewSession()
2054+ if err != nil {
2055+ t.Fatalf("failed to create sub session: %v", err)
2056+ }
2057+ defer func() { _ = subSession.Close() }()
2058+
2059+ if err := subSession.Start("sub health-test -c"); err != nil {
2060+ t.Fatalf("failed to start sub: %v", err)
2061+ }
2062+
2063+ time.Sleep(100 * time.Millisecond)
2064+
2065+ // Pub to trigger ping
2066+ pubClient, err := user.NewClient()
2067+ if err != nil {
2068+ t.Fatalf("failed to connect publisher: %v", err)
2069+ }
2070+ defer func() { _ = pubClient.Close() }()
2071+
2072+ _, err = user.RunCommandWithStdin(pubClient, "pub health-test -c", "ping")
2073+ if err != nil {
2074+ t.Logf("pub completed: %v", err)
2075+ }
2076+
2077+ // Immediately check status
2078+ statusClient, err := user.NewClient()
2079+ if err != nil {
2080+ t.Fatalf("failed to connect for status: %v", err)
2081+ }
2082+ defer func() { _ = statusClient.Close() }()
2083+
2084+ output, err := user.RunCommand(statusClient, "status")
2085+ if err != nil {
2086+ t.Logf("status completed: %v", err)
2087+ }
2088+
2089+ if strings.Contains(output, "unhealthy") {
2090+ t.Errorf("status should be healthy immediately after ping, got: %s", output)
2091+ }
2092+ if !strings.Contains(output, "healthy") {
2093+ t.Errorf("status should show healthy, got: %s", output)
2094+ }
2095+}
Back to top