amadan

carlos / platform Public

usage: close the month, and catch up the one before it

statementsPass derived one month from now and never revisited it, but
host metrics roll an hour up after it ends, so a month's final hours
land after nothing is updating that month any more. A bill drawn from
that statement is short by up to a day.

It now runs two periods on billPass's pattern, and past the same
three-day grace the previous month closes and stops being recomputed.
Billing reads only closed statements.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011cEe5te3cyEcdF3JL2rN5M
Paul Campbell pushed by paul@keymail.dev 5c75701cd683010fcb8564301de99d39e41a216f parent 170306a
4 files changed, +255 −32
  • internal/console/economicspass.go +74 −31
  • internal/console/economicspass_test.go +83 −0
  • internal/usage/passes_test.go +81 −0
  • internal/usage/statement.go +17 −1
diff --git a/internal/console/economicspass.go b/internal/console/economicspass.go
index 25f1eb36d..627b15a29 100644
--- a/internal/console/economicspass.go
+++ b/internal/console/economicspass.go
@@ -738,8 +738,9 @@ func hasDeviceTag(tags string) bool {
// 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).
+// - fleetMeters failing makes statementsForMonth itself fail for that
+// period (see statementsForMonth'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
@@ -808,16 +809,68 @@ func (s *Server) fleetMeters(ctx context.Context, from, to time.Time) (map[strin
return out, nil
}
-// statementsPass folds the month's metrics, edge and storage days (plus
+// statementCloseDayGrace is the day of the following month after which the
+// previous month stops being recomputed. Matches billCEDayGrace: the same
+// argument (late-arriving data has landed by now) and no reason to differ.
+const statementCloseDayGrace = 3
+
+// statementsPass runs statementsForMonth for two periods on billPass's
+// pattern: this month to date, and the previous month while it is still
+// inside its statementCloseDayGrace catch-up window. Host metrics only roll
+// an hour up after it ends, so a month's last hours land in the next
+// month's objects, after now has already moved past it — without this
+// second period those hours would never reach the closed statement. Once
+// past the grace, the previous month's period is marked closing, and
+// UpsertStatement (its CAS loop, not this caller — see its doc comment)
+// refuses every write to a statement once it is closed. Each period's
+// failure is logged and does not skip the other, matching billPass and
+// economicsPass's stage-degrades-independently posture.
+func (s *Server) statementsPass(ctx context.Context, now time.Time) error {
+ reg, err := s.openPassRegistry()
+ if err != nil {
+ return fmt.Errorf("console: statements pass: %w", err)
+ }
+ routes, err := reg.List()
+ if err != nil {
+ return fmt.Errorf("console: statements pass: %w", err)
+ }
+
+ nowUTC := now.UTC()
+ firstOfThisMonth := time.Date(nowUTC.Year(), nowUTC.Month(), 1, 0, 0, 0, 0, time.UTC)
+ firstOfLastMonth := firstOfThisMonth.AddDate(0, -1, 0)
+
+ periods := []struct {
+ month string
+ start, end time.Time
+ closing bool
+ }{
+ {usage.MonthOf(nowUTC), firstOfThisMonth, nowUTC, false},
+ {usage.MonthOf(firstOfLastMonth), firstOfLastMonth, firstOfThisMonth.AddDate(0, 0, -1), nowUTC.Day() > statementCloseDayGrace},
+ }
+
+ for _, p := range periods {
+ if err := s.statementsForMonth(ctx, p.month, p.start, p.end, nowUTC, p.closing, routes); err != nil {
+ log.Printf("console: statements pass: %s: %v", p.month, err)
+ }
+ }
+ return nil
+}
+
+// statementsForMonth folds one month's metrics, edge and storage days (plus
// census, for app_months, and fleet box-hours, for fleet_box_hours_server/
// device) into one usage.Statement per account, create-or-replace via
// usage.UpsertStatement — safe to run repeatedly as new days' objects
-// appear. It is a thin caller: open the registry (refusing early — see
-// openPassRegistry — since this is the WRITE pass), sweep the whole month to
-// date (sweepUsage), fold in storage/census meters (foldStorageAndCensus),
-// fold in fleet meters (fleetMeters — a union merge, since an account with
-// ONLY fleet usage and no platform routes at all must still get every meter
-// key it's owed), then upsert.
+// appear. statementsPass calls this once per period (this month, and the
+// previous month while it is still inside its catch-up grace) so that a
+// month's final hours, which host metrics only roll up after the hour ends,
+// still reach that month's statement even once now has crossed into the
+// next one. It is a thin worker: sweep the period (sweepUsage), fold in
+// storage/census meters (foldStorageAndCensus), fold in fleet meters
+// (fleetMeters — a union merge, since an account with ONLY fleet usage and
+// no platform routes at all must still get every meter key it's owed), then
+// upsert. daysInMonth is derived from the period's own month (start), not
+// from now — the previous-month period must get its own month's day count,
+// not this month's.
//
// Known shortfall, recorded rather than fixed at modelling grade: an account
// deleted mid-month drops out of s.store.Accounts() (its query filters
@@ -826,41 +879,30 @@ func (s *Server) fleetMeters(ctx context.Context, from, to time.Time) (map[strin
// 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
+// prefix) makes this WHOLE period 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
+// statementsPass logs and moves on to the other period rather than
+// propagating this. 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 {
- return fmt.Errorf("console: statements pass: %w", err)
- }
- routes, err := reg.List()
- if err != nil {
- return fmt.Errorf("console: statements pass: %w", err)
- }
-
- nowUTC := now.UTC()
- y, mo, _ := nowUTC.Date()
+func (s *Server) statementsForMonth(ctx context.Context, month string, start, end, now time.Time, closing bool, routes []registry.Route) error {
+ y, mo, _ := start.Date()
daysInMonth := time.Date(y, mo+1, 0, 0, 0, 0, 0, time.UTC).Day()
- month := usage.MonthOf(nowUTC)
- monthStart := time.Date(y, mo, 1, 0, 0, 0, 0, time.UTC)
- sweep, err := s.sweepUsage(ctx, monthStart, nowUTC, routes)
+ sweep, err := s.sweepUsage(ctx, start, end, routes)
if err != nil {
return fmt.Errorf("console: statements pass: %w", err)
}
meters := sweep.PerAccount // account -> meter -> value; folded into below
- if err := s.foldStorageAndCensus(ctx, monthStart, nowUTC, meters); err != nil {
+ if err := s.foldStorageAndCensus(ctx, start, end, meters); err != nil {
return fmt.Errorf("console: statements pass: %w", err)
}
- fleetMM, err := s.fleetMeters(ctx, monthStart, nowUTC)
+ fleetMM, err := s.fleetMeters(ctx, start, end)
if err != nil {
return fmt.Errorf("console: statements pass: %w", err)
}
@@ -913,7 +955,8 @@ func (s *Server) statementsPass(ctx context.Context, now time.Time) error {
CoveredDays: coveredDays,
SourceDays: sourceDays,
DaysInMonth: daysInMonth,
- UpdatedAt: nowUTC,
+ UpdatedAt: now,
+ Closed: closing,
}
if err := usage.UpsertStatement(ctx, s.cfg.Releases, st); err != nil {
return fmt.Errorf("console: statements pass: upsert %s: %w", acct, err)
diff --git a/internal/console/economicspass_test.go b/internal/console/economicspass_test.go
index 97d84e2e9..2bd5c9cc6 100644
--- a/internal/console/economicspass_test.go
+++ b/internal/console/economicspass_test.go
@@ -1021,3 +1021,86 @@ func TestSweepFallsBackForRouteMissingFromCensus(t *testing.T) {
t.Errorf("acme cpu_core_hours = %v, want 0.5", got)
}
}
+
+// Data that arrives after a month ends must still reach that month's
+// statement, until the month closes.
+func TestStatementsPassUpdatesPreviousMonth(t *testing.T) {
+ fx := newConsoleFixture(t)
+ ctx := context.Background()
+
+ putStoreJSON(t, fx.rel, metrics.DayKey("box1", "2026-08-31"), metrics.DayRollup{
+ V: 1, Box: "box1", Date: "2026-08-01",
+ Hours: map[string]metrics.HourAgg{
+ "00": {Units: map[string]map[string]metrics.Agg{
+ "carlos-app@a.example.com.service": {"cpu": {Avg: 0.5, N: 60}},
+ }},
+ },
+ })
+
+ // 1 September: the August rollup only became readable now.
+ now := time.Date(2026, 9, 1, 6, 0, 0, 0, time.UTC)
+ if err := fx.srv.statementsPass(ctx, now); err != nil {
+ t.Fatal(err)
+ }
+
+ raw, _, err := fx.rel.Get(ctx, usage.StatementKey(fixtureAcct, "2026-08"))
+ if err != nil {
+ t.Fatalf("August statement missing on 1 September: %v", err)
+ }
+ var st usage.Statement
+ if err := json.Unmarshal(raw, &st); err != nil {
+ t.Fatal(err)
+ }
+ if got := st.Meters[usage.MeterCPUCoreHours]; math.Abs(got-0.5) > 1e-9 {
+ t.Errorf("august cpu_core_hours = %v, want 0.5", got)
+ }
+ if st.Closed {
+ t.Error("August must not be closed on 1 September — inside the grace")
+ }
+}
+
+// Past the grace, the month closes and stops changing.
+func TestStatementsPassClosesAfterGrace(t *testing.T) {
+ fx := newConsoleFixture(t)
+ ctx := context.Background()
+
+ now := time.Date(2026, 9, 5, 6, 0, 0, 0, time.UTC)
+ if err := fx.srv.statementsPass(ctx, now); err != nil {
+ t.Fatal(err)
+ }
+ raw, _, err := fx.rel.Get(ctx, usage.StatementKey(fixtureAcct, "2026-08"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ var st usage.Statement
+ if err := json.Unmarshal(raw, &st); err != nil {
+ t.Fatal(err)
+ }
+ if !st.Closed {
+ t.Fatal("August must be closed on 5 September — past the grace")
+ }
+
+ // Late data after closure must not move a closed month.
+ putStoreJSON(t, fx.rel, metrics.DayKey("box1", "2026-08-31"), metrics.DayRollup{
+ V: 1, Box: "box1", Date: "2026-08-01",
+ Hours: map[string]metrics.HourAgg{
+ "00": {Units: map[string]map[string]metrics.Agg{
+ "carlos-app@a.example.com.service": {"cpu": {Avg: 0.5, N: 60}},
+ }},
+ },
+ })
+ if err := fx.srv.statementsPass(ctx, now.AddDate(0, 0, 1)); err != nil {
+ t.Fatal(err)
+ }
+ raw2, _, err := fx.rel.Get(ctx, usage.StatementKey(fixtureAcct, "2026-08"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ var st2 usage.Statement
+ if err := json.Unmarshal(raw2, &st2); err != nil {
+ t.Fatal(err)
+ }
+ if got := st2.Meters[usage.MeterCPUCoreHours]; got != 0 {
+ t.Errorf("closed August moved to %v — a closed month must not change", got)
+ }
+}
diff --git a/internal/usage/passes_test.go b/internal/usage/passes_test.go
index 5a68344c3..6ddd33976 100644
--- a/internal/usage/passes_test.go
+++ b/internal/usage/passes_test.go
@@ -115,3 +115,84 @@ func TestMeterCompleteNeedsEveryDay(t *testing.T) {
t.Error("a statement with no SourceDays must not report complete")
}
}
+
+func TestUpsertStatementRefusesClosedMonth(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ ctx := context.Background()
+
+ closed := Statement{V: 1, Account: "acme", Month: "2026-08", Closed: true,
+ Meters: map[string]float64{MeterInstanceHours: 10}}
+ if err := UpsertStatement(ctx, s, closed); err != nil {
+ t.Fatal(err)
+ }
+ // A writer that read the month before it closed now tries to write.
+ late := Statement{V: 1, Account: "acme", Month: "2026-08",
+ Meters: map[string]float64{MeterInstanceHours: 999}}
+ if err := UpsertStatement(ctx, s, late); err != nil {
+ t.Fatalf("a refused write is not an error: %v", err)
+ }
+ raw, _, err := s.Get(ctx, StatementKey("acme", "2026-08"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ var got Statement
+ if err := json.Unmarshal(raw, &got); err != nil {
+ t.Fatal(err)
+ }
+ if got.Meters[MeterInstanceHours] != 10 {
+ t.Errorf("closed statement moved to %v — closure is not immutable", got.Meters[MeterInstanceHours])
+ }
+}
+
+// closeDuringPut closes the statement the first time a Put is attempted, so
+// the writer's Put loses its CAS race and retries — and the retry's Get must
+// then see Closed and refuse. This is the interleaving two consoles produce.
+type closeDuringPut struct {
+ release.Store
+ fired bool
+ ctx context.Context
+ st Statement
+}
+
+func (c *closeDuringPut) Put(ctx context.Context, key string, body []byte, etag string) (string, error) {
+ if !c.fired {
+ c.fired = true
+ closed := c.st
+ closed.Closed = true
+ closed.Meters = map[string]float64{MeterInstanceHours: 10}
+ if err := UpsertStatement(c.ctx, c.Store, closed); err != nil {
+ return "", err
+ }
+ }
+ return c.Store.Put(ctx, key, body, etag)
+}
+
+func TestUpsertStatementLosesToAConcurrentClose(t *testing.T) {
+ ctx := context.Background()
+ base := release.NewDirStore(t.TempDir())
+ open := Statement{V: 1, Account: "acme", Month: "2026-08",
+ Meters: map[string]float64{MeterInstanceHours: 1}}
+ if err := UpsertStatement(ctx, base, open); err != nil {
+ t.Fatal(err)
+ }
+
+ s := &closeDuringPut{Store: base, ctx: ctx, st: open}
+ late := Statement{V: 1, Account: "acme", Month: "2026-08",
+ Meters: map[string]float64{MeterInstanceHours: 999}}
+ if err := UpsertStatement(ctx, s, late); err != nil {
+ t.Fatalf("a refused write is not an error: %v", err)
+ }
+
+ raw, _, err := base.Get(ctx, StatementKey("acme", "2026-08"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ var got Statement
+ if err := json.Unmarshal(raw, &got); err != nil {
+ t.Fatal(err)
+ }
+ if got.Meters[MeterInstanceHours] != 10 {
+ t.Errorf("instance_hours = %v, want 10 — a close between read and write must win",
+ got.Meters[MeterInstanceHours])
+ }
+}
diff --git a/internal/usage/statement.go b/internal/usage/statement.go
index 8a0cdb6b3..2ff60cd82 100644
--- a/internal/usage/statement.go
+++ b/internal/usage/statement.go
@@ -31,6 +31,11 @@ type Statement struct {
SourceDays map[string]int `json:"source_days,omitempty"`
DaysInMonth int `json:"days_in_month"`
UpdatedAt time.Time `json:"updated_at"`
+ // Closed marks a month that will not be recomputed again: past the
+ // grace, every source has had time to land its final hours. Billing
+ // reads only closed statements, so a bill cannot be drawn from a month
+ // still receiving data.
+ Closed bool `json:"closed,omitempty"`
}
// StatementKey is the bucket key for an account's month statement.
@@ -52,12 +57,23 @@ func UpsertStatement(ctx context.Context, s release.Store, st Statement) error {
return err
}
for attempt := 0; attempt < 3; attempt++ {
- _, etag, err := s.Get(ctx, key)
+ raw, etag, err := s.Get(ctx, key)
if err != nil {
if !errors.Is(err, os.ErrNotExist) {
return err
}
etag = ""
+ } else {
+ // Closure is enforced HERE, against the same read that
+ // produced the ETag we are about to write against — a check
+ // in the caller races with another console's close.
+ var prev Statement
+ if jerr := json.Unmarshal(raw, &prev); jerr != nil {
+ return fmt.Errorf("usage: %s: %w", key, jerr)
+ }
+ if prev.Closed {
+ return nil // a closed month never changes again
+ }
}
if _, err := s.Put(ctx, key, body, etag); err == nil {
return nil