amadan

carlos / platform Public

host: mark hours written only after their durable writes succeed

Round-3's generalization (writeHourChunk merging every hour, not just
the current one) marked sm.writtenHours[hour] right after the merge,
before WriteRaw even ran. If WriteRaw failed — or, for an old hour, a
LATER hour's write or that hour's UpsertRollup failed so metricsFlush
returned before trimming buf — the hour would be wrongly marked as
already written. A retry would then skip the merge for that hour, and
plain-overwrite its already-correct merged chunk with just the bare
in-memory samples still sitting in buf: a silent regression of data
that had already been durably (and correctly) written.

writeHourChunk no longer marks anything itself; it just reports whether
this call merged. metricsFlush's oldHours loop now marks a hour (via the
new markHourWritten) and immediately drops its samples from buf (via the
new dropHourFromBuf) only once BOTH that hour's WriteRaw and its
UpsertRollup have succeeded — so a later hour's failure within the same
flush call can no longer cause an earlier, already-durable hour to be
reprocessed and clobbered on retry. The current-hour path marks written
only after its own WriteRaw succeeds, the same discipline.

Also corrects alertTick's "Honest limit" doc comment: a publish outage
during a sustain-window rule's build-up does not "keep resetting the
clock" — Since commits normally on every event-free tick regardless of
publish's health, so an outage only defers the firing tick itself,
bounded by the outage's own duration, not compounding.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell 33df074f1d8dd58fb4e1ccd69fc98158b16e7644 parent 8b6c3c1
2 files changed, +210 −20
  • internal/host/metrics_loop.go +77 −20
  • internal/host/metrics_loop_test.go +133 −0
