diff --git a/internal/usage/cache_bounds_test.go b/internal/usage/cache_bounds_test.go index e07c8122..167a4992 100644 --- a/internal/usage/cache_bounds_test.go +++ b/internal/usage/cache_bounds_test.go @@ -125,9 +125,12 @@ func holdLogTick(t *testing.T, tick time.Time) { t.Cleanup(func() { logChangeTime, logIndexNow = changeTime, now }) } +// The parse is the fingerprint (#1357), so a rewrite can land before the read +// (after the stat that decided to read) or between the read and the +// after-read fingerprint check. Neither may leave the old bytes cached. func TestUsageRewriteBeforeFingerprintIsNotCached(t *testing.T) { - for _, rewriteAt := range []int{1, 2} { - t.Run(fmt.Sprint("hash=", rewriteAt), func(t *testing.T) { + for _, rewriteAt := range []string{"before-read", "after-read"} { + t.Run(rewriteAt, func(t *testing.T) { pageHome(t) now := holdClock(t, time.Now()) tick := time.Now().Truncate(time.Second) @@ -146,23 +149,34 @@ func TestUsageRewriteBeforeFingerprintIsNotCached(t *testing.T) { if bytes.Equal(edited, data) || len(edited) != len(data) { t.Fatal("fixture must change at the same size") } - hash := logRecordHash - hashCalls := 0 rewritten := false - logRecordHash = func(path string, n int64) string { - hashCalls++ - if hashCalls == rewriteAt { - rewritten = true - if err := os.WriteFile(path, edited, 0600); err != nil { - t.Fatal(err) - } - if err := os.Chtimes(path, info.ModTime(), info.ModTime()); err != nil { - t.Fatal(err) - } + rewrite := func(path string) { + if rewritten { + return + } + rewritten = true + if err := os.WriteFile(path, edited, 0600); err != nil { + t.Fatal(err) + } + if err := os.Chtimes(path, info.ModTime(), info.ModTime()); err != nil { + t.Fatal(err) } - return hash(path, n) } - t.Cleanup(func() { logRecordHash = hash }) + hash, changeTime := logRecordHash, logChangeTime + if rewriteAt == "before-read" { + // The build's first look at the change time follows its stat + // and precedes the open that reads and hashes the lines. + logChangeTime = func(i os.FileInfo) time.Time { + rewrite(Path()) + return changeTime(i) + } + } else { + logRecordHash = func(path string, n int64) string { + rewrite(path) + return hash(path, n) + } + } + t.Cleanup(func() { logRecordHash, logChangeTime = hash, changeTime }) Summarize(All) if !rewritten { t.Fatal("fixture did not rewrite at the read boundary") diff --git a/internal/usage/log_hash_cost_test.go b/internal/usage/log_hash_cost_test.go new file mode 100644 index 00000000..a85e4db5 --- /dev/null +++ b/internal/usage/log_hash_cost_test.go @@ -0,0 +1,100 @@ +package usage + +import ( + "encoding/json" + "fmt" + "os" + "testing" + "time" +) + +// While the gateway writes, the log's change time is nearly always under a +// second old, so each read verifies the snapshot by a fingerprint (#1357). +// The parse itself produces that fingerprint: a rebuild re-reads the file only +// to verify the prefix it continues and to check nothing was rewritten during +// the read, not a third time before the read. Hashing doesn't allocate, so +// the allocation tests don't see this; this counts the bytes re-read. +func TestUsageAppendReadHashesNoPreReadScan(t *testing.T) { + pageHome(t) + holdClock(t, time.Now()) + historyLog(t, 20000) + now := logIndexNow + logIndexNow = func() time.Time { return time.Now().Add(2 * logStampSettle) } + Summarize(Today) // settled first, as a log the gateway hasn't touched for a while + logIndexNow = now + t.Cleanup(func() { logIndexNow = now }) + hash := logRecordHash + var hashed int64 + logRecordHash = func(path string, n int64) string { + hashed += n + return hash(path, n) + } + t.Cleanup(func() { logRecordHash = hash }) + const cycles = 5 + start := time.Now() + for range cycles { + Append(Record{Agent: "codex", Provider: "relay", Model: "m", Input: 1, Status: 200}) + Summarize(Today) + } + took := time.Since(start) + info, err := os.Stat(Path()) + if err != nil { + t.Fatal(err) + } + perCycle := hashed / cycles + t.Logf("log %d bytes: %d bytes hashed per append + read (%.2f× the log), %v per cycle", info.Size(), perCycle, float64(perCycle)/float64(info.Size()), took/cycles) + if s := Summarize(Today); s.Calls != 1+cycles { + t.Fatalf("calls today: %d", s.Calls) + } + // An unsettled append verifies its prefix and re-hashes after the read: + // two passes. A third, before the read, is the scan the parse replaces. + if perCycle > 2*info.Size()+info.Size()/10 { + t.Fatalf("an append + read hashed %d bytes over a %d-byte log, more than the prefix check and the after-read check", perCycle, info.Size()) + } +} + +// Another writer can be mid-line when a read stats the log. The fingerprint +// covers only the complete lines parsed, so the line's completion continues +// the snapshot rather than counting the half line as read. +func TestUsagePartialLineThenCompletion(t *testing.T) { + for _, settle := range []bool{false, true} { + t.Run(fmt.Sprint("settled=", settle), func(t *testing.T) { + pageHome(t) + now := holdClock(t, time.Now()) + if settle { + settleLogClock(t) + } + historyLog(t, logBlockRows+3) + line, err := json.Marshal(Record{Time: now, Agent: "codex", Provider: "relay", Model: "m", Session: "late", Input: 41, Status: 200}) + if err != nil { + t.Fatal(err) + } + line = append(line, '\n') + half := len(line) / 2 + write := func(b []byte) { + t.Helper() + f, err := os.OpenFile(Path(), os.O_APPEND|os.O_WRONLY, 0600) + if err != nil { + t.Fatal(err) + } + if _, err := f.Write(b); err != nil { + t.Fatal(err) + } + if err := f.Close(); err != nil { + t.Fatal(err) + } + } + write(line[:half]) + before := Summarize(All) + if s := readLogSnapshot(); s.hash != "" && s.hash != recordHash(Path(), s.off) { + t.Fatal("fingerprint does not cover the parsed bytes") + } + write(line[half:]) + got, want := Summarize(All), summarize(All, now, Load(time.Time{})) + if got.Totals != want.Totals || got.Calls != before.Calls+1 { + t.Fatalf("completed line: got %+v (%d calls, %d before) want %+v", got.Totals, got.Calls, before.Calls, want.Totals) + } + equalPage(t, QueryPage(All, Filter{}, 0, 100), pageFromLedger(All, Filter{}, 0, 100, LedgerOf(All, Filter{}))) + }) + } +} diff --git a/internal/usage/log_index.go b/internal/usage/log_index.go index 61555a29..9e051814 100644 --- a/internal/usage/log_index.go +++ b/internal/usage/log_index.go @@ -5,7 +5,11 @@ package usage // Readers decode only blocks which can overlap their time window. import ( "bufio" + "crypto/sha256" + "encoding" "encoding/json" + "fmt" + "hash" "io" "maps" "os" @@ -29,7 +33,8 @@ type logSnapshot struct { path string info os.FileInfo off int64 - hash string + hash string // SHA-256 of the log's first off bytes, the ones parsed + digest []byte // the hash's running state, which the next parse continues version uint64 blocks []*rowChunk first time.Time @@ -67,7 +72,7 @@ func logSnapshotFor(metadataOnly bool) *logSnapshot { unchanged := old != nil && old.path == path && (err != nil && old.info == nil || err == nil && sameLogInfo(old.info, info)) if unchanged && info != nil { if logChangeStamp(info) == "" || !old.settled { - unchanged = old.hash != "" && logRecordHash(path, info.Size()) == old.hash + unchanged = old.hash != "" && logRecordHash(path, old.off) == old.hash if unchanged && settled { promoted := *old promoted.info, promoted.settled = info, true @@ -94,8 +99,11 @@ func logSnapshotFor(metadataOnly bool) *logSnapshot { notePath, noteBase, noteLast := logAppends.path, logAppends.base, logAppends.last logAppends.Unlock() trusted := old != nil && old.settled && logChangeStamp(info) != "" && notePath == path && sameLogInfo(oldInfo(old), noteBase) && sameLogInfo(info, noteLast) - continued := old != nil && old.path == path && old.info != nil && sameLogFile(old.info, info) && info.Size() > old.info.Size() && (trusted || old.hash != "" && logRecordHash(path, old.info.Size()) == old.hash) - if continued && !old.uncached { + continued := old != nil && old.path == path && old.info != nil && sameLogFile(old.info, info) && info.Size() > old.info.Size() && (trusted || old.hash != "" && logRecordHash(path, old.off) == old.hash) + // The parse is the fingerprint: it hashes the lines it reads, continuing + // the old snapshot's hash when it continues its blocks. + sum := sha256.New() + if continued && !old.uncached && resumeLogHash(sum, old.digest) { next.off, next.first = old.off, old.first next.blocks = slices.Clone(old.blocks) next.keyProviders = maps.Clone(old.keyProviders) @@ -108,9 +116,6 @@ func logSnapshotFor(metadataOnly bool) *logSnapshot { if _, err = f.Seek(next.off, 0); err != nil { return next } - if !trusted || !settled { - next.hash = logRecordHash(path, info.Size()) - } var tail *rowChunk if n := len(next.blocks); n > 0 && next.blocks[n-1].Count < logBlockRows { tail = next.blocks[n-1].writable() @@ -123,6 +128,7 @@ func logSnapshotFor(metadataOnly bool) *logSnapshot { break } // a partial final line is read again after it completes next.off += int64(len(b)) + sum.Write(b) var r Record if json.Unmarshal(b, &r) != nil { continue @@ -145,9 +151,10 @@ func logSnapshotFor(metadataOnly bool) *logSnapshot { if tail != nil { next.blocks = append(next.blocks, tail.freeze()) } + next.hash, next.digest = fmt.Sprintf("%x", sum.Sum(nil)), marshalLogHash(sum) // A rewrite during the read must not give old blocks the new content's hash. - if (!trusted || !settled) && next.hash != logRecordHash(path, info.Size()) { - next.hash = "" + if (!trusted || !settled) && next.hash != logRecordHash(path, next.off) { + next.hash, next.digest = "", nil next.settled = false } logAppends.Lock() @@ -212,6 +219,27 @@ func (c *rowChunk) freeze() *rowChunk { return c.pack() } +func resumeLogHash(h hash.Hash, state []byte) bool { + u, ok := h.(encoding.BinaryUnmarshaler) + if !ok || state == nil || u.UnmarshalBinary(state) != nil { + h.Reset() + return false + } + return true +} + +func marshalLogHash(h hash.Hash) []byte { + m, ok := h.(encoding.BinaryMarshaler) + if !ok { + return nil + } + state, err := m.MarshalBinary() + if err != nil { + return nil + } + return state +} + func oldInfo(s *logSnapshot) os.FileInfo { if s == nil { return nil