usage: the parse is the log fingerprint, dropping the build's full re-read before it (#1357)

What changed. A usage-log snapshot's fingerprint is now produced by the
parse itself: each complete line read is fed into a SHA-256. When a build
continues the old snapshot's blocks, it continues the old snapshot's digest
too (crypto/sha256's state, kept on the snapshot as `digest`); otherwise the
digest starts fresh from byte 0 with the parse. The build's before-read
logRecordHash over the whole file is gone. The fingerprint covers the bytes
actually parsed (`off`), not info.Size(), so a half line another writer is
in the middle of appending is no longer counted as read; the unchanged
check, the prefix check and the after-read check all hash `off` bytes.

What stays, as #1328's tests require (ggbdpq's analysis on #1357): the
unsettled unchanged check, the prefix re-verification on a non-trusted
continuation, and the after-read check, which now compares the parse's own
digest against the file as it is after the read. A same-tick rewrite is
still caught at the build, not at a later read. On the trusted, settled
path nothing is re-read, as before; that snapshot now also carries a
fingerprint (the continued digest), so a later non-trusted append can
verify its prefix instead of rebuilding from byte 0.

Test changed: TestUsageRewriteBeforeFingerprintIsNotCached counted
logRecordHash calls and rewrote the file at call 1 (after the before-read
hash, before the parse) or call 2 (after the read). With the parse being
the fingerprint there is no call 1 any more, so the hash=2 case could no
longer fire. Its guarantee still holds and is still tested at both
boundaries: "before-read" rewrites at the build's first change-time look,
after its stat and before it opens and parses; "after-read" rewrites at the
after-read fingerprint check. Removing the after-read check makes
after-read fail. No other test changed.

New tests:
- TestUsageAppendReadHashesNoPreReadScan: 20,000-row settled log
  (2,760,615 bytes), 5 x Append + Summarize(Today), counts the bytes
  logRecordHash re-reads. origin/main: 7,728,984 bytes per cycle (2.80x the
  log), FAIL; this commit: 4,968,615 (1.80x), PASS. Wall time per cycle on
  an M-series Mac, 3 runs each: 3.43-3.52 ms before, 2.28-2.52 ms after
  (~32% less). The remaining two passes are the prefix check and the
  after-read check, which the same-tick guarantee needs.
- TestUsagePartialLineThenCompletion (settled and unsettled): another
  writer's half line, a read, then the line completed; the fingerprint
  covers the parsed bytes only and the completed record is counted. On
  origin/main it fails on the fingerprint check (it covered the half line).

Break checks: no after-read check -> RewriteBeforeFingerprint/after-read
fails; no prefix verification on a continuation -> SameSizeAndMtimeRewrite
append=true, SameTickRewrite append=true and LogGrowingRewriteThenAppend
fail; lines not fed into the digest -> SameTickRewrite fails.

Run: go test -tags nogui ./internal/usage/... ./internal/gateway/... (ok,
temp HOME, pinned caches, removed after); go test -race ./internal/usage/...
(ok); -race -count=10 on the rewrite/append tests (ok); gofmt -l (empty);
go vet, GOOS=linux and GOOS=windows go vet -tags nogui ./internal/usage/
(ok). Not tried on Windows or Linux hardware; the change is
platform-independent hashing over the same read.
This commit is contained in:
yetone committed 2026-10-09 06:54:28 +08:00
1 parent baecbb1be0
commit 439cfc4ac4
3 files changed
+167 -25

No files matched your search

+30 -16
View File
@@ -125,9 +125,12 @@ func holdLogTick(t *testing.T, tick time.Time) {
t.Cleanup(func() { logChangeTime, logIndexNow = changeTime, now }) 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) { func TestUsageRewriteBeforeFingerprintIsNotCached(t *testing.T) {
for _, rewriteAt := range []int{1, 2} { for _, rewriteAt := range []string{"before-read", "after-read"} {
t.Run(fmt.Sprint("hash=", rewriteAt), func(t *testing.T) { t.Run(rewriteAt, func(t *testing.T) {
pageHome(t) pageHome(t)
now := holdClock(t, time.Now()) now := holdClock(t, time.Now())
tick := time.Now().Truncate(time.Second) tick := time.Now().Truncate(time.Second)
@@ -146,23 +149,34 @@ func TestUsageRewriteBeforeFingerprintIsNotCached(t *testing.T) {
if bytes.Equal(edited, data) || len(edited) != len(data) { if bytes.Equal(edited, data) || len(edited) != len(data) {
t.Fatal("fixture must change at the same size") t.Fatal("fixture must change at the same size")
} }
hash := logRecordHash
hashCalls := 0
rewritten := false rewritten := false
logRecordHash = func(path string, n int64) string { rewrite := func(path string) {
hashCalls++ if rewritten {
if hashCalls == rewriteAt { return
rewritten = true }
if err := os.WriteFile(path, edited, 0600); err != nil { rewritten = true
t.Fatal(err) 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) 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) Summarize(All)
if !rewritten { if !rewritten {
t.Fatal("fixture did not rewrite at the read boundary") t.Fatal("fixture did not rewrite at the read boundary")
+100
View File
@@ -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{})))
})
}
}
+37 -9
View File
@@ -5,7 +5,11 @@ package usage
// Readers decode only blocks which can overlap their time window. // Readers decode only blocks which can overlap their time window.
import ( import (
"bufio" "bufio"
"crypto/sha256"
"encoding"
"encoding/json" "encoding/json"
"fmt"
"hash"
"io" "io"
"maps" "maps"
"os" "os"
@@ -29,7 +33,8 @@ type logSnapshot struct {
path string path string
info os.FileInfo info os.FileInfo
off int64 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 version uint64
blocks []*rowChunk blocks []*rowChunk
first time.Time 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)) unchanged := old != nil && old.path == path && (err != nil && old.info == nil || err == nil && sameLogInfo(old.info, info))
if unchanged && info != nil { if unchanged && info != nil {
if logChangeStamp(info) == "" || !old.settled { 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 { if unchanged && settled {
promoted := *old promoted := *old
promoted.info, promoted.settled = info, true promoted.info, promoted.settled = info, true
@@ -94,8 +99,11 @@ func logSnapshotFor(metadataOnly bool) *logSnapshot {
notePath, noteBase, noteLast := logAppends.path, logAppends.base, logAppends.last notePath, noteBase, noteLast := logAppends.path, logAppends.base, logAppends.last
logAppends.Unlock() logAppends.Unlock()
trusted := old != nil && old.settled && logChangeStamp(info) != "" && notePath == path && sameLogInfo(oldInfo(old), noteBase) && sameLogInfo(info, noteLast) 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) 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)
if continued && !old.uncached { // 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.off, next.first = old.off, old.first
next.blocks = slices.Clone(old.blocks) next.blocks = slices.Clone(old.blocks)
next.keyProviders = maps.Clone(old.keyProviders) next.keyProviders = maps.Clone(old.keyProviders)
@@ -108,9 +116,6 @@ func logSnapshotFor(metadataOnly bool) *logSnapshot {
if _, err = f.Seek(next.off, 0); err != nil { if _, err = f.Seek(next.off, 0); err != nil {
return next return next
} }
if !trusted || !settled {
next.hash = logRecordHash(path, info.Size())
}
var tail *rowChunk var tail *rowChunk
if n := len(next.blocks); n > 0 && next.blocks[n-1].Count < logBlockRows { if n := len(next.blocks); n > 0 && next.blocks[n-1].Count < logBlockRows {
tail = next.blocks[n-1].writable() tail = next.blocks[n-1].writable()
@@ -123,6 +128,7 @@ func logSnapshotFor(metadataOnly bool) *logSnapshot {
break break
} // a partial final line is read again after it completes } // a partial final line is read again after it completes
next.off += int64(len(b)) next.off += int64(len(b))
sum.Write(b)
var r Record var r Record
if json.Unmarshal(b, &r) != nil { if json.Unmarshal(b, &r) != nil {
continue continue
@@ -145,9 +151,10 @@ func logSnapshotFor(metadataOnly bool) *logSnapshot {
if tail != nil { if tail != nil {
next.blocks = append(next.blocks, tail.freeze()) 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. // A rewrite during the read must not give old blocks the new content's hash.
if (!trusted || !settled) && next.hash != logRecordHash(path, info.Size()) { if (!trusted || !settled) && next.hash != logRecordHash(path, next.off) {
next.hash = "" next.hash, next.digest = "", nil
next.settled = false next.settled = false
} }
logAppends.Lock() logAppends.Lock()
@@ -212,6 +219,27 @@ func (c *rowChunk) freeze() *rowChunk {
return c.pack() 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 { func oldInfo(s *logSnapshot) os.FileInfo {
if s == nil { if s == nil {
return nil return nil