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 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

fix: window and ping fixes

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

Semantic diff summary

0 added, 4 modified, 0 signature changed, 0 removed across 2 analyzed files

pkg/apps/pipe/cli.go

function_declaration monitor modified
function_declaration status modified
function_declaration updateMonitor modified

pkg/db/db.go

function_declaration Status modified

+13 -15 pkg/apps/pipe/cli.go #

......@@ -367,7 +367,7 @@ func (handler *CliHandler) monitor(cmd *CliCmd, user *db.User) error {
		return fmt.Errorf("invalid duration %q: %w", durStr, err)
	}

	winEnd := time.Now().Add(dur)
	winEnd := time.Now().UTC().Add(dur)
	err = handler.DBPool.UpsertPipeMonitor(user.ID, resolvedTopic, dur, &winEnd)
	if err != nil {
		return fmt.Errorf("failed to create monitor: %w", err)

......@@ -393,7 +393,7 @@ func (handler *CliHandler) status(cmd *CliCmd, user *db.User) error {
	}

	writer := tabwriter.NewWriter(cmd.sesh, 0, 0, 2, ' ', tabwriter.TabIndent)
	_, _ = fmt.Fprintln(writer, "Topic\tStatus\tReason\tWindow\tLast Ping\tCreated")
	_, _ = fmt.Fprintln(writer, "Topic\tStatus\tWindow\tLast Ping\tWindow End\tReason")

	for _, m := range monitors {
		status := "healthy"

......@@ -405,12 +405,12 @@ func (handler *CliHandler) status(cmd *CliCmd, user *db.User) error {

		lastPing := "never"
		if m.LastPing != nil {
			lastPing = m.LastPing.Format("2006-01-02 15:04:05")
			lastPing = m.LastPing.UTC().Format("2006-01-02 15:04:05Z")
		}

		createdAt := ""
		if m.CreatedAt != nil {
			createdAt = m.CreatedAt.Format("2006-01-02 15:04:05")
		windowEnd := ""
		if m.WindowEnd != nil {
			windowEnd = m.WindowEnd.UTC().Format("2006-01-02 15:04:05Z")
		}

		_, _ = fmt.Fprintf(

......@@ -418,10 +418,10 @@ func (handler *CliHandler) status(cmd *CliCmd, user *db.User) error {
			"%s\t%s\t%s\t%s\t%s\t%s\r\n",
			m.Topic,
			status,
			reason,
			m.WindowDur.String(),
			lastPing,
			createdAt,
			windowEnd,
			reason,
		)
	}
	_ = writer.Flush()

......@@ -709,17 +709,15 @@ func (handler *CliHandler) updateMonitor(cmd *CliCmd, topic string) {
		return
	}

	now := time.Now()
	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)
	}

	// Advance window if current time is past window_end
	if monitor.WindowEnd != nil && now.After(*monitor.WindowEnd) {
		nextWindow := monitor.GetNextWindow()
		if err := handler.DBPool.UpsertPipeMonitor(cmd.user.ID, topic, monitor.WindowDur, nextWindow); err != nil {
			handler.Logger.Error("failed to advance monitor window", "err", err, "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)
	}
}

+12 -7 pkg/db/db.go #

......@@ -397,15 +397,20 @@ func (m *PipeMonitor) Status() error {
	if m.WindowEnd == nil {
		return fmt.Errorf("window end not set")
	}
	now := time.Now().UTC()
	if now.After(*m.WindowEnd) {
		return fmt.Errorf(
			"window expired at %s",
			m.WindowEnd.UTC().Format("2006-01-02 15:04:05Z"),
		)
	}
	windowStart := m.WindowEnd.Add(-m.WindowDur)
	lastPingAfterStart := m.LastPing.After(windowStart)
	lastPingAfterStart := !m.LastPing.Before(windowStart)
	if !lastPingAfterStart {
		return fmt.Errorf("last ping before window start: last_ping=%s window_start=%s", m.LastPing, windowStart)
	}
	lastPingBeforeEnd := m.LastPing.Before(*m.WindowEnd)
	if !lastPingBeforeEnd {
		// should not happen but just for data validity we add it
		return fmt.Errorf("last ping after window end: last_ping=%s window_end=%s", m.LastPing, m.WindowEnd)
		return fmt.Errorf(
			"last ping before window start: %s",
			windowStart.UTC().Format("2006-01-02 15:04:05Z"),
		)
	}
	return nil
}

...	...		@@ -367,7 +367,7 @@ func (handler CliHandler) monitor(cmd CliCmd, user *db.User) error {
367	367		return fmt.Errorf("invalid duration %q: %w", durStr, err)
368	368		}
369	369
370		-	winEnd := time.Now().Add(dur)
	370	+	winEnd := time.Now().UTC().Add(dur)
371	371		err = handler.DBPool.UpsertPipeMonitor(user.ID, resolvedTopic, dur, &winEnd)
372	372		if err != nil {
373	373		return fmt.Errorf("failed to create monitor: %w", err)

...	...		@@ -405,12 +405,12 @@ func (handler CliHandler) status(cmd CliCmd, user *db.User) error {
405	405
406	406		lastPing := "never"
407	407		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")
409	409		}
410	410
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")
414	414		}
415	415
416	416		_, _ = fmt.Fprintf(

...	...		@@ -709,17 +709,15 @@ func (handler CliHandler) updateMonitor(cmd CliCmd, topic string) {
709	709		return
710	710		}
711	711
712		-	now := time.Now()
	712	+	now := time.Now().UTC()
713	713		if err := handler.DBPool.UpdatePipeMonitorLastPing(cmd.user.ID, topic, &now); err != nil {
714	714		handler.Logger.Error("failed to update monitor last_ping", "err", err, "topic", topic)
715	715		}
716	716
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)
723	721		}
724	722		}
725	723

...	...		@@ -397,15 +397,20 @@ func (m *PipeMonitor) Status() error {
397	397		if m.WindowEnd == nil {
398	398		return fmt.Errorf("window end not set")
399	399		}
	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	+	}
400	407		windowStart := m.WindowEnd.Add(-m.WindowDur)
401		-	lastPingAfterStart := m.LastPing.After(windowStart)
	408	+	lastPingAfterStart := !m.LastPing.Before(windowStart)
402	409		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	+	)
409	414		}
410	415		return nil
411	416		}

...	...		@@ -393,7 +393,7 @@ func (handler CliHandler) status(cmd CliCmd, user *db.User) error {
393	393		}
394	394
395	395		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")
397	397
398	398		for _, m := range monitors {
399	399		status := "healthy"

...	...		@@ -418,10 +418,10 @@ func (handler CliHandler) status(cmd CliCmd, user *db.User) error {
418	418		"%s\t%s\t%s\t%s\t%s\t%s\r\n",
419	419		m.Topic,
420	420		status,
421		-	reason,
422	421		m.WindowDur.String(),
423	422		lastPing,
424		-	createdAt,
	423	+	windowEnd,
	424	+	reason,
425	425		)
426	426		}
427	427		_ = writer.Flush()

pico

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

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