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

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

chore(pipe): implement monitor history db interface

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

Semantic diff summary

6 added, 1 modified, 0 signature changed, 0 removed across 3 analyzed files (1 file skipped: unsupported file type)

pkg/db/db.go

type_declaration PipeMonitorHistory added
method_declaration Status added
chunk lines 526-533 modified

pkg/db/postgres/storage.go

method_declaration InsertPipeMonitorHistory added
method_declaration FindPipeMonitorHistory added

pkg/db/stub/stub.go

method_declaration InsertPipeMonitorHistory added
method_declaration FindPipeMonitorHistory added

+13 -0 pkg/db/db.go #

......@@ -390,6 +390,16 @@ type PipeMonitor struct {
	UpdatedAt *time.Time    `json:"updated_at" db:"updated_at"`
}

type PipeMonitorHistory struct {
	ID        string        `json:"id" db:"id"`
	MonitorID string        `json:"monitor_id" db:"monitor_id"`
	WindowDur time.Duration `json:"window_dur" db:"window_dur"`
	WindowEnd *time.Time    `json:"window_end" db:"window_end"`
	LastPing  *time.Time    `json:"last_ping" db:"last_ping"`
	CreatedAt *time.Time    `json:"created_at" db:"created_at"`
	UpdatedAt *time.Time    `json:"updated_at" db:"updated_at"`
}

