amadan

carlos / platform Public

relay: meter logs corruption and evicts dead apps (review fix)

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev d4653dcf6e2bfa40a25a3a141ab9078a49edb396 parent 439be99
2 files changed, +141 −12
  • internal/relay/meter.go +41 −9
  • internal/relay/meter_test.go +100 −3
diff --git a/internal/relay/meter.go b/internal/relay/meter.go
index 559a8a477..07b4abc9d 100644
--- a/internal/relay/meter.go
+++ b/internal/relay/meter.go
@@ -8,6 +8,7 @@ import (
"encoding/json"
"errors"
"os"
+ "strconv"
"sync"
"time"
@@ -60,13 +61,22 @@ type Meter struct {
mu sync.Mutex
box string
now func() time.Time
+ // Log receives diagnostics (a corrupt chunk found, etc); nil discards.
+ Log func(string, ...any)
cur map[Identity]*MeterApp
}
// NewMeter returns a Meter for box, using now for both sample timestamps
-// and hour selection (a func so tests control it).
-func NewMeter(box string, now func() time.Time) *Meter {
- return &Meter{box: box, now: now, cur: map[Identity]*MeterApp{}}
+// and hour selection (a func so tests control it), and logf for
+// diagnostics (nil discards them).
+func NewMeter(box string, now func() time.Time, logf func(string, ...any)) *Meter {
+ return &Meter{box: box, now: now, Log: logf, cur: map[Identity]*MeterApp{}}
+}
+
+func (m *Meter) log(format string, args ...any) {
+ if m.Log != nil {
+ m.Log(format, args...)
+ }
}
func (m *Meter) get(id Identity) *MeterApp {
@@ -99,11 +109,21 @@ func (m *Meter) SetAllocs(id Identity, n int) {
// Flush writes one sample of deltas since the previous flush, appended to
// this box's hourly object by read-append-CAS. The relay is the only
// writer of these objects, so the CAS only ever races with itself.
+//
+// An app with nothing to report — every delta zero and no open
+// allocations — is dropped from the sample and from the meter's live set,
+// so a burst of short-lived apps does not grow m.cur without bound. An app
+// with open allocations is kept (with zero deltas) even while idle, so its
+// allocation gauge keeps being reported.
func (m *Meter) Flush(ctx context.Context, s release.Store) error {
m.mu.Lock()
now := m.now()
sample := MeterSample{At: now, Apps: map[string]MeterApp{}}
for id, a := range m.cur {
+ if a.In == 0 && a.Out == 0 && a.Rejected == 0 && a.Allocs == 0 {
+ delete(m.cur, id)
+ continue
+ }
sample.Apps[id.Account+"/"+id.App] = *a
m.cur[id] = &MeterApp{Allocs: a.Allocs} // deltas reset, gauge kept
}
@@ -121,7 +141,12 @@ func (m *Meter) Flush(ctx context.Context, s release.Store) error {
case err != nil:
return err
default:
- if json.Unmarshal(body, &c) != nil {
+ if uerr := json.Unmarshal(body, &c); uerr != nil {
+ m.log("relay: meter chunk %s unreadable (%v); starting a fresh chunk, %d bytes discarded", key, uerr, len(body))
+ corruptKey := key + ".corrupt-" + strconv.FormatInt(now.Unix(), 10)
+ if _, perr := s.Put(ctx, corruptKey, body, ""); perr != nil {
+ m.log("relay: could not preserve corrupt meter chunk %s as %s: %v", key, corruptKey, perr)
+ }
c = MeterChunk{V: 1, Box: m.box, Hour: hour}
}
}
@@ -139,11 +164,14 @@ func (m *Meter) Flush(ctx context.Context, s release.Store) error {
}
// LoadWindow returns every sample from the previous and current hour's
-// objects, for rebuilding rolling-hour budgets after a restart.
-func LoadWindow(ctx context.Context, s release.Store, box string, now time.Time) ([]MeterSample, error) {
+// objects, for rebuilding rolling-hour budgets after a restart. An hour
+// whose chunk is corrupt is logged via logf (nil discards) and skipped
+// rather than failing the whole load.
+func LoadWindow(ctx context.Context, s release.Store, box string, now time.Time, logf func(string, ...any)) ([]MeterSample, error) {
var out []MeterSample
for _, h := range []string{metrics.HourOf(now.Add(-time.Hour)), metrics.HourOf(now)} {
- body, _, err := s.Get(ctx, MeterKey(box, h))
+ key := MeterKey(box, h)
+ body, _, err := s.Get(ctx, key)
if errors.Is(err, os.ErrNotExist) {
continue
}
@@ -151,9 +179,13 @@ func LoadWindow(ctx context.Context, s release.Store, box string, now time.Time)
return nil, err
}
var c MeterChunk
- if json.Unmarshal(body, &c) == nil {
- out = append(out, c.Samples...)
+ if uerr := json.Unmarshal(body, &c); uerr != nil {
+ if logf != nil {
+ logf("relay: meter chunk %s unreadable (%v); budgets rebuilt without it", key, uerr)
+ }
+ continue
}
+ out = append(out, c.Samples...)
}
return out, nil
}
diff --git a/internal/relay/meter_test.go b/internal/relay/meter_test.go
index 16bcea5bb..2d77a2ad0 100644
--- a/internal/relay/meter_test.go
+++ b/internal/relay/meter_test.go
@@ -3,6 +3,8 @@ package relay
import (
"context"
"encoding/json"
+ "fmt"
+ "strings"
"testing"
"time"
)
@@ -10,7 +12,7 @@ import (
func TestMeterFlushAppendsDeltas(t *testing.T) {
s := newMemStore(t)
now := time.Date(2026, 9, 7, 12, 30, 0, 0, time.UTC)
- m := NewMeter("box1", func() time.Time { return now })
+ m := NewMeter("box1", func() time.Time { return now }, nil)
id := Identity{"abc", "meet"}
m.Add(id, 100, 200, 5)
if err := m.Flush(context.Background(), s); err != nil {
@@ -31,11 +33,106 @@ func TestMeterFlushAppendsDeltas(t *testing.T) {
func TestLoadWindowSpansTwoHours(t *testing.T) {
s := newMemStore(t)
now := time.Date(2026, 9, 7, 12, 1, 0, 0, time.UTC)
- m := NewMeter("box1", func() time.Time { return now.Add(-30 * time.Minute) }) // 11:31
+ m := NewMeter("box1", func() time.Time { return now.Add(-30 * time.Minute) }, nil) // 11:31
m.Add(Identity{"abc", "meet"}, 7, 0, 0)
m.Flush(context.Background(), s)
- samples, err := LoadWindow(context.Background(), s, "box1", now)
+ samples, err := LoadWindow(context.Background(), s, "box1", now, nil)
if err != nil || len(samples) != 1 || samples[0].Apps["abc/meet"].In != 7 {
t.Fatalf("window = %+v %v", samples, err)
}
}
+
+// TestMeterFlushPreservesCorruptChunk checks that a Flush landing on a
+// chunk that fails to unmarshal doesn't just clobber it: the unreadable
+// bytes are preserved as a ".corrupt-<unix>" sibling, a fresh chunk is
+// written in the real key's place, and the corruption is logged.
+func TestMeterFlushPreservesCorruptChunk(t *testing.T) {
+ s := newMemStore(t)
+ now := time.Date(2026, 9, 7, 12, 30, 0, 0, time.UTC)
+ key := MeterKey("box1", "2026-09-07-12")
+ if _, err := s.Put(context.Background(), key, []byte("not json"), ""); err != nil {
+ t.Fatal(err)
+ }
+
+ var logs []string
+ m := NewMeter("box1", func() time.Time { return now }, func(format string, args ...any) {
+ logs = append(logs, fmt.Sprintf(format, args...))
+ })
+ m.Add(Identity{"abc", "meet"}, 1, 2, 0)
+ if err := m.Flush(context.Background(), s); err != nil {
+ t.Fatal(err)
+ }
+
+ body, _, err := s.Get(context.Background(), key)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var c MeterChunk
+ if err := json.Unmarshal(body, &c); err != nil {
+ t.Fatalf("chunk at %s still unreadable: %v", key, err)
+ }
+ if len(c.Samples) != 1 {
+ t.Fatalf("chunk = %+v", c)
+ }
+
+ keys, err := s.List(context.Background(), "metrics/relay/box1/")
+ if err != nil {
+ t.Fatal(err)
+ }
+ found := false
+ for _, k := range keys {
+ if strings.Contains(k, ".corrupt-") {
+ found = true
+ }
+ }
+ if !found {
+ t.Fatalf("no .corrupt- sibling among %v", keys)
+ }
+ if len(logs) == 0 {
+ t.Fatal("expected a log message about the corrupt chunk")
+ }
+}
+
+// TestMeterEvictsDeadApps checks that an app with nothing left to report
+// (zero deltas, zero allocations) drops out of the meter's live set — and
+// out of the sample — after being flushed once, while an app still holding
+// open allocations keeps being reported even with zero deltas.
+func TestMeterEvictsDeadApps(t *testing.T) {
+ s := newMemStore(t)
+ now := time.Date(2026, 9, 7, 12, 30, 0, 0, time.UTC)
+ m := NewMeter("box1", func() time.Time { return now }, nil)
+ a := Identity{"abc", "meet"}
+ b := Identity{"abc", "docs"}
+ m.Add(a, 1, 2, 0)
+ m.Add(b, 3, 4, 0)
+ m.SetAllocs(a, 1)
+ m.SetAllocs(b, 0)
+
+ if err := m.Flush(context.Background(), s); err != nil {
+ t.Fatal(err)
+ }
+ if err := m.Flush(context.Background(), s); err != nil {
+ t.Fatal(err)
+ }
+
+ body, _, err := s.Get(context.Background(), MeterKey("box1", "2026-09-07-12"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ var c MeterChunk
+ if err := json.Unmarshal(body, &c); err != nil {
+ t.Fatal(err)
+ }
+ if len(c.Samples) != 2 {
+ t.Fatalf("chunk = %+v", c)
+ }
+
+ second := c.Samples[1]
+ appA, ok := second.Apps["abc/meet"]
+ if !ok || appA.Allocs != 1 || appA.In != 0 || appA.Out != 0 || appA.Rejected != 0 {
+ t.Fatalf("second sample abc/meet = %+v ok=%v", appA, ok)
+ }
+ if _, ok := second.Apps["abc/docs"]; ok {
+ t.Fatalf("second sample still has dead app abc/docs: %+v", second.Apps)
+ }
+}