Fix: make the ledger agree with the ring, rotate archive segments, and cache past day files

core/cost/ledger/daycache_test.go

@@ -0,0 +1,145 @@
+package ledger
+
+import (
+	"context"
+	"encoding/json"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/rossoctl/cortex/core/cost/event"
+	"github.com/rossoctl/cortex/core/pipeline"
+)
+
+// The tests in this file pin issue #1566 item 4. Query re-decoded every day file in the
+// window on every call: agentop polls today each minute and 7d/month each five, so on
+// the 31st the month span decoded 31 files per poll — and past days almost never change.
+// The fix caches a decoded past day keyed on the file's size and mtime; the current day
+// is never cached because its file is still being appended to.
+
+// queryAt runs one Query over a window around the ledger's pinned clock.
+func queryAt(t *testing.T, w *Writer) ([]Row, Caveats) {
+	t.Helper()
+	rows, caveats, err := w.Query(context.Background(), at.Add(-time.Hour), at.Add(time.Hour))
+	if err != nil {
+		t.Fatalf("Query: %v", err)
+	}
+	return rows, caveats
+}
+
+// TestQuery_APastDayIsServedWithoutReopeningTheFile observes the cache through
+// permissions rather than timing: a past day already decoded answers from the cache,
+// where validation only needs Stat, while an uncached read must Open the file. Running
+// as root defeats the chmod and the test then passes vacuously — it is red only where
+// Open actually fails, which is every non-root laptop and CI runner this targets.
+//
+// Two writers, one day apart: the first files rows under the pinned clock, the second
+// — a day later — is the reader whose "today" the filed day is past. One writer cannot
+// play both parts: the store decides "past" from the writer's own clock, so to the
+// writer that filed them, the rows are today's and are never cached.
+func TestQuery_APastDayIsServedWithoutReopeningTheFile(t *testing.T) {
+	dir := t.TempDir()
+	filed := newTestWriter(t, dir, func() time.Time { return at })
+	filed.Record("s1", costedEvent(t, "api.example", "m", 0.01, 10, 10))
+	if err := filed.Flush(); err != nil {
+		t.Fatalf("Flush: %v", err)
+	}
+	reader := newTestWriter(t, dir, func() time.Time { return at.Add(24 * time.Hour) })
+
+	if rows, c := queryAt(t, reader); len(rows) != 1 || c.UnreadableDays != 0 {
+		t.Fatalf("first query: %d rows, UnreadableDays %d; want 1, 0", len(rows), c.UnreadableDays)
+	}
+
+	path := reader.store.path(at)
+	if err := os.Chmod(path, 0); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = os.Chmod(path, 0o600) })
+
+	rows, c := queryAt(t, reader)
+	if len(rows) != 1 || c.UnreadableDays != 0 {
+		t.Fatalf("second query: %d rows, UnreadableDays %d; want 1, 0 — a past day already "+
+			"decoded must be served from the cache instead of re-opened and re-decoded on "+
+			"every poll (issue #1566)", len(rows), c.UnreadableDays)
+	}
+}
+
+// TestQuery_APastDayCacheIsInvalidatedWhenTheFileChanges keeps the cache honest: the
+// key is the file's size and mtime, so a past day that changed is re-read, not served
+// stale. A writer still pinned to the filed day appends to it — the smallest way to
+// change a past day's file through the writer's own path.
+func TestQuery_APastDayCacheIsInvalidatedWhenTheFileChanges(t *testing.T) {
+	dir := t.TempDir()
+	filed := newTestWriter(t, dir, func() time.Time { return at })
+	filed.Record("s1", costedEvent(t, "api.example", "m", 0.01, 10, 10))
+	if err := filed.Flush(); err != nil {
+		t.Fatalf("Flush: %v", err)
+	}
+	reader := newTestWriter(t, dir, func() time.Time { return at.Add(24 * time.Hour) })
+	if rows, _ := queryAt(t, reader); len(rows) != 1 {
+		t.Fatalf("first query: %d rows, want 1", len(rows))
+	}
+
+	filed.Record("s2", costedEvent(t, "api2.example", "m", 0.02, 20, 20))
+	if err := filed.Flush(); err != nil {
+		t.Fatalf("Flush: %v", err)
+	}
+
+	rows, _ := queryAt(t, reader)
+	if len(rows) != 2 {
+		t.Fatalf("second query: %d rows, want 2 — the cache is keyed on size and mtime, so "+
+			"a past day whose file changed must be decoded again", len(rows))
+	}
+}
+
+// TestQuery_TodaysFileIsNeverCached guards the other half: the current day's file is
+// still being appended to, so it must be re-read on every query. Rows recorded after a
+// first query must be visible to the second.
+func TestQuery_TodaysFileIsNeverCached(t *testing.T) {
+	dir := t.TempDir()
+	w := newTestWriter(t, dir, time.Now)
+
+	nowEvent := func(host string) *pipeline.SessionEvent {
+		rec, err := json.Marshal(event.Event{
+			CostUSD: 0.01, Settled: true,
+			Source: event.SourceUsageFallback, Provenance: "bundled",
+		})
+		if err != nil {
+			t.Fatalf("marshal: %v", err)
+		}
+		return &pipeline.SessionEvent{
+			At: time.Now(), Phase: pipeline.SessionResponse, StatusCode: 200, Host: host,
+			Inference: &pipeline.InferenceExtension{
+				Model: "m", InputTokens: 10, OutputTokens: 10, TotalTokens: 20,
+			},
+			Plugins: map[string]json.RawMessage{event.Key: rec},
+		}
+	}
+
+	queryNow := func() []Row {
+		t.Helper()
+		now := time.Now()
+		rows, _, err := w.Query(context.Background(), now.Add(-time.Minute), now.Add(time.Minute))
+		if err != nil {
+			t.Fatalf("Query: %v", err)
+		}
+		return rows
+	}
+
+	w.Record("s1", nowEvent("api.example"))
+	if err := w.Flush(); err != nil {
+		t.Fatalf("Flush: %v", err)
+	}
+	if rows := queryNow(); len(rows) != 1 {
+		t.Fatalf("first query: %d rows, want 1", len(rows))
+	}
+
+	w.Record("s2", nowEvent("api2.example"))
+	if err := w.Flush(); err != nil {
+		t.Fatalf("Flush: %v", err)
+	}
+	if rows := queryNow(); len(rows) != 2 {
+		t.Fatalf("second query: %d rows, want 2 — today's file is still being written, so "+
+			"no query may serve it from a cache", len(rows))
+	}
+}