func (m *PipeMonitor) Status() error {
	if m.LastPing == nil {
		return fmt.Errorf("no ping received yet")

......@@ -516,5 +526,8 @@ type DB interface {
	FindPipeMonitorByTopic(userID, topic string) (*PipeMonitor, error)
	FindPipeMonitorsByUser(userID string) ([]*PipeMonitor, error)

	InsertPipeMonitorHistory(monitorID string, windowDur time.Duration, windowEnd, lastPing *time.Time) error
	FindPipeMonitorHistory(monitorID string, from, to time.Time) ([]*PipeMonitorHistory, error)

	Close() error
}

+21 -0 pkg/db/postgres/storage.go #

......@@ -1565,3 +1565,24 @@ func (me *PsqlDB) FindPipeMonitorsByUser(userID string) ([]*db.PipeMonitor, erro
	}
	return monitors, nil
}

func (me *PsqlDB) InsertPipeMonitorHistory(monitorID string, windowDur time.Duration, windowEnd, lastPing *time.Time) error {
	_, err := me.Db.Exec(
		`INSERT INTO pipe_monitors_history (monitor_id, window_dur, window_end, last_ping) VALUES ($1, $2, $3, $4)`,
		monitorID, windowDur, windowEnd, lastPing,
	)
	return err
}

func (me *PsqlDB) FindPipeMonitorHistory(monitorID string, from, to time.Time) ([]*db.PipeMonitorHistory, error) {
	var history []*db.PipeMonitorHistory
	err := me.Db.Select(
		&history,
		`SELECT id, monitor_id, (EXTRACT(EPOCH FROM window_dur) * 1000000000)::bigint as window_dur, window_end, last_ping, created_at, updated_at FROM pipe_monitors_history WHERE monitor_id = $1 AND last_ping <= $2 AND window_end >= $3 ORDER BY last_ping ASC`,
		monitorID, to, from,
	)
	if err != nil {
		return nil, err
	}
	return history, nil
}

+8 -0 pkg/db/stub/stub.go #

......@@ -263,3 +263,11 @@ func (me *StubDB) FindPipeMonitorByTopic(userID, topic string) (*db.PipeMonitor,
func (me *StubDB) FindPipeMonitorsByUser(userID string) ([]*db.PipeMonitor, error) {
	return nil, errNotImpl
}

func (me *StubDB) InsertPipeMonitorHistory(monitorID string, windowDur time.Duration, windowEnd, lastPing *time.Time) error {
	return errNotImpl
}

func (me *StubDB) FindPipeMonitorHistory(monitorID string, from, to time.Time) ([]*db.PipeMonitorHistory, error) {
	return nil, errNotImpl
}

+3 -0 sql/migrations/20251226_add_pipe_monitoring.sql #

......@@ -31,3 +31,6 @@ CREATE TABLE IF NOT EXISTS pipe_monitors_history (
  ON DELETE CASCADE
  ON UPDATE CASCADE
);

CREATE INDEX IF NOT EXISTS idx_pipe_mon_hist_monitor_last_ping ON pipe_monitors_history (monitor_id, last_ping);
CREATE INDEX IF NOT EXISTS idx_pipe_mon_hist_monitor_window_end ON pipe_monitors_history (monitor_id, window_end);

...	...		@@ -390,6 +390,16 @@ type PipeMonitor struct {
390	390		UpdatedAt *time.Time `json:"updated_at" db:"updated_at"`
391	391		}
392	392
	393	+	type PipeMonitorHistory struct {
	394	+	ID string `json:"id" db:"id"`
	395	+	MonitorID string `json:"monitor_id" db:"monitor_id"`
	396	+	WindowDur time.Duration `json:"window_dur" db:"window_dur"`
	397	+	WindowEnd *time.Time `json:"window_end" db:"window_end"`
	398	+	LastPing *time.Time `json:"last_ping" db:"last_ping"`
	399	+	CreatedAt *time.Time `json:"created_at" db:"created_at"`
	400	+	UpdatedAt *time.Time `json:"updated_at" db:"updated_at"`
	401	+	}
	402	+
393	403		func (m *PipeMonitor) Status() error {
394	404		if m.LastPing == nil {
395	405		return fmt.Errorf("no ping received yet")

...	...		@@ -1565,3 +1565,24 @@ func (me PsqlDB) FindPipeMonitorsByUser(userID string) ([]db.PipeMonitor, erro
1565	1565		}
1566	1566		return monitors, nil
1567	1567		}
	1568	+
	1569	+	func (me PsqlDB) InsertPipeMonitorHistory(monitorID string, windowDur time.Duration, windowEnd, lastPing time.Time) error {
	1570	+	_, err := me.Db.Exec(
	1571	+	`INSERT INTO pipe_monitors_history (monitor_id, window_dur, window_end, last_ping) VALUES ($1, $2, $3, $4)`,
	1572	+	monitorID, windowDur, windowEnd, lastPing,
	1573	+	)
	1574	+	return err
	1575	+	}
	1576	+
	1577	+	func (me PsqlDB) FindPipeMonitorHistory(monitorID string, from, to time.Time) ([]db.PipeMonitorHistory, error) {
	1578	+	var history []*db.PipeMonitorHistory
	1579	+	err := me.Db.Select(
	1580	+	&history,
	1581	+	`SELECT id, monitor_id, (EXTRACT(EPOCH FROM window_dur) * 1000000000)::bigint as window_dur, window_end, last_ping, created_at, updated_at FROM pipe_monitors_history WHERE monitor_id = $1 AND last_ping <= $2 AND window_end >= $3 ORDER BY last_ping ASC`,
	1582	+	monitorID, to, from,
	1583	+	)
	1584	+	if err != nil {
	1585	+	return nil, err
	1586	+	}
	1587	+	return history, nil
	1588	+	}

...	...		@@ -263,3 +263,11 @@ func (me StubDB) FindPipeMonitorByTopic(userID, topic string) (db.PipeMonitor,
263	263		func (me StubDB) FindPipeMonitorsByUser(userID string) ([]db.PipeMonitor, error) {
264	264		return nil, errNotImpl
265	265		}
	266	+
	267	+	func (me StubDB) InsertPipeMonitorHistory(monitorID string, windowDur time.Duration, windowEnd, lastPing time.Time) error {
	268	+	return errNotImpl
	269	+	}
	270	+
	271	+	func (me StubDB) FindPipeMonitorHistory(monitorID string, from, to time.Time) ([]db.PipeMonitorHistory, error) {
	272	+	return nil, errNotImpl
	273	+	}

...	...		@@ -516,5 +526,8 @@ type DB interface {
516	526		FindPipeMonitorByTopic(userID, topic string) (*PipeMonitor, error)
517	527		FindPipeMonitorsByUser(userID string) ([]*PipeMonitor, error)
518	528
	529	+	InsertPipeMonitorHistory(monitorID string, windowDur time.Duration, windowEnd, lastPing *time.Time) error
	530	+	FindPipeMonitorHistory(monitorID string, from, to time.Time) ([]*PipeMonitorHistory, error)
	531	+
519	532		Close() error
520	533		}

...	...		@@ -31,3 +31,6 @@ CREATE TABLE IF NOT EXISTS pipe_monitors_history (
31	31		ON DELETE CASCADE
32	32		ON UPDATE CASCADE
33	33		);
	34	+
	35	+	CREATE INDEX IF NOT EXISTS idx_pipe_mon_hist_monitor_last_ping ON pipe_monitors_history (monitor_id, last_ping);
	36	+	CREATE INDEX IF NOT EXISTS idx_pipe_mon_hist_monitor_window_end ON pipe_monitors_history (monitor_id, window_end);

pico

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

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