amadan

carlos / platform Public

host: recover the hour a restart across a boundary never rolled up

Rollups were derived only from what a process itself buffered, so an
agent that stopped at 09:58 and started again at 10:07 never held an
hour-09 sample and never rolled that hour up. Its raw chunk — durably
written before the stop — then sat with no day entry until pruneStaleRaw
deleted it 48h later, losing the hour from the trend history
permanently, silently, and only for hours containing a restart.

On its first flush a process now looks back one hour: if that hour has a
stored raw chunk and the day object has no entry for it, roll the stored
samples up. Deliberately narrow — one hour back (a longer outage wrote
no chunk to recover), an existing entry is left untouched, and a
missing/empty/unreadable chunk is a no-op. The repair never fails the
flush; a failed attempt just retries on the next one.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell b9b21a3b47e21a1d02307862cd6b39538da03f8e parent 1f72d1f
2 files changed, +214 −0
  • internal/host/metrics_loop.go +88 −0
  • internal/host/metrics_loop_test.go +126 −0
diff --git a/internal/host/metrics_loop.go b/internal/host/metrics_loop.go
index dab133a13..33ba9714e 100644
--- a/internal/host/metrics_loop.go
+++ b/internal/host/metrics_loop.go
@@ -63,6 +63,7 @@ type sampler struct {
prevAt time.Time
buf []metrics.Sample
writtenHours map[string]bool // hour -> this process has written/merged it at least once
+ backfilled bool // the one-shot previous-hour rollup repair has run (see backfillPrevHourRollup)
}
// tick reads p once and appends one metrics.Sample to sm.buf. The very
@@ -159,6 +160,19 @@ func metricsRawPrefix(box string) string {
// would silently clobber that hour's pre-crash samples and permanently
// corrupt its day rollup the moment RollupHour runs over the truncated set.
func metricsFlush(ctx context.Context, s release.Store, box string, sm *sampler, now time.Time) error {
+ // Once per process, before anything else: repair the hour a restart may
+ // have left un-rolled-up (see backfillPrevHourRollup). It is deliberately
+ // not allowed to fail the flush — it is repair work for an hour that is
+ // already over, and this flush's own samples matter more. A failure just
+ // leaves the flag unset so the next flush tries again.
+ if !sm.backfilled {
+ if err := backfillPrevHourRollup(ctx, s, box, now); err != nil {
+ log.Printf("host: previous-hour rollup backfill for %s failed, retrying next flush: %v", box, err)
+ } else {
+ sm.backfilled = true
+ }
+ }
+
if len(sm.buf) == 0 {
return nil
}
@@ -228,6 +242,80 @@ func metricsFlush(ctx context.Context, s release.Store, box string, sm *sampler,
return nil
}
+// backfillPrevHourRollup rolls up the PREVIOUS hour's stored raw chunk if the
+// day object has no entry for that hour yet. It runs once per process, on the
+// first flush.
+//
+// The gap it closes: rollups are otherwise derived only from what this
+// process itself buffered. An agent that stops at 09:58 and starts again at
+// 10:07 never has an hour-09 sample in buf, so nothing ever calls RollupHour
+// for hour 09 — the raw chunk written before the stop sits there with no day
+// entry, and 48h later pruneStaleRaw deletes it. The hour is then gone from
+// the trend history permanently, silently, and only for the hours that
+// contain a restart (i.e. exactly the interesting ones).
+//
+// Deliberately narrow, in three ways. Only the immediately previous hour is
+// considered: a longer outage has no raw chunk to recover from anyway, since
+// the agent that would have written it was down. An hour whose rollup already
+// exists is left completely alone — this repairs a gap, it never rewrites a
+// closed hour. And a missing chunk, an empty one, or an unreadable one is a
+// no-op rather than an error: there is simply nothing to recover.
+func backfillPrevHourRollup(ctx context.Context, s release.Store, box string, now time.Time) error {
+ prev := now.Add(-time.Hour)
+ hour := metrics.HourOf(prev)
+ key := metrics.RawKey(box, hour)
+
+ body, _, err := s.Get(ctx, key)
+ if err != nil {
+ if errors.Is(err, os.ErrNotExist) {
+ return nil // no chunk for that hour: nothing to roll up
+ }
+ return err
+ }
+ var chunk metrics.RawChunk
+ if err := json.Unmarshal(body, &chunk); err != nil {
+ log.Printf("host: raw chunk %s unreadable, skipping rollup backfill: %v", key, err)
+ return nil
+ }
+ if len(chunk.Samples) == 0 {
+ return nil
+ }
+
+ has, err := rollupHasHour(ctx, s, box, prev)
+ if err != nil {
+ return err
+ }
+ if has {
+ return nil
+ }
+
+ log.Printf("host: hour %s has a raw chunk but no day rollup (restart across the boundary) — rolling it up now", hour)
+ return metrics.UpsertRollup(ctx, s, box, prev, metrics.RollupHour(chunk.Samples))
+}
+
+// rollupHasHour reports whether box's day object already carries hour's UTC
+// hour bucket. A missing day object is a plain "no". An unreadable one
+// reports "yes": UpsertRollup would fail on the same unmarshal anyway, and a
+// corrupt day object is a separate problem the normal rollover path will
+// surface — the backfill must not spin on it every flush.
+func rollupHasHour(ctx context.Context, s release.Store, box string, hour time.Time) (bool, error) {
+ key := metrics.DayKey(box, hour.UTC().Format("2006-01-02"))
+ body, _, err := s.Get(ctx, key)
+ if err != nil {
+ if errors.Is(err, os.ErrNotExist) {
+ return false, nil
+ }
+ return false, err
+ }
+ var day metrics.DayRollup
+ if err := json.Unmarshal(body, &day); err != nil {
+ log.Printf("host: day rollup %s unreadable, skipping rollup backfill: %v", key, err)
+ return true, nil
+ }
+ _, ok := day.Hours[hour.UTC().Format("15")]
+ return ok, nil
+}
+
// writeHourChunk writes samples as hour's raw chunk, merging in whatever
// chunk already exists at that key the first time THIS sampler writes that
// particular hour (tracked in sm.writtenHours) — see metricsFlush's doc
diff --git a/internal/host/metrics_loop_test.go b/internal/host/metrics_loop_test.go
index 6d2a422ef..a84456516 100644
--- a/internal/host/metrics_loop_test.go
+++ b/internal/host/metrics_loop_test.go
@@ -644,3 +644,129 @@ func TestMetricsFlushPartialFailureDoesNotRegressEarlierHour(t *testing.T) {
t.Errorf("day rollup hour 09 cpu N=%d, want 2 (merged count)", n)
}
}
+
+// TestMetricsFlushBackfillsPrevHourRollupAfterRestart pins the round-3 review
+// fix: rollups were derived only from in-memory buf, so an agent that stopped
+// at 09:58 and came back at 10:07 left hour 09's raw chunk — durably written
+// before the stop — with no day-rollup entry, ever. Nothing else rolls that
+// hour up afterwards, and the 48h prune deletes the chunk two days later, so
+// the hour was permanently lost from the trend history. The first flush of a
+// process now looks back one hour and rolls up the stored chunk if the day
+// object lacks it.
+func TestMetricsFlushBackfillsPrevHourRollupAfterRestart(t *testing.T) {
+ ctx := context.Background()
+ store := release.NewDirStore(t.TempDir())
+ box := "box-a"
+
+ hour09Start := time.Date(2026, 8, 2, 9, 0, 0, 0, time.UTC)
+ cpu := 0.4
+ preStop := make([]metrics.Sample, 0, 10)
+ for i := 0; i < 10; i++ {
+ v := cpu
+ preStop = append(preStop, metrics.Sample{
+ At: hour09Start.Add(time.Duration(i) * time.Minute),
+ Host: metrics.HostSample{CPUFrac: &v},
+ })
+ }
+ if err := metrics.WriteRaw(ctx, store, metrics.RawChunk{
+ V: 1, Box: box, Hour: metrics.HourOf(hour09Start), Samples: preStop,
+ }); err != nil {
+ t.Fatal(err)
+ }
+ // No day rollup at all: the agent stopped before hour 09 ever closed.
+ if _, _, err := store.Get(ctx, metrics.DayKey(box, "2026-08-02")); err == nil {
+ t.Fatal("fixture invalid: a day rollup already exists")
+ }
+
+ // Restart at 10:07: a fresh sampler with one hour-10 tick, so the flush
+ // itself never touches hour 09.
+ postCPU := 0.9
+ sm := &sampler{buf: []metrics.Sample{
+ {At: time.Date(2026, 8, 2, 10, 7, 0, 0, time.UTC), Host: metrics.HostSample{CPUFrac: &postCPU}},
+ }}
+ now := time.Date(2026, 8, 2, 10, 7, 0, 0, time.UTC)
+ if err := metricsFlush(ctx, store, box, sm, now); err != nil {
+ t.Fatal(err)
+ }
+
+ body, _, err := store.Get(ctx, metrics.DayKey(box, "2026-08-02"))
+ if err != nil {
+ t.Fatalf("day rollup missing — hour 09 was never rolled up: %v", err)
+ }
+ var day metrics.DayRollup
+ if err := json.Unmarshal(body, &day); err != nil {
+ t.Fatal(err)
+ }
+ agg, ok := day.Hours["09"]
+ if !ok {
+ t.Fatalf("day rollup has no hour 09: %+v", day.Hours)
+ }
+ if n := agg.Host["cpu"].N; n != 10 {
+ t.Fatalf("hour 09 cpu N=%d, want 10 (rolled up from the stored raw chunk)", n)
+ }
+ if avg := agg.Host["cpu"].Avg; avg < 0.399 || avg > 0.401 {
+ t.Fatalf("hour 09 cpu avg=%v, want ~0.4 (the stored chunk's readings)", avg)
+ }
+}
+
+// TestMetricsFlushBackfillLeavesExistingRollupAlone: the backfill is repair
+// work for a gap, not a rewrite. An hour whose rollup is already present —
+// the normal case, written when the hour closed under the previous process —
+// must be left exactly as it is, and the backfill must not run again on
+// later flushes of the same process.
+func TestMetricsFlushBackfillLeavesExistingRollupAlone(t *testing.T) {
+ ctx := context.Background()
+ store := release.NewDirStore(t.TempDir())
+ box := "box-a"
+
+ hour09Start := time.Date(2026, 8, 2, 9, 0, 0, 0, time.UTC)
+ cpu := 0.4
+ if err := metrics.WriteRaw(ctx, store, metrics.RawChunk{
+ V: 1, Box: box, Hour: metrics.HourOf(hour09Start),
+ Samples: []metrics.Sample{
+ {At: hour09Start, Host: metrics.HostSample{CPUFrac: &cpu}},
+ {At: hour09Start.Add(time.Minute), Host: metrics.HostSample{CPUFrac: &cpu}},
+ },
+ }); err != nil {
+ t.Fatal(err)
+ }
+ // A rollup that plainly did NOT come from the two samples above, so a
+ // clobber is unmistakable.
+ sentinel := metrics.HourAgg{Host: map[string]metrics.Agg{"cpu": {Min: 0.1, Avg: 0.2, Max: 0.3, N: 57}}}
+ if err := metrics.UpsertRollup(ctx, store, box, hour09Start, sentinel); err != nil {
+ t.Fatal(err)
+ }
+
+ postCPU := 0.9
+ sm := &sampler{buf: []metrics.Sample{
+ {At: time.Date(2026, 8, 2, 10, 7, 0, 0, time.UTC), Host: metrics.HostSample{CPUFrac: &postCPU}},
+ }}
+ if err := metricsFlush(ctx, store, box, sm, time.Date(2026, 8, 2, 10, 7, 0, 0, time.UTC)); err != nil {
+ t.Fatal(err)
+ }
+
+ readHour09 := func() metrics.HourAgg {
+ t.Helper()
+ body, _, err := store.Get(ctx, metrics.DayKey(box, "2026-08-02"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ var day metrics.DayRollup
+ if err := json.Unmarshal(body, &day); err != nil {
+ t.Fatal(err)
+ }
+ return day.Hours["09"]
+ }
+ if got := readHour09().Host["cpu"]; got != sentinel.Host["cpu"] {
+ t.Fatalf("hour 09 cpu = %+v, want the pre-existing %+v untouched", got, sentinel.Host["cpu"])
+ }
+
+ // A second flush must not reconsider it either.
+ sm.buf = []metrics.Sample{{At: time.Date(2026, 8, 2, 10, 12, 0, 0, time.UTC), Host: metrics.HostSample{CPUFrac: &postCPU}}}
+ if err := metricsFlush(ctx, store, box, sm, time.Date(2026, 8, 2, 10, 12, 0, 0, time.UTC)); err != nil {
+ t.Fatal(err)
+ }
+ if got := readHour09().Host["cpu"]; got != sentinel.Host["cpu"] {
+ t.Fatalf("second flush changed hour 09 to %+v", got)
+ }
+}