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

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
+13 -15 pkg/apps/pipe/cli.go #
......@@ -367,7 +367,7 @@ func (handler *CliHandler) monitor(cmd *CliCmd, user *db.User) error {
367367 return fmt.Errorf("invalid duration %q: %w", durStr, err)
368368 }
369369
370- winEnd := time.Now().Add(dur)
370+ winEnd := time.Now().UTC().Add(dur)
371371 err = handler.DBPool.UpsertPipeMonitor(user.ID, resolvedTopic, dur, &winEnd)
372372 if err != nil {
373373 return fmt.Errorf("failed to create monitor: %w", err)
......@@ -393,7 +393,7 @@ func (handler *CliHandler) status(cmd *CliCmd, user *db.User) error {
393393 }
394394
395395 writer := tabwriter.NewWriter(cmd.sesh, 0, 0, 2, ' ', tabwriter.TabIndent)
396- _, _ = fmt.Fprintln(writer, "Topic\tStatus\tReason\tWindow\tLast Ping\tCreated")
396+ _, _ = fmt.Fprintln(writer, "Topic\tStatus\tWindow\tLast Ping\tWindow End\tReason")
397397
398398 for _, m := range monitors {
399399 status := "healthy"
......@@ -405,12 +405,12 @@ func (handler *CliHandler) status(cmd *CliCmd, user *db.User) error {
405405
406406 lastPing := "never"
407407 if m.LastPing != nil {
408- lastPing = m.LastPing.Format("2006-01-02 15:04:05")
408+ lastPing = m.LastPing.UTC().Format("2006-01-02 15:04:05Z")
409409 }
410410
411- createdAt := ""
412- if m.CreatedAt != nil {
413- createdAt = m.CreatedAt.Format("2006-01-02 15:04:05")
411+ windowEnd := ""
412+ if m.WindowEnd != nil {
413+ windowEnd = m.WindowEnd.UTC().Format("2006-01-02 15:04:05Z")
414414 }
415415
416416 _, _ = fmt.Fprintf(
......@@ -418,10 +418,10 @@ func (handler *CliHandler) status(cmd *CliCmd, user *db.User) error {
418418 "%s\t%s\t%s\t%s\t%s\t%s\r\n",
419419 m.Topic,
420420 status,
421- reason,
422421 m.WindowDur.String(),
423422 lastPing,
424- createdAt,
423+ windowEnd,
424+ reason,
425425 )
426426 }
427427 _ = writer.Flush()
......@@ -709,17 +709,15 @@ func (handler *CliHandler) updateMonitor(cmd *CliCmd, topic string) {
709709 return
710710 }
711711
712- now := time.Now()
712+ now := time.Now().UTC()
713713 if err := handler.DBPool.UpdatePipeMonitorLastPing(cmd.user.ID, topic, &now); err != nil {
714714 handler.Logger.Error("failed to update monitor last_ping", "err", err, "topic", topic)
715715 }
716716
717- // Advance window if current time is past window_end
718- if monitor.WindowEnd != nil && now.After(*monitor.WindowEnd) {
719- nextWindow := monitor.GetNextWindow()
720- if err := handler.DBPool.UpsertPipeMonitor(cmd.user.ID, topic, monitor.WindowDur, nextWindow); err != nil {
721- handler.Logger.Error("failed to advance monitor window", "err", err, "topic", topic)
722- }
717+ // Always reset the window to now + duration on ping
718+ newWindowEnd := now.Add(monitor.WindowDur)
719+ if err := handler.DBPool.UpsertPipeMonitor(cmd.user.ID, topic, monitor.WindowDur, &newWindowEnd); err != nil {
720+ handler.Logger.Error("failed to reset monitor window", "err", err, "topic", topic)
723721 }
724722 }
725723
+12 -7 pkg/db/db.go #
......@@ -397,15 +397,20 @@ func (m *PipeMonitor) Status() error {
397397 if m.WindowEnd == nil {
398398 return fmt.Errorf("window end not set")
399399 }
400+ now := time.Now().UTC()
401+ if now.After(*m.WindowEnd) {
402+ return fmt.Errorf(
403+ "window expired at %s",
404+ m.WindowEnd.UTC().Format("2006-01-02 15:04:05Z"),
405+ )
406+ }
400407 windowStart := m.WindowEnd.Add(-m.WindowDur)
401- lastPingAfterStart := m.LastPing.After(windowStart)
408+ lastPingAfterStart := !m.LastPing.Before(windowStart)
402409 if !lastPingAfterStart {
403- return fmt.Errorf("last ping before window start: last_ping=%s window_start=%s", m.LastPing, windowStart)
404- }
405- lastPingBeforeEnd := m.LastPing.Before(*m.WindowEnd)
406- if !lastPingBeforeEnd {
407- // should not happen but just for data validity we add it
408- return fmt.Errorf("last ping after window end: last_ping=%s window_end=%s", m.LastPing, m.WindowEnd)
410+ return fmt.Errorf(
411+ "last ping before window start: %s",
412+ windowStart.UTC().Format("2006-01-02 15:04:05Z"),
413+ )
409414 }
410415 return nil
411416 }
Back to top