amadan

carlos / platform Public

console: fleet meters read the real fleets/ prefix, drop label tie-break

Review fix round 1 on 31d8d5e (task-3, fleet box-hours):

CRITICAL — fleetMeters was reading metrics.DayKey(label, date) at the
platform root, but a door-backed fleet box publishes under
fleets/<fleet>/ (resolveStatusPrefix, internal/host/agent.go; the
edge write ACL admits only fleets/<own fleet>/**,
internal/edge/storedoor.go). Every fleet box's rollups were therefore
invisible to this meter. Fixed: read through
release.NewPrefixStore(s.cfg.Releases, "fleets/"+fleet+"/") per fleet,
same rebase the console's own fleet page already uses for heartbeats.

IMPORTANT — the label map is now keyed by (fleet, label), not label
alone, since two fleets' same-labeled boxes live at different prefixed
keys. Deleted the alphabetically-first collision tie-break and its log
line entirely: under prefixed reads it would have been silent data
loss for the losing fleet's box-hours, not a display-only ambiguity.

MINOR — rollupPresenceHours now clamps each hour's contribution to at
most 1.0 (math.Min): a mid-hour agent restart can land more than 60
samples in one hour bucket. Extended TestRollupPresenceHours with an
N=90 case.

MINOR — doc-comment rulings added to fleetMeters/statementsPass: the
hard-fail posture on a fleetMeters error is deliberate for the write
side (a billing meter silently going to zero is worse than the pass
not moving forward for one tick) and must NOT be inherited by Task 5's
read-side view; retagging a box mid-month re-prices the whole current
month (statements are create-or-replace, not an append-only ledger);
grandfathered bucket-backed fleet boxes without a fleets/ prefix are
not found by this meter until they republish under one.

TestStatementsPassFleetMeters now seeds rollups under fleets/fx/,
proving the real path; goldens (24/12) unchanged.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell ee969c7d3c42f54cf150ffdc0db0a57dbb333e02 parent 31d8d5e
2 files changed, +123 −59
  • internal/console/economicspass.go +105 −50
  • internal/console/economicspass_test.go +18 −9
diff --git a/internal/console/economicspass.go b/internal/console/economicspass.go
index 29935be43..cd8aff0f7 100644
--- a/internal/console/economicspass.go
+++ b/internal/console/economicspass.go
@@ -7,6 +7,7 @@ import (
"errors"
"fmt"
"log"
+ "math"
"net/http"
"os"
"strconv"
@@ -516,26 +517,50 @@ func (s *Server) foldStorageAndCensus(ctx context.Context, from, to time.Time, d
// - else — Host itself has no aggs at all for the hour — a flat 60-minute
// (1.0h) fallback, since a present-but-metric-less hour key still means
// the box was up and dialed in for that hour.
+//
+// Each hour's contribution is clamped to at most 1.0 (review, fix round 1):
+// a mid-hour agent restart can land more than 60 samples in one hour bucket
+// — mergeExistingChunk dedups exact-timestamp collisions only, not a
+// restart's fresh cadence landing alongside the old one — and an N above 60
+// must read as "fully present that hour", never as more than one hour billed
+// for one hour of wall time.
func rollupPresenceHours(d metrics.DayRollup) float64 {
var total float64
for _, hour := range d.Hours {
+ var minutes float64
if cpu, ok := hour.Host["cpu"]; ok {
- total += float64(cpu.N) / 60
- continue
+ minutes = float64(cpu.N)
+ } else if maxN := maxHostAggN(hour.Host); maxN > 0 {
+ minutes = float64(maxN)
+ } else {
+ minutes = 60
}
- maxN := 0
- for _, agg := range hour.Host {
- if agg.N > maxN {
- maxN = agg.N
- }
+ total += math.Min(1.0, minutes/60)
+ }
+ return total
+}
+
+// maxHostAggN returns the largest N across host's Agg entries, 0 for an
+// empty (or nil) map.
+func maxHostAggN(host map[string]metrics.Agg) int {
+ maxN := 0
+ for _, agg := range host {
+ if agg.N > maxN {
+ maxN = agg.N
}
- if maxN > 0 {
- total += float64(maxN) / 60
- continue
+ }
+ return maxN
+}
+
+// hasDeviceTag reports whether tags (FleetBox.Tags' comma-separated list)
+// carries the literal segment "device" — the device-vs-server meter split.
+func hasDeviceTag(tags string) bool {
+ for _, tag := range strings.Split(tags, ",") {
+ if tag == "device" {
+ return true
}
- total += 1.0
}
- return total
+ return false
}
// fleetMeters derives per-account fleet box-hours for [from, to]: for every
@@ -549,19 +574,53 @@ func rollupPresenceHours(d metrics.DayRollup) float64 {
// metrics/boxes/ once and attributes by UNIT (unitOwner), which has no
// concept of a fleet box at all — a fleet box's day rollup carries no units
// most of the time (see rollupPresenceHours's doc comment), only a host
-// aggregate, so folding this in there would mean threading a label->owner
-// map through sweepUsage's hot loop and its box-vs-fleet-box branching for a
-// meter family only two of usageSweep's several callers need at all. This
-// keeps sweepUsage and its existing tests completely untouched: the read
-// here is bounded to exactly the boxes that ARE fleet boxes (one Get per
-// fleet-box label per day in range; a day with no rollup is skipped, same
-// missing-means-skip convention sweepUsage itself uses).
+// aggregate, so folding this in there would mean threading a
+// (fleet,label)->owner map through sweepUsage's hot loop and its
+// box-vs-fleet-box branching for a meter family only two of usageSweep's
+// several callers need at all. This keeps sweepUsage and its existing tests
+// completely untouched: the read here is bounded to exactly the boxes that
+// ARE fleet boxes (one Get per (fleet,label) per day in range; a day with no
+// rollup is skipped, same missing-means-skip convention sweepUsage itself
+// uses).
+//
+// Fleet box rollups do NOT live at the platform root's metrics.DayKey(label,
+// date) — a door-backed fleet box's write ACL admits only
+// fleets/<own fleet>/** (internal/edge/storedoor.go's aclRead/aclWrite), and
+// resolveStatusPrefix (internal/host/agent.go) defaults every door-backed
+// box's publish prefix to "fleets/<fleet>/" for exactly that reason. This
+// reads each fleet's rollups through the same
+// release.NewPrefixStore(s.cfg.Releases, "fleets/"+fleet+"/") rebase the
+// console's own fleet page already uses for heartbeats
+// (loadFleetPage/fleets.go), with the plain metrics.DayKey(label, date)
+// inside it — so the map below is keyed by (fleet, label), not label alone.
//
-// Label collisions across fleets — the registry enforces per-fleet box-label
-// uniqueness only, not a global one — resolve to the alphabetically-first
-// fleet name (ListFleets already returns fleets ordered by name, so the
-// first owner claimed for a label during that ordered walk wins) with one
-// log line per pass per colliding label.
+// (fleet, label) keying also means a same-labeled box in two DIFFERENT
+// fleets is not a collision at all — each lives under its own fleet prefix
+// at a distinct object key, so both are read and billed independently.
+// (Review, fix round 1: an earlier version of this function keyed by label
+// alone and resolved a same-label clash to the alphabetically-first fleet —
+// which, under prefixed reads, would have been silent DATA LOSS for the
+// second fleet's box-hours, not a display-only tie-break. Removed entirely,
+// not just fixed, since the underlying ambiguity no longer exists.)
+//
+// Known gaps, recorded rather than fixed at modelling grade (review, fix
+// round 1):
+// - A fleet box provisioned before this prefixing existed — a grandfathered
+// bucket-backed box publishing straight to the platform root, or one an
+// operator hand-configured with CARLOS_STATUS_PREFIX="" — is invisible to
+// this meter: its rollups never appear under fleets/<fleet>/, so it
+// contributes zero fleet box-hours, and its sampled units (if any) fall
+// through unitOwner's unmatched case to release.OpsAccount like any other
+// unrecognized unit, same as a platform box's stray unit would. This
+// closes itself once the box (re)publishes under its fleet prefix — no
+// separate migration is needed.
+// - fleetMeters failing makes statementsPass itself fail (see
+// statementsPass's doc comment for why that's deliberate here).
+// - Retagging a box "device" mid-month re-prices the box's ENTIRE current
+// month the next time this pass runs, not just days after the retag —
+// statements are a create-OR-REPLACE snapshot of the whole month to
+// date, not an append-only ledger, so there is no "as of" boundary to
+// honor. Modelling-grade, matching every other meter this pass derives.
func (s *Server) fleetMeters(ctx context.Context, from, to time.Time) (map[string]map[string]float64, error) {
reg, err := s.openPassRegistry()
if err != nil {
@@ -576,33 +635,16 @@ func (s *Server) fleetMeters(ctx context.Context, from, to time.Time) (map[strin
acct string
device bool
}
- labelOwner := map[string]boxOwner{}
- labelFleet := map[string]string{} // label -> fleet that claimed it, for the collision log line
- loggedCollision := map[string]bool{}
+ type boxKey struct{ fleet, label string }
+ boxes := map[boxKey]boxOwner{}
for _, f := range fleets {
- boxes, err := reg.ListFleetBoxes(f.Name)
+ fboxes, err := reg.ListFleetBoxes(f.Name)
if err != nil {
return nil, fmt.Errorf("console: fleet meters: list boxes %s: %w", f.Name, err)
}
- for _, b := range boxes {
- if _, exists := labelOwner[b.Label]; exists {
- if !loggedCollision[b.Label] {
- loggedCollision[b.Label] = true
- log.Printf("console: fleet meters: box label %q claimed by both fleet %q and fleet %q; %q (alphabetically first) wins",
- b.Label, labelFleet[b.Label], f.Name, labelFleet[b.Label])
- }
- continue
- }
- device := false
- for _, tag := range strings.Split(b.Tags, ",") {
- if tag == "device" {
- device = true
- break
- }
- }
- labelOwner[b.Label] = boxOwner{acct: f.Account, device: device}
- labelFleet[b.Label] = f.Name
+ for _, b := range fboxes {
+ boxes[boxKey{fleet: f.Name, label: b.Label}] = boxOwner{acct: f.Account, device: hasDeviceTag(b.Tags)}
}
}
@@ -618,21 +660,22 @@ func (s *Server) fleetMeters(ctx context.Context, from, to time.Time) (map[strin
fromDay := from.UTC().Truncate(24 * time.Hour)
toDay := to.UTC().Truncate(24 * time.Hour)
- for label, owner := range labelOwner {
+ for bk, owner := range boxes {
meter := usage.MeterFleetBoxHoursServer
if owner.device {
meter = usage.MeterFleetBoxHoursDevice
}
+ fleetStore := release.NewPrefixStore(s.cfg.Releases, "fleets/"+bk.fleet+"/")
for d := fromDay; !d.After(toDay); d = d.AddDate(0, 0, 1) {
date := usage.DayOf(d)
- key := metrics.DayKey(label, date)
- raw, _, gerr := s.cfg.Releases.Get(ctx, key)
+ key := metrics.DayKey(bk.label, date)
+ raw, _, gerr := fleetStore.Get(ctx, key)
if gerr != nil {
continue // missing = skip, same convention as sweepUsage
}
var roll metrics.DayRollup
if jerr := json.Unmarshal(raw, &roll); jerr != nil {
- log.Printf("console: fleet meters: parse %s: %v", key, jerr)
+ log.Printf("console: fleet meters: parse fleets/%s/%s: %v", bk.fleet, key, jerr)
continue
}
touch(owner.acct)[meter] += rollupPresenceHours(roll)
@@ -657,6 +700,18 @@ func (s *Server) fleetMeters(ctx context.Context, from, to time.Time) (map[strin
// deleted_at), so this pass stops upserting its statement from the day of
// deletion onward — the account's last statement object simply stops moving
// forward, rather than the month closing it out explicitly.
+//
+// fleetMeters failing (a bad registry read, a bucket hiccup under a fleet
+// prefix) makes this WHOLE pass return an error rather than upsert
+// statements missing that account's fleet hours — deliberate (review, fix
+// round 1): a billing meter silently going to zero for a tick is worse than
+// this hourly pass simply not moving the statement forward that tick and
+// trying again next hour with the same last-good numbers still standing.
+// This is the WRITE side's posture only — Task 5's read side
+// (buildEconomicsView) must NOT inherit it: a live view degrading to "could
+// not read fleet hours" for one request is fine, refusing the whole
+// economics page over it is not, the same asymmetry sweepUsage's own doc
+// comment draws between its two callers.
func (s *Server) statementsPass(ctx context.Context, now time.Time) error {
reg, err := s.openPassRegistry()
if err != nil {
diff --git a/internal/console/economicspass_test.go b/internal/console/economicspass_test.go
index 202c468b2..aea0f1ee4 100644
--- a/internal/console/economicspass_test.go
+++ b/internal/console/economicspass_test.go
@@ -453,17 +453,21 @@ func TestStatementsPassRejectsInvalidStorageAccountSegment(t *testing.T) {
}
// TestRollupPresenceHours pins rollupPresenceHours's exact rule (task-3
-// brief): per present hour (a key in Hours), Host["cpu"].N minutes/60 when
-// present; else the max N across any other Host agg; else — Host itself
-// carrying no aggs at all — a flat 60-minute (1.0h) fallback. Two hours: one
-// with Host["cpu"].N=60 (-> 1.0h), one with an empty Host map (-> 1.0h
-// fallback); total 2.0.
+// brief, clamp added in review fix round 1): per present hour (a key in
+// Hours), Host["cpu"].N minutes/60 when present; else the max N across any
+// other Host agg; else — Host itself carrying no aggs at all — a flat
+// 60-minute (1.0h) fallback; each hour's contribution then clamps to at
+// most 1.0. Three hours: Host["cpu"].N=60 (-> 1.0h), an empty Host map
+// (-> 1.0h fallback), and Host["cpu"].N=90 — more than 60 samples in one
+// hour, e.g. a mid-hour agent restart — which must still clamp to 1.0h, not
+// 1.5h; total 3.0.
func TestRollupPresenceHours(t *testing.T) {
d := metrics.DayRollup{V: 1, Box: "bx", Date: "2026-08-10", Hours: map[string]metrics.HourAgg{
"09": {Host: map[string]metrics.Agg{"cpu": {N: 60}}},
"10": {Host: map[string]metrics.Agg{}},
+ "11": {Host: map[string]metrics.Agg{"cpu": {N: 90}}},
}}
- if got, want := rollupPresenceHours(d), 2.0; math.Abs(got-want) > 1e-9 {
+ if got, want := rollupPresenceHours(d), 3.0; math.Abs(got-want) > 1e-9 {
t.Fatalf("rollupPresenceHours = %v, want %v", got, want)
}
}
@@ -471,7 +475,11 @@ func TestRollupPresenceHours(t *testing.T) {
// TestStatementsPassFleetMeters: a fleet box's day rollups price into the
// owning account's fleet_box_hours_server/device meters, split by the
// literal "device" tag — bx-server (untagged) contributes to the server
-// meter, bx-pi (tags "pi,device") to the device meter.
+// meter, bx-pi (tags "pi,device") to the device meter. Rollups are written
+// under fleets/<fleet>/ — the SAME prefix a door-backed fleet box's write
+// ACL admits (internal/edge/storedoor.go aclWrite) and resolveStatusPrefix
+// (internal/host/agent.go) defaults a fleet box's publishing to — proving
+// fleetMeters reads the real path, not the platform root.
func TestStatementsPassFleetMeters(t *testing.T) {
fx := newConsoleFixture(t)
ctx := context.Background()
@@ -497,11 +505,12 @@ func TestStatementsPassFleetMeters(t *testing.T) {
t.Fatal(err)
}
+ fleetPrefix := "fleets/fx/"
serverHours := map[string]metrics.HourAgg{}
for h := 0; h < 24; h++ {
serverHours[hourKey(h)] = metrics.HourAgg{Host: map[string]metrics.Agg{"cpu": {N: 60}}}
}
- putStoreJSON(t, fx.rel, metrics.DayKey("bx-server", "2026-08-10"), metrics.DayRollup{
+ putStoreJSON(t, fx.rel, fleetPrefix+metrics.DayKey("bx-server", "2026-08-10"), metrics.DayRollup{
V: 1, Box: "bx-server", Date: "2026-08-10", Hours: serverHours,
})
@@ -509,7 +518,7 @@ func TestStatementsPassFleetMeters(t *testing.T) {
for h := 0; h < 12; h++ {
piHours[hourKey(h)] = metrics.HourAgg{Host: map[string]metrics.Agg{"cpu": {N: 60}}}
}
- putStoreJSON(t, fx.rel, metrics.DayKey("bx-pi", "2026-08-10"), metrics.DayRollup{
+ putStoreJSON(t, fx.rel, fleetPrefix+metrics.DayKey("bx-pi", "2026-08-10"), metrics.DayRollup{
V: 1, Box: "bx-pi", Date: "2026-08-10", Hours: piHours,
})