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) + } + } +}