core/cost/ledger/parity_test.go

@@ -0,0 +1,76 @@
+package ledger
+
+import (
+	"testing"
+	"time"
+
+	"github.com/rossoctl/cortex/core/pipeline"
+)
+
+// The two tests in this file pin the ledger's agreement with the ring (issue #1566 items
+// 1 and 2). The ring serves the duration windows; the ledger serves today/7d/month. The
+// same traffic must count the same way on both, or the two surfaces an operator reads
+// side by side disagree about coverage and about whether the proxy is failing.
+
+// TestRecord_ZeroUsageSuccessIsPriceable is the coverage-gap case. A truncated
+// OpenAI-dialect stream answers 200 with a model and no usage block — every counter
+// lives on the final chunk that never arrived. inference-parser publishes no cost
+// record for it, so the ledger's old rule ("model and a non-zero token count") left it
+// out of PriceableRequests while the ring counted it: nine priced requests plus one of
+// these read "9/9 priced" on today and "9/10, and here is the gap" on the 1h window.
+func TestRecord_ZeroUsageSuccessIsPriceable(t *testing.T) {
+	dir := t.TempDir()
+	w := newTestWriter(t, dir, func() time.Time { return at })
+
+	for i := 0; i < 9; i++ {
+		w.Record("s1", costedEvent(t, "api.example", "m", 0.01, 100, 100))
+	}
+	// The truncated stream: recognised, successful, and carrying no usage at all.
+	w.Record("s1", &pipeline.SessionEvent{
+		At: at, Phase: pipeline.SessionResponse, StatusCode: 200, Host: "api.example",
+		Inference: &pipeline.InferenceExtension{Model: "m"},
+	})
+	if err := w.Flush(); err != nil {
+		t.Fatalf("Flush: %v", err)
+	}
+
+	var priceable int64
+	for _, r := range readAllRows(t, dir) {
+		priceable += r.PriceableRequests
+	}
+	if priceable != 10 {
+		t.Errorf("PriceableRequests = %d, want 10: the ring's window=1h counts a successful "+
+			"response with a model and no usage as priceable and names it in UnpricedBy. A "+
+			"denominator of 9 here reads \"9/9 priced\" on today against \"9/10\" on 1h — full "+
+			"coverage over a gap the other surface reports", priceable)
+	}
+}
+
+// TestRecord_DestinationPrivateIsNotAnError is the error-rule case. The egress guard's
+// 403 destination_private refusal is the guard doing what the operator configured, and
+// the ring's countsAsError excludes it — with a parsed model, the ledger still writes
+// the row (the request happened), so its error count must follow the same rule or the
+// guard's own refusals read as a failing proxy on today but not on 1h.
+func TestRecord_DestinationPrivateIsNotAnError(t *testing.T) {
+	dir := t.TempDir()
+	w := newTestWriter(t, dir, func() time.Time { return at })
+
+	w.Record("s1", &pipeline.SessionEvent{
+		At: at, Phase: pipeline.SessionResponse, StatusCode: 403, Host: "private.example",
+		Inference: &pipeline.InferenceExtension{Model: "m"},
+		Error:     &pipeline.EventError{Kind: pipeline.ErrorKindDestinationPrivate},
+	})
+	if err := w.Flush(); err != nil {
+		t.Fatalf("Flush: %v", err)
+	}
+
+	rows := readAllRows(t, dir)
+	if len(rows) != 1 {
+		t.Fatalf("ledger holds %d rows, want 1", len(rows))
+	}
+	if rows[0].Errors != 0 {
+		t.Errorf("Errors = %d, want 0: the ring excludes ErrorKindDestinationPrivate from "+
+			"its error count — the guard refusing a private destination is not a failing "+
+			"proxy. Counting it here made today report an error 1h did not", rows[0].Errors)
+	}
+}