diff --git a/internal/host/metrics_loop.go b/internal/host/metrics_loop.go
index 0cbf605d7..86e6deb9f 100644
--- a/internal/host/metrics_loop.go
+++ b/internal/host/metrics_loop.go
@@ -180,7 +180,7 @@ func metricsFlush(ctx context.Context, s release.Store, box string, sm *sampler,
}
for _, hour := range oldHours {
- samples, err := sm.writeHourChunk(ctx, s, box, hour, oldByHour[hour])
+ samples, merged, err := sm.writeHourChunk(ctx, s, box, hour, oldByHour[hour])
if err != nil {
return err
}
@@ -192,6 +192,19 @@ func metricsFlush(ctx context.Context, s release.Store, box string, sm *sampler,
if err := metrics.UpsertRollup(ctx, s, box, samples[0].At, agg); err != nil {
return err
}
+ // Both durable writes for this hour succeeded — NOW it's safe to
+ // mark it written (only meaningful when this call actually merged;
+ // see writeHourChunk/markHourWritten) and retire its samples from
+ // buf immediately, rather than waiting for the whole flush call to
+ // finish. If a LATER hour's write fails below, buf must not still
+ // be carrying THIS hour's already-durably-written samples — a
+ // retry that reprocessed them would see writtenHours already true,
+ // skip the merge, and overwrite this hour's correctly-merged chunk
+ // with just the bare in-memory set, silently regressing it.
+ if merged {
+ sm.markHourWritten(hour)
+ }
+ sm.dropHourFromBuf(hour)
}
if len(oldHours) > 0 {
@@ -201,11 +214,14 @@ func metricsFlush(ctx context.Context, s release.Store, box string, sm *sampler,
}
if len(curSamples) > 0 {
- merged, err := sm.writeHourChunk(ctx, s, box, curHour, curSamples)
+ merged, wasMerged, err := sm.writeHourChunk(ctx, s, box, curHour, curSamples)
if err != nil {
return err
}
curSamples = merged
+ if wasMerged {
+ sm.markHourWritten(curHour)
+ }
}
sm.buf = curSamples
@@ -220,24 +236,59 @@ func metricsFlush(ctx context.Context, s release.Store, box string, sm *sampler,
// merge: by then sm.buf already carries forward everything this process
// itself wrote there, so re-merging would be redundant (and for an old
// hour, over by definition — it's dropped from buf right after this call).
-// Returns the actual sample set written (merged or not) so the caller uses
-// the right data for RollupHour / for keeping in buf.
-func (sm *sampler) writeHourChunk(ctx context.Context, s release.Store, box, hour string, samples []metrics.Sample) ([]metrics.Sample, error) {
- if sm.writtenHours == nil {
- sm.writtenHours = map[string]bool{}
- }
+//
+// Deliberately does NOT mark sm.writtenHours itself — only that the merge
+// happened this call (the returned merged bool). Marking is the caller's
+// job, done only once every durable write this hour's data depends on has
+// actually succeeded (for an old hour that's this WriteRaw AND its
+// UpsertRollup; for the current hour, just this WriteRaw). Marking here,
+// before the caller even knows whether those later steps succeed, would
+// let a downstream failure retry with the merge silently skipped — exactly
+// the bug this shape avoids. Returns the actual sample set written (merged
+// or not) so the caller uses the right data for RollupHour / for keeping
+// in buf.
+func (sm *sampler) writeHourChunk(ctx context.Context, s release.Store, box, hour string, samples []metrics.Sample) (out []metrics.Sample, merged bool, err error) {
if !sm.writtenHours[hour] {
- merged, err := mergeExistingChunk(ctx, s, box, hour, samples)
+ m, err := mergeExistingChunk(ctx, s, box, hour, samples)
if err != nil {
- return nil, err
+ return nil, false, err
}
- samples = merged
- sm.writtenHours[hour] = true
+ samples = m
+ merged = true
}
if err := metrics.WriteRaw(ctx, s, metrics.RawChunk{V: 1, Box: box, Hour: hour, Samples: samples}); err != nil {
- return nil, err
+ return nil, false, err
+ }
+ return samples, merged, nil
+}
+
+// markHourWritten records that hour's chunk has been durably written (and,
+// where applicable, rolled up) at least once by this sampler — see
+// writeHourChunk's doc comment for why the caller, not writeHourChunk
+// itself, decides when that's true.
+func (sm *sampler) markHourWritten(hour string) {
+ if sm.writtenHours == nil {
+ sm.writtenHours = map[string]bool{}
+ }
+ sm.writtenHours[hour] = true
+}
+
+// dropHourFromBuf removes every sample belonging to hour from sm.buf. Old
+// hours are retired from buf the moment their write is fully durable
+// (metricsFlush calls this right after, per hour, inside its oldHours
+// loop) rather than in one pass at the very end — so that if a LATER old
+// hour's write fails within the same flush call, THIS hour's samples are
+// already gone from buf and cannot be reprocessed (and, worse, rewritten
+// unmerged — see metricsFlush's doc comment at the call site) on the next
+// flush attempt.
+func (sm *sampler) dropHourFromBuf(hour string) {
+ kept := make([]metrics.Sample, 0, len(sm.buf))
+ for _, samp := range sm.buf {
+ if metrics.HourOf(samp.At) != hour {
+ kept = append(kept, samp)
+ }
}
- return samples, nil
+ sm.buf = kept
}
// mergeExistingChunk reads whatever raw chunk already exists for hour (most
@@ -331,12 +382,18 @@ func pruneStaleRaw(ctx context.Context, s release.Store, box string, now time.Ti
// exactly what this process believes right now — never a state the store
// was never told about.
//
-// Honest limit: for a sustain-window rule (mem, cpu — sustain > 0), not
-// committing *st on a publish failure also means that rule's Since
-// bookkeeping doesn't advance either, so a publish outage lasting the whole
-// sustain window would keep resetting the clock and the rule would never
-// actually fire until the store recovers. Immediate rules (disk, lag;
-// sustain 0) are unaffected — they fire the same tick regardless.
+// Honest limit: for a sustain-window rule (mem, cpu — sustain > 0), the
+// Since bookkeeping itself is NOT at risk from a publish outage — every
+// tick before the rule is actually ready to fire carries zero events, so
+// publish is never even attempted on those ticks and *st (Since included)
+// commits normally regardless of the store's health. The outage only bites
+// on the tick(s) where the rule is ready to fire: if publish fails exactly
+// then, *st doesn't advance, so the next tick re-evaluates the same
+// already-overdue transition and retries. The alert ends up firing (and
+// published) as soon as the store recovers, delayed by at most the
+// outage's own duration — it does not compound beyond that. Immediate
+// rules (disk, lag; sustain 0) retry the same way, just starting from
+// their very first over-threshold tick instead of a sustain deadline.
func alertTick(ctx context.Context, s release.Store, box string, rules metrics.Rules,
st *metrics.AlertState, latest metrics.Sample, statePath, webhookURL string,
post func(url string, body []byte) error) error {
diff --git a/internal/host/metrics_loop_test.go b/internal/host/metrics_loop_test.go
index 35a259ac0..6d2a422ef 100644
--- a/internal/host/metrics_loop_test.go
+++ b/internal/host/metrics_loop_test.go
@@ -511,3 +511,136 @@ func TestAlertTickRetriesTransitionAfterPublishFailure(t *testing.T) {
t.Errorf("published state disk rule not firing: %+v", published.Rules["disk"])
}
}
+
+// putFailsOnceForKey wraps a release.Store and makes Put fail exactly once
+// for one specific key — the fixture for
+// TestMetricsFlushPartialFailureDoesNotRegressEarlierHour's "a later hour's
+// write fails, an earlier hour's succeeded" scenario. Every other key, and
+// every call to failKey after the first, passes through unchanged.
+type putFailsOnceForKey struct {
+ release.Store
+ failKey string
+ failed bool
+}
+
+func (f *putFailsOnceForKey) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
+ if key == f.failKey && !f.failed {
+ f.failed = true
+ return "", errors.New("put fails once for this key (test)")
+ }
+ return f.Store.Put(ctx, key, body, ifMatch)
+}
+
+// TestMetricsFlushPartialFailureDoesNotRegressEarlierHour pins the
+// round-3 review's Critical fix: writtenHours must only be marked (and a
+// hour retired from buf) once ALL of that hour's durable writes actually
+// succeeded. Two old hours (A older, B less old) each already hold a
+// pre-crash sample in the store; a freshly "restarted" sampler holds one
+// post-restart sample for each in buf. The first flush call's Put fails
+// only for hour B's raw key (hour A, processed first, succeeds in full).
+// Before the round-3 fix, hour A would have been marked written and
+// dropped from buf only via the final `sm.buf = curSamples` at the very
+// end of metricsFlush, which this failure never reaches — so the retry
+// would reprocess hour A too, see it (wrongly) already marked, skip the
+// merge, and overwrite its correct 2-sample chunk with just the bare
+// 1-sample in-memory set. This test asserts hour A's chunk survives the
+// retry intact, and hour B ends up correctly merged too.
+func TestMetricsFlushPartialFailureDoesNotRegressEarlierHour(t *testing.T) {
+ ctx := context.Background()
+ realStore := release.NewDirStore(t.TempDir())
+ box := "box-a"
+
+ hourAStart := time.Date(2026, 8, 2, 8, 0, 0, 0, time.UTC)
+ hourBStart := time.Date(2026, 8, 2, 9, 0, 0, 0, time.UTC)
+
+ cpuA := 0.1
+ preA := []metrics.Sample{{At: hourAStart.Add(10 * time.Minute), Host: metrics.HostSample{CPUFrac: &cpuA}}}
+ if err := metrics.WriteRaw(ctx, realStore, metrics.RawChunk{
+ V: 1, Box: box, Hour: metrics.HourOf(hourAStart), Samples: preA,
+ }); err != nil {
+ t.Fatal(err)
+ }
+ cpuB := 0.2
+ preB := []metrics.Sample{{At: hourBStart.Add(10 * time.Minute), Host: metrics.HostSample{CPUFrac: &cpuB}}}
+ if err := metrics.WriteRaw(ctx, realStore, metrics.RawChunk{
+ V: 1, Box: box, Hour: metrics.HourOf(hourBStart), Samples: preB,
+ }); err != nil {
+ t.Fatal(err)
+ }
+
+ // A fresh (post-restart) sampler with one post-restart tick in each old
+ // hour — no current-hour sample, so the whole flush is "old hours".
+ postA, postB := 0.3, 0.4
+ sm := &sampler{buf: []metrics.Sample{
+ {At: hourAStart.Add(20 * time.Minute), Host: metrics.HostSample{CPUFrac: &postA}},
+ {At: hourBStart.Add(20 * time.Minute), Host: metrics.HostSample{CPUFrac: &postB}},
+ }}
+ now := time.Date(2026, 8, 2, 10, 5, 0, 0, time.UTC) // current hour 10 — both A and B are old
+
+ hourBKey := metrics.RawKey(box, metrics.HourOf(hourBStart))
+ failingStore := &putFailsOnceForKey{Store: realStore, failKey: hourBKey}
+
+ if err := metricsFlush(ctx, failingStore, box, sm, now); err == nil {
+ t.Fatal("expected the first flush to fail (hour B's Put fails once)")
+ }
+
+ // Hour A must already be fully durable: merged (2-sample) chunk.
+ bodyA, _, err := realStore.Get(ctx, metrics.RawKey(box, metrics.HourOf(hourAStart)))
+ if err != nil {
+ t.Fatalf("hour A chunk missing after the partial failure: %v", err)
+ }
+ var chunkA metrics.RawChunk
+ if err := json.Unmarshal(bodyA, &chunkA); err != nil {
+ t.Fatal(err)
+ }
+ if len(chunkA.Samples) != 2 {
+ t.Fatalf("hour A chunk has %d samples after the partial failure, want 2 (merged)", len(chunkA.Samples))
+ }
+
+ // Retry: hour B's Put now succeeds.
+ if err := metricsFlush(ctx, failingStore, box, sm, now); err != nil {
+ t.Fatalf("retry flush: %v", err)
+ }
+
+ // Hour A must STILL be the merged 2-sample chunk — the retry must not
+ // have reprocessed and regressed it.
+ bodyA2, _, err := realStore.Get(ctx, metrics.RawKey(box, metrics.HourOf(hourAStart)))
+ if err != nil {
+ t.Fatal(err)
+ }
+ var chunkA2 metrics.RawChunk
+ if err := json.Unmarshal(bodyA2, &chunkA2); err != nil {
+ t.Fatal(err)
+ }
+ if len(chunkA2.Samples) != 2 {
+ t.Fatalf("hour A chunk regressed to %d samples after the retry, want still 2", len(chunkA2.Samples))
+ }
+
+ // Hour B must now be correctly merged.
+ bodyB, _, err := realStore.Get(ctx, metrics.RawKey(box, metrics.HourOf(hourBStart)))
+ if err != nil {
+ t.Fatalf("hour B chunk missing after retry: %v", err)
+ }
+ var chunkB metrics.RawChunk
+ if err := json.Unmarshal(bodyB, &chunkB); err != nil {
+ t.Fatal(err)
+ }
+ if len(chunkB.Samples) != 2 {
+ t.Fatalf("hour B chunk has %d samples, want 2 (merged)", len(chunkB.Samples))
+ }
+
+ dayBody, _, err := realStore.Get(ctx, metrics.DayKey(box, "2026-08-02"))
+ if err != nil {
+ t.Fatalf("day rollup missing: %v", err)
+ }
+ var day metrics.DayRollup
+ if err := json.Unmarshal(dayBody, &day); err != nil {
+ t.Fatal(err)
+ }
+ if n := day.Hours["08"].Host["cpu"].N; n != 2 {
+ t.Errorf("day rollup hour 08 cpu N=%d, want 2 (merged count)", n)
+ }
+ if n := day.Hours["09"].Host["cpu"].N; n != 2 {
+ t.Errorf("day rollup hour 09 cpu N=%d, want 2 (merged count)", n)
+ }
+}