From 9ba1fcd50b329053c38280d55bf3ccc05a618592 Mon Sep 17 00:00:00 2001 From: Hai Huang Date: Fri, 30 Oct 2026 09:12:54 -0400 Subject: [PATCH] fix: make the ledger agree with the ring, rotate archive segments, and cache past day files MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Five defects from the cost/session read-path audit (#1566): 1. Coverage parity: the ledger's priceable rule required a non-zero token count, so a truncated OpenAI-dialect stream (200, a model, no usage block) hid a coverage gap on today/7d/month that the ring's duration windows reported. usage.Priceable is the ring's rule stated once; the ledger writer now calls it. 2. Error parity: the ledger counted any 400+/denial as an error, predating the ring's destination_private and upgrade_declined exemptions, so a guard refusal read as a failing proxy on today but not on 1h. The rule is now usage.CountsAsError, called from both. 3. Archive segments never rotated on size: a session busy past the idle window kept one segment growing without limit, and planPrune skips open segments in both its age and maxBytes loops, so the archive's size bound was unenforceable against exactly the busiest sessions. Segments now rotate at 128 MiB, checked in the tick after the flush (a segment's byte count is only real once its zstd encoder has been flushed). 4. Every ledger query re-decoded every day file in the window: agentop polls today each minute and 7d/month each five, so a month span decoded 31 files per poll and past days almost never change. Past days are now cached keyed on (size, mtime); today is never cached. 5. Four comments in core/cost described code that has since changed (the accumulation "open defect" that Counts.Add's saturating addInto closed, "NO IMPORTER ON THIS BRANCH", "neither exists at this commit", and Dropped()'s "no non-test caller") — corrected so a reader stops concluding that sums can wrap and token bounds are unenforced. Assisted-By: Claude (Anthropic AI) Signed-off-by: Hai Huang --- core/cost/ledger/daycache_test.go | 145 ++++++++++++++++++++++++++ core/cost/ledger/parity_test.go | 76 ++++++++++++++ core/cost/ledger/store.go | 91 +++++++++++++++- core/cost/ledger/writer.go | 29 ++++-- core/cost/pricing/cost.go | 28 +++-- core/cost/usage/usage.go | 37 ++++++- core/session/archive/archive.go | 41 +++++++- core/session/archive/rotation_test.go | 82 +++++++++++++++ 8 files changed, 494 insertions(+), 35 deletions(-) create mode 100644 core/cost/ledger/daycache_test.go create mode 100644 core/cost/ledger/parity_test.go create mode 100644 core/session/archive/rotation_test.go diff --git a/core/cost/ledger/daycache_test.go b/core/cost/ledger/daycache_test.go new file mode 100644 index 000000000..1b3c538d8 --- /dev/null +++ b/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)) + } +} diff --git a/core/cost/ledger/parity_test.go b/core/cost/ledger/parity_test.go new file mode 100644 index 000000000..d4ad746e6 --- /dev/null +++ b/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) + } +} diff --git a/core/cost/ledger/store.go b/core/cost/ledger/store.go index aae58ca58..076ac1198 100644 --- a/core/cost/ledger/store.go +++ b/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 diff --git a/core/cost/ledger/writer.go b/core/cost/ledger/writer.go index 849a779c6..e20b1958c 100644 --- a/core/cost/ledger/writer.go +++ b/core/cost/ledger/writer.go @@ -236,6 +236,10 @@ func New(dir string, opts ...Option) (*Writer, error) { return nil, err } w.store = s + // The cache's "is this day still being written" must answer at the writer's + // instant, not the machine's: a pinned test clock filing rows into a past day + // would otherwise have that day cached as immutable while the writer appends. + s.now = w.now if perr := s.prune(now); perr != nil { // Not fatal, and not returned. A ledger that cannot delete an old file is // still a ledger that can record today's spend, and refusing to start the @@ -511,12 +515,21 @@ func (w *Writer) Record(_ string, e *pipeline.SessionEvent) { if c := pricing.UnitOf(w.rates, e.Host, model); c != "" { r.Currency = rowLabel(c) } - if e.StatusCode >= 400 || e.Phase == pipeline.SessionDenied { + // BOTH RULES ARE THE RING'S, called rather than restated (#1566). This writer and + // the aggregator answer the same traffic on different surfaces — today/7d/month + // here, the duration windows there — and a restated rule drifts: the ledger's copy + // of the error rule predated the destination_private exemption, so a guard refusal + // read as an error on today but not on 1h, and its copy of the priceable rule + // required a non-zero token count, so a truncated stream hid a coverage gap here + // that the ring reported there. + if usage.CountsAsError(e) { r.Errors = 1 } - // Priceable: carried a model and a non-zero token count. The denominator for - // coverage — Requests is the wrong one, since only inference can be priced. - if r.Model != "" && r.Tokens > 0 { + // Priceable: the parser recognised a model, so a rate could have been named — the + // denominator for coverage, since Requests is the wrong one (only inference can be + // priced). usage.Priceable, not `r.Model != "" && r.Tokens > 0`: a successful + // response with a model and no usage is a coverage gap and must be counted as one. + if usage.Priceable(e) { r.PriceableRequests = 1 } // The figure core/cost/settle settled. Record, not Decode: an unpriced record @@ -926,11 +939,9 @@ func (w *Writer) logDrops() { // writer, not about anybody's read: it happened once, no later read can rediscover it, and // every reader of this ledger is equally short because of it. // -// No non-test caller today, and it stays exported anyway: it is the completeness signal the -// package doc and Window's contract both cite, and the /v1/usage response has a Degraded -// field already carrying the two READ-side counts beside it (see Caveats). This is the -// write-side one that belongs there next, which is a sessionapi change rather than a reason -// to withdraw the number. +// The non-test caller is sessionapi's /v1/usage response (the Degraded field beside +// the two READ-side counts from Caveats): the completeness signal the package doc and +// Window's contract both cite, surfaced where an operator reads it. func (w *Writer) Dropped() int64 { return w.dropped.Load() } // run is the ONLY goroutine that touches the filesystem after construction. diff --git a/core/cost/pricing/cost.go b/core/cost/pricing/cost.go index 437d2eef3..b2f2505ec 100644 --- a/core/cost/pricing/cost.go +++ b/core/cost/pricing/cost.go @@ -17,12 +17,12 @@ import "math" // ledger row was expected. A second bound declared over there would be free to drift from // this one; there is only ever one answer to "past which figure is this a bug". // -// WHAT IT DOES NOT BOUND: THE ACCUMULATED SUM. 1,024 requests at this bound wrap -// usage.Counts.Add to a large negative total, and NO per-request bound can close that — for any -// bound C the sum wraps after MaxInt64/C requests, so a smaller C buys distance, not closure. -// Closing it takes a checked accumulate where the sum is kept, which the aggregate PR in this -// series does. TestNoPerRequestBoundClosesTheAccumulationWrap pins the reasoning; until that PR -// lands this is an open defect and not history, reachable through a header on a parsed path. +// WHAT IT DOES NOT BOUND: THE ACCUMULATED SUM — no per-request bound can, since for any +// bound C the sum grows past MaxInt64 after MaxInt64/C requests. The sum is closed where +// it is kept: usage.Counts.Add accumulates through addInto, which CLAMPS to the int64 +// range instead of wrapping, so 1,024 requests at this bound saturate the total rather +// than turn it negative. TestNoPerRequestBoundClosesTheAccumulationWrap pins the +// reasoning; the saturating accumulate is the closure it asks for. // // EXCLUSIVE: exactly MaxCostMicros is out of range. 2^53 is the first integer whose successor // float64 cannot represent, so 2^53+1 rounds back onto it — at that one value an out-of-range @@ -90,11 +90,10 @@ const MaxPlausibleRequestCostMicros int64 = maxPlausibleTokens * maxPlausibleMic // a forged response that cannot move the dollar total past $10,000 could still move the // token total by 9.2e18, and a client renders the two side by side. // -// NO IMPORTER ON THIS BRANCH: the consumers are usage.maxPlausibleRequestTokens (#1013 of this -// stack) and the ledger writer's admission guard (#1014), so a grep at this commit finds only the -// declaration. Exported anyway because the split put the derivation here and its readers in the -// next two PRs — and because two literals that must agree is the shape that let config's -// retention floor drift from the window it protects. +// The importers are usage.maxPlausibleRequestTokens and the ledger writer's admission +// guard — the derivation lives here so both compare against the same literal, because +// two literals that must agree is the shape that let config's retention floor drift +// from the window it protects. const MaxPlausibleTokens = maxPlausibleTokens // BundleableRate reports whether a vendor list rate is one the BUNDLED TABLE may carry: @@ -199,10 +198,9 @@ func MicrosFromUSD(usd float64) (int64, bool) { // - A count above maxPlausibleTokens, on the same reasoning and from the same // wire. THE MODELLED FIGURE WAS THE UNBOUNDED ONE: headerCost refuses a // gateway's figure past MaxPlausibleRequestCostMicros ($10,000), and -// the aggregate PR in this series WILL refuse an implausible token report and -// count it (usage.plausibleTokenReport, Counts.RefusedTokenRequests — neither -// exists at this commit) — while the money derived from those same counts -// arrived here and was believed, bounded only by MaxCostMicros, which is +// usage.plausibleTokenReport refuses an implausible token report and counts it +// in Counts.RefusedTokenRequests — while the money derived from those same +// counts arrived here and was believed, bounded only by MaxCostMicros, which is // $9 billion. So one response could have its tokens called impossible and its // dollars kept, in the same aggregate, side by side in the same client. // - A modelled figure past MaxPlausibleRequestCostMicros ($10,000), which is the diff --git a/core/cost/usage/usage.go b/core/cost/usage/usage.go index 5ec5c54c4..2565336f4 100644 --- a/core/cost/usage/usage.go +++ b/core/cost/usage/usage.go @@ -813,7 +813,10 @@ func (a *Aggregator) costOf(e *pipeline.SessionEvent, ce event.Event, haveRec bo // denial after the parser ran, reports a model with zero usage. Naming it // would advertise a missing rate for a model that may well have one, and // adding that entry would never make the row disappear. - if e.StatusCode >= 400 || e.Phase == pipeline.SessionDenied { + // + // Priceable is this rule stated once, so the ledger's today/7d/month windows + // answer with the same denominator the duration windows do (#1566). + if !Priceable(e) { return eventCost{} } // A SUCCEEDED one is a coverage gap and must be counted as one. Returning a @@ -1280,7 +1283,12 @@ func (a *Aggregator) evictColdestLocked() { // configured it to, on traffic that was never going to be allowed through — a // nonzero "error rate" made of refusals is the policy working, not the proxy // failing. The Finisher sees it as an allow for the same reason. -func countsAsError(e *pipeline.SessionEvent) bool { +// +// Exported because the cost LEDGER counts the same traffic for its today/7d/month +// windows and must apply the same rule (#1566): a refusal the ring excludes read as +// an error on today but not on the 1h window, and two surfaces an operator reads +// side by side disagreed about whether the proxy was failing. +func CountsAsError(e *pipeline.SessionEvent) bool { if e.Error != nil && (e.Error.Kind == pipeline.ErrorKindUpgradeDeclined || e.Error.Kind == pipeline.ErrorKindDestinationPrivate) { return false @@ -1288,6 +1296,29 @@ func countsAsError(e *pipeline.SessionEvent) bool { return e.Error != nil || e.StatusCode >= 400 || e.Phase == pipeline.SessionDenied } +// Priceable reports whether e is inference traffic the coverage ratio's denominator +// counts: the parser recognised a model on it, so a rate could have been named for it. +// This is the ring's rule, exported so the ledger answers today/7d/month with the same +// denominator the ring gives the duration windows (#1566): the ledger used to require +// a non-zero token count as well, so a truncated OpenAI-dialect stream — 200, a model, +// and no usage block, because every counter lives on the final chunk — was priceable +// on 1h and invisible on today, and nine priced requests plus one of those read +// "9/9 priced" over a gap the other surface reported as 9/10. +// +// A FAILED response with a model and no usage is not priceable: a 5xx or a denial +// after the parser ran is not a pricing gap, and naming it would advertise a missing +// rate for a model that may well have one. See eventCost for the full reasoning; this +// function is that rule stated once. +func Priceable(e *pipeline.SessionEvent) bool { + if e.Inference == nil || e.Inference.Model == "" { + return false + } + if pricing.UsageFromInference(e.Inference) != (pricing.Usage{}) { + return true + } + return e.StatusCode < 400 && e.Phase != pipeline.SessionDenied +} + // foldInto accumulates one event into one ring. // // sessionID is passed rather than derived because the ring itself does not know @@ -1368,7 +1399,7 @@ func (a *Aggregator) foldInto(ring []bucket, t time.Time, sessionID string, e *p if ec.incompleteReason != "" { addLabel(&b.byIncomplete, ringLabel(ec.incompleteReason), Counts{Requests: 1}) } - if countsAsError(e) { + if CountsAsError(e) { one.Errors = 1 } b.Counts.Add(one) diff --git a/core/session/archive/archive.go b/core/session/archive/archive.go index 97cb0ab92..28a8c97a1 100644 --- a/core/session/archive/archive.go +++ b/core/session/archive/archive.go @@ -46,6 +46,15 @@ const ( // wrap time arithmetic must not turn into a cutoff in the future. MaxRetentionDays = 3650 defaultMaxBytes = 2 << 30 + // defaultMaxSegBytes rotates one session's segment once it holds this many bytes + // (#1566). Segments used to close only on idle, the open-writer cap or + // delete/clear/rename — never on size — and planPrune skips open segments in both + // its age and maxBytes loops, so a session busy past the idle window kept one + // segment growing without limit and maxBytes was a bound the busiest session was + // exempt from. A cap of maxBytes/defaultMaxOpen bounds a session's disk footprint + // about the same way maxOpen bounds the heap. Rotation runs on the tick after the + // cap is crossed, because a segment's byte count is only real once flushed. + defaultMaxSegBytes = defaultMaxBytes / 16 ) // Option configures an Archive. @@ -61,6 +70,11 @@ func WithRetentionDays(days int) Option { return func(a *Archive) { a.retainDays // WithMaxBytes sets the archive's size bound. Zero or negative is the default, 2 GiB. func WithMaxBytes(n int64) Option { return func(a *Archive) { a.maxBytes = n } } +// WithMaxSegmentBytes rotates a session's segment once it holds this many bytes, so +// retention can prune a busy session's history instead of watching one open segment +// grow past the archive's size bound. Zero or negative is the default, 128 MiB. +func WithMaxSegmentBytes(n int64) Option { return func(a *Archive) { a.maxSegBytes = n } } + // WithQueueDepth sets the request path's queue depth, for tests. func WithQueueDepth(n int) Option { return func(a *Archive) { a.depth = n } } @@ -98,10 +112,13 @@ type Archive struct { now func() time.Time retainDays int maxBytes int64 - depth int - idleClose time.Duration - maxOpen int - tickEvery time.Duration + // maxSegBytes rotates a session's segment once it holds this many bytes. Writer + // goroutine only, like everything that touches segment writers. + maxSegBytes int64 + depth int + idleClose time.Duration + maxOpen int + tickEvery time.Duration // Test seams: gate holds the writer before each op; segHook sees each new segment writer. gate chan struct{} @@ -222,6 +239,9 @@ func Open(root string, opts ...Option) (*Archive, error) { if a.maxBytes <= 0 { a.maxBytes = defaultMaxBytes } + if a.maxSegBytes <= 0 { + a.maxSegBytes = defaultMaxSegBytes + } if a.depth <= reserve { a.depth = reserve + 1 } @@ -732,6 +752,19 @@ func (a *Archive) tick() { a.writeFailed(s, err) continue } + // Rotation at the cap (#1566). planPrune skips open segments in both its age and + // maxBytes loops, so a session that stays busy past the idle window would otherwise + // keep one segment growing without limit, and maxBytes would be a bound the busiest + // session was exempt from. Closing here moves the segment into the closed list, + // where retention can delete it; the session's next write opens a fresh one. + // + // AFTER the flush, because Bytes counts what the zstd encoder has handed the file: + // an unflushed writer reports zero no matter how much is buffered in it, so a + // write-time check could not see the cap cross. The cost is one tick of overshoot. + if s.w.info().Bytes >= a.maxSegBytes { + a.closeWriter(s) + continue + } if now.Sub(s.lastWrite) >= a.idleClose { a.closeWriter(s) } diff --git a/core/session/archive/rotation_test.go b/core/session/archive/rotation_test.go new file mode 100644 index 000000000..a9bebb72e --- /dev/null +++ b/core/session/archive/rotation_test.go @@ -0,0 +1,82 @@ +package archive + +import ( + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/rossoctl/cortex/core/pipeline" +) + +// TestWrite_RotatesASegmentAtTheByteCap pins issue #1566 item 3. A segment used to close +// only on 60-minute idle, the open-writer cap, or delete/clear/rename/write-failure — +// never on size — and planPrune skips open segments in both its age and maxBytes loops. +// A session busy all day therefore kept one segment growing without limit, and the +// archive's maxBytes was unenforceable against exactly the sessions that produced the +// most bytes. Rotation closes the segment at the cap; the next write opens a fresh one, +// which retention can then actually delete. +func TestWrite_RotatesASegmentAtTheByteCap(t *testing.T) { + root := t.TempDir() + clk := newClock() + // A one-byte cap makes every event its own rotation: the test wants the mechanism, + // not a particular segment size. + a := openTest(t, root, clk, WithMaxSegmentBytes(1)) + rec := newRecorder(a) + + var evs []pipeline.SessionEvent + for i := 0; i < 8; i++ { + evs = append(evs, pipeline.SessionEvent{ + Direction: pipeline.Outbound, Phase: pipeline.SessionRequest, Host: "api.example.com", + Inference: &pipeline.InferenceExtension{Model: "m", + Messages: []pipeline.InferenceMessage{{Role: "user", Content: "hello"}}}, + }) + } + rec.record("s1", evs[:5]...) + // Rotation runs in the tick after the cap is crossed, not in write: a segment's + // byte count is only real once its zstd encoder has been flushed. The first tick + // closes the over-cap segment; the writes after it open the next one, which is + // what puts a second segment on disk. + settle(a) + tick(a) + rec.record("s1", evs[5:]...) + settle(a) + // The clock has to move for session.json to be rewritten: writeMeta runs on the + // metaEvery cadence while dirty, and a frozen clock never crosses it. + clk.advance(metaEvery + time.Second) + tick(a) + + dir := filepath.Join(root, dataDirName, dirName("s1")) + entries, err := os.ReadDir(dir) + if err != nil { + t.Fatal(err) + } + var segs int + for _, e := range entries { + // HasSuffix, not filepath.Ext: the extension is ".jsonl.zst", and Ext + // stops at the last dot, returning ".zst". + if strings.HasSuffix(e.Name(), segmentExt) { + segs++ + } + } + if segs < 2 { + t.Fatalf("session holds %d segment(s), want >= 2: a session written past the byte "+ + "cap must rotate, or its segment stays open forever, planPrune can never delete "+ + "it, and maxBytes is a bound the busiest session is exempt from", segs) + } + + // Rotation must not lose or reorder anything: every event stays readable through the + // same reader the request path uses, in sequence order. + got := onDisk(t, root, "s1") + if len(got) != len(evs) { + t.Fatalf("onDisk holds %d events, want %d — rotation must be a file-boundary change, "+ + "not a data change", len(got), len(evs)) + } + for i := range got { + if got[i].Seq != uint64(i+1) { + t.Fatalf("event %d has Seq %d, want %d: the reader must see the same sequence "+ + "across the rotated boundary", i, got[i].Seq, i+1) + } + } +}