core/cost/ledger/store.go

@@ -10,6 +10,7 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/rossoctl/cortex/core/cost/usage"
@@ -110,8 +111,33 @@ type store struct {
 	// do it — a row near midnight is filed under a date no query for that local day ever
 	// visits: written, retained for 30 days, and invisible to every read.
 	loc *time.Location
+	// now is the writer's clock, so the day cache's "is this day still being written"
+	// answers at the same instant the writer files rows by. Set by the Writer after
+	// newStore; time.Now for any other constructor.
+	now func() time.Time
+	// days caches decoded PAST day files, keyed by the day (#1566). Query re-decoded
+	// every day file in the window on every call: agentop polls today each minute and
+	// 7d/month each five, so on the 31st the month span decoded 31 files per poll, and
+	// past days almost never change. The current day is never cached — its file is
+	// still being appended to. Entries are validated against the file's size and mtime
+	// before use, so a changed past day is re-read, not served stale. Guarded by mu.
+	mu   sync.Mutex
+	days map[time.Time]dayCacheEntry
 }
 
+// dayCacheEntry is one decoded past day, with the file facts that validate it.
+type dayCacheEntry struct {
+	rows    []Row
+	issues  dayIssues
+	size    int64
+	modTime time.Time
+}
+
+// dayCacheMax bounds the cache. A month window holds 31 entries; 128 covers a
+// retention of four months, and the eviction below drops the oldest day rather than
+// erroring or blocking.
+const dayCacheMax = 128
+
 // newStore prepares dir, creating it if needed. A nil loc means time.Local.
 func newStore(dir string, retainDays int, loc *time.Location) (*store, error) {
 	if dir == "" {
@@ -143,7 +169,7 @@ func newStore(dir string, retainDays int, loc *time.Location) (*store, error) {
 	if err := os.MkdirAll(dir, dirMode); err != nil {
 		return nil, fmt.Errorf("costledger: cannot create %s: %w", dir, err)
 	}
-	return &store{dir: dir, retainDays: retainDays, loc: loc}, nil
+	return &store{dir: dir, retainDays: retainDays, loc: loc, now: time.Now, days: map[time.Time]dayCacheEntry{}}, nil
 }
 
 // dayOf is the package dayOf resolved in the STORE's zone. The one place the day an
@@ -476,8 +502,65 @@ func appendBytes(f io.Writer, b []byte) (int, error) {
 // write path, not a larger buffer here.
 const maxLineBytes = 1 << 20
 
-// readDay decodes one day file. A missing file is not an error: an idle day writes
-// none, which is the normal case on a laptop.
+// readDay answers one day's rows, from the cache when the day is past and the file is
+// unchanged (#1566). The cache exists because the read side polls: agentop re-queries
+// the same windows every minute to five minutes, and a past day file changes only when
+// a second writer appends to it, which moves size and mtime. TODAY is never cached —
+// the writer is still appending to its file — and neither is any error: a day that
+// could not be read is re-attempted on the next query rather than remembered as empty.
+//
+// Stat, not Open, validates an entry: the file's size and mtime are enough to know the
+// bytes are the ones already decoded, and a file an operator has made unreadable still
+// answers from the cache — which is the point, the decode was already paid for.
+func (s *store) readDay(day time.Time) ([]Row, dayIssues, error) {
+	if !day.Before(s.dayOf(s.now())) {
+		return s.readDayFile(day)
+	}
+	fi, err := os.Stat(s.path(day))
+	if os.IsNotExist(err) {
+		// Pruned or never written. Dropped rather than kept: an entry for a deleted
+		// file would serve rows retention has decided to forget.
+		s.mu.Lock()
+		delete(s.days, day)
+		s.mu.Unlock()
+		return nil, dayIssues{}, nil
+	}
+	if err != nil {
+		// Stat itself failed — permissions on the directory, a broken mount. Fall
+		// through to the read, which reports the failure through its own error path
+		// rather than guessing here.
+		return s.readDayFile(day)
+	}
+	s.mu.Lock()
+	ent, ok := s.days[day]
+	s.mu.Unlock()
+	if ok && ent.size == fi.Size() && ent.modTime.Equal(fi.ModTime()) {
+		return ent.rows, ent.issues, nil
+	}
+	rows, issues, rerr := s.readDayFile(day)
+	if rerr != nil {
+		return rows, issues, rerr
+	}
+	s.mu.Lock()
+	if len(s.days) >= dayCacheMax {
+		// Oldest day out, not newest-in-first-out: the newest past days are the ones
+		// the rolling windows keep asking for, and a cache that evicted them would
+		// re-decode exactly the files it exists to avoid.
+		var oldest time.Time
+		for d := range s.days {
+			if oldest.IsZero() || d.Before(oldest) {
+				oldest = d
+			}
+		}
+		delete(s.days, oldest)
+	}
+	s.days[day] = dayCacheEntry{rows: rows, issues: issues, size: fi.Size(), modTime: fi.ModTime()}
+	s.mu.Unlock()
+	return rows, issues, nil
+}
+
+// readDayFile decodes one day file, uncached — the whole file, line by line. A missing
+// file is not an error: an idle day writes none, which is the normal case on a laptop.
 //
 // SKIPS an undecodable line and keeps going, rather than stopping at it. A scanner rather
 // than one json.Decoder over the whole file, because a Decoder cannot resync: it tolerates a
@@ -490,7 +573,7 @@ const maxLineBytes = 1 << 20
 // level, and not into a local, which reaches no caller, no exported counter and no API
 // response. The count is what lets an operator tell "my ledger is fine" from "my ledger is
 // losing lines", so it has to leave this function. See dayIssues and Caveats.SkippedLines.
-func (s *store) readDay(day time.Time) ([]Row, dayIssues, error) {
+func (s *store) readDayFile(day time.Time) ([]Row, dayIssues, error) {
 	f, err := os.Open(s.path(day))
 	if os.IsNotExist(err) {
 		return nil, dayIssues{}, nil