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

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

feat(pipe): monitor calculate uptime

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

Semantic diff summary

4 added, 0 modified, 0 signature changed, 0 removed across 1 analyzed file

pkg/db/db.go

type_declaration UptimeResult added
function_declaration ComputeUptime added
type_declaration interval added
method_declaration Status added

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

......@@ -400,6 +400,86 @@ type PipeMonitorHistory struct {
	UpdatedAt *time.Time    `json:"updated_at" db:"updated_at"`
}

type UptimeResult struct {
	TotalDuration  time.Duration
	UptimeDuration time.Duration
	UptimePercent  float64
}

func ComputeUptime(history []*PipeMonitorHistory, from, to time.Time) UptimeResult {
	totalDuration := to.Sub(from)
	if totalDuration <= 0 {
		return UptimeResult{}
	}

	if len(history) == 0 {
		return UptimeResult{TotalDuration: totalDuration}
	}

	type interval struct {
		start, end time.Time
	}

	var intervals []interval
	for _, h := range history {
		if h.WindowEnd == nil {
			continue
		}
		windowStart := h.WindowEnd.Add(-h.WindowDur)
		windowEnd := *h.WindowEnd

		if windowStart.Before(from) {
			windowStart = from
		}
		if windowEnd.After(to) {
			windowEnd = to
		}

		if windowStart.Before(windowEnd) {
			intervals = append(intervals, interval{start: windowStart, end: windowEnd})
		}
	}

	if len(intervals) == 0 {
		return UptimeResult{TotalDuration: totalDuration}
	}

	// Sort by start time
	for i := range intervals {
		for j := i + 1; j < len(intervals); j++ {
			if intervals[j].start.Before(intervals[i].start) {
				intervals[i], intervals[j] = intervals[j], intervals[i]
			}
		}
	}

	// Merge overlapping intervals
	merged := []interval{intervals[0]}
	for _, curr := range intervals[1:] {
		last := &merged[len(merged)-1]
		if !curr.start.After(last.end) {
			if curr.end.After(last.end) {
				last.end = curr.end
			}
		} else {
			merged = append(merged, curr)
		}
	}

	var uptimeDuration time.Duration
	for _, iv := range merged {
		uptimeDuration += iv.end.Sub(iv.start)
	}

	uptimePercent := float64(uptimeDuration) / float64(totalDuration) * 100

	return UptimeResult{
		TotalDuration:  totalDuration,
		UptimeDuration: uptimeDuration,
		UptimePercent:  uptimePercent,
	}
}

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

...	...		@@ -400,6 +400,86 @@ type PipeMonitorHistory struct {
400	400		UpdatedAt *time.Time `json:"updated_at" db:"updated_at"`
401	401		}
402	402
	403	+	type UptimeResult struct {
	404	+	TotalDuration time.Duration
	405	+	UptimeDuration time.Duration
	406	+	UptimePercent float64
	407	+	}
	408	+
	409	+	func ComputeUptime(history []*PipeMonitorHistory, from, to time.Time) UptimeResult {
	410	+	totalDuration := to.Sub(from)
	411	+	if totalDuration <= 0 {
	412	+	return UptimeResult{}
	413	+	}
	414	+
	415	+	if len(history) == 0 {
	416	+	return UptimeResult{TotalDuration: totalDuration}
	417	+	}
	418	+
	419	+	type interval struct {
	420	+	start, end time.Time
	421	+	}
	422	+
	423	+	var intervals []interval
	424	+	for _, h := range history {
	425	+	if h.WindowEnd == nil {
	426	+	continue
	427	+	}
	428	+	windowStart := h.WindowEnd.Add(-h.WindowDur)
	429	+	windowEnd := *h.WindowEnd
	430	+
	431	+	if windowStart.Before(from) {
	432	+	windowStart = from
	433	+	}
	434	+	if windowEnd.After(to) {
	435	+	windowEnd = to
	436	+	}
	437	+
	438	+	if windowStart.Before(windowEnd) {
	439	+	intervals = append(intervals, interval{start: windowStart, end: windowEnd})
	440	+	}
	441	+	}
	442	+
	443	+	if len(intervals) == 0 {
	444	+	return UptimeResult{TotalDuration: totalDuration}
	445	+	}
	446	+
	447	+	// Sort by start time
	448	+	for i := range intervals {
	449	+	for j := i + 1; j < len(intervals); j++ {
	450	+	if intervals[j].start.Before(intervals[i].start) {
	451	+	intervals[i], intervals[j] = intervals[j], intervals[i]
	452	+	}
	453	+	}
	454	+	}
	455	+
	456	+	// Merge overlapping intervals
	457	+	merged := []interval{intervals[0]}
	458	+	for _, curr := range intervals[1:] {
	459	+	last := &merged[len(merged)-1]
	460	+	if !curr.start.After(last.end) {
	461	+	if curr.end.After(last.end) {
	462	+	last.end = curr.end
	463	+	}
	464	+	} else {
	465	+	merged = append(merged, curr)
	466	+	}
	467	+	}
	468	+
	469	+	var uptimeDuration time.Duration
	470	+	for _, iv := range merged {
	471	+	uptimeDuration += iv.end.Sub(iv.start)
	472	+	}
	473	+
	474	+	uptimePercent := float64(uptimeDuration) / float64(totalDuration) * 100
	475	+
	476	+	return UptimeResult{
	477	+	TotalDuration: totalDuration,
	478	+	UptimeDuration: uptimeDuration,
	479	+	UptimePercent: uptimePercent,
	480	+	}
	481	+	}
	482	+
403	483		func (m *PipeMonitor) Status() error {
404	484		if m.LastPing == nil {
405	485		return fmt.Errorf("no ping received yet")

pico

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

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