From 5f4970cd3944d20d9d0af2766ac4c4994da45314 Mon Sep 17 00:00:00 2001 From: "shreyas@blacksmith.sh" Date: Mon, 2 Nov 2026 15:01:18 +0000 Subject: [PATCH 1/5] cache/disk: census counts an eviction as never-read from its access state, so boot-scanned entries stop inflating evicted_never_read after a restart Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- cache/disk/census.go | 2 +- cache/disk/census_test.go | 8 ++++++++ 2 files changed, 9 insertions(+), 1 deletion(-) diff --git a/cache/disk/census.go b/cache/disk/census.go index 8ae274a..e23193e 100644 --- a/cache/disk/census.go +++ b/cache/disk/census.go @@ -425,7 +425,7 @@ func (r *censusRecorder) OnEvict(key string, value lruItem) { if now > value.addedAt { createdAge = now - value.addedAt } - neverRead := value.lastAccess == value.addedAt + neverRead := value.accessState == itemStateNew lastReadIdx := censusBucketIdx(lastReadAge) createdIdx := censusBucketIdx(createdAge) diff --git a/cache/disk/census_test.go b/cache/disk/census_test.go index 98a2e34..89b4f3d 100644 --- a/cache/disk/census_test.go +++ b/cache/disk/census_test.go @@ -409,6 +409,14 @@ func TestCensusBootScannedIsUnknownNotNeverRead(t *testing.T) { if rows[0].EvictedNeverReadBytesByClass[class] != 0 { t.Errorf("EvictedNeverReadBytesByClass = %d, want 0", rows[0].EvictedNeverReadBytesByClass[class]) } + // The scalar and its counters follow the same access state: a boot-scanned + // entry has addedAt == lastAccess, which must not read as never-read. + if rows[0].EvictedNeverReadBytes != 0 || rows[0].EvictedBytes != 50 { + t.Errorf("EvictedNeverReadBytes/EvictedBytes = %d/%d, want 0/50", rows[0].EvictedNeverReadBytes, rows[0].EvictedBytes) + } + if got := testutil.ToFloat64(rec.evictedNeverReadEntries.WithLabelValues("unknown")); got != 0 { + t.Errorf("evictedNeverReadEntries = %v, want 0", got) + } if rows[0].ResidentUnknownBytesByClass[class] != 0 || rows[0].ResidentBytesByClass[class] != 50 { t.Errorf("resident unknown/total = %d/%d, want 0/50", rows[0].ResidentUnknownBytesByClass[class], rows[0].ResidentBytesByClass[class]) From 7d9ab6ea8efdb01fdfa2d6f54d7bbb99f9d75722 Mon Sep 17 00:00:00 2001 From: "shreyas@blacksmith.sh" Date: Mon, 2 Nov 2026 15:08:11 +0000 Subject: [PATCH 2/5] cache/disk: boot-scanned evictions carry no last-read history either Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- cache/disk/census.go | 7 +++++-- cache/disk/census_test.go | 9 +++++++++ 2 files changed, 14 insertions(+), 2 deletions(-) diff --git a/cache/disk/census.go b/cache/disk/census.go index e23193e..61b20ab 100644 --- a/cache/disk/census.go +++ b/cache/disk/census.go @@ -436,13 +436,16 @@ func (r *censusRecorder) OnEvict(key string, value lruItem) { acc.residentEntries-- acc.evictedBytes += value.sizeOnDisk acc.evictedEntries++ - if neverRead { + switch value.accessState { + case itemStateNew: acc.evictedNeverReadByes += value.sizeOnDisk - } else { + case itemStateTouched: if lastReadAge < censusLiveThreshold { acc.evictedLiveBytes += value.sizeOnDisk } acc.evictedByLastReadAge[lastReadIdx] += value.sizeOnDisk + // itemStateBootScanned: lastAccess is file atime (frozen at creation on + // noatime mounts), not a read, so the entry has no last-read history. } acc.evictedByCreatedAge[createdIdx] += value.sizeOnDisk acc.removeResident(class, value.sizeOnDisk, value.accessState) diff --git a/cache/disk/census_test.go b/cache/disk/census_test.go index 89b4f3d..cd649f4 100644 --- a/cache/disk/census_test.go +++ b/cache/disk/census_test.go @@ -417,6 +417,15 @@ func TestCensusBootScannedIsUnknownNotNeverRead(t *testing.T) { if got := testutil.ToFloat64(rec.evictedNeverReadEntries.WithLabelValues("unknown")); got != 0 { t.Errorf("evictedNeverReadEntries = %v, want 0", got) } + // Nor as read: its lastAccess is file atime, not a hit. + if rows[0].EvictedLiveBytes != 0 { + t.Errorf("EvictedLiveBytes = %d, want 0", rows[0].EvictedLiveBytes) + } + for i, v := range rows[0].EvictedByLastReadAge { + if v != 0 { + t.Errorf("EvictedByLastReadAge[%d] = %d, want 0", i, v) + } + } if rows[0].ResidentUnknownBytesByClass[class] != 0 || rows[0].ResidentBytesByClass[class] != 50 { t.Errorf("resident unknown/total = %d/%d, want 0/50", rows[0].ResidentUnknownBytesByClass[class], rows[0].ResidentBytesByClass[class]) From 181fdc4295ca70dc8abb5463fd10383861c725eb Mon Sep 17 00:00:00 2001 From: "shreyas@blacksmith.sh" Date: Mon, 2 Nov 2026 15:10:40 +0000 Subject: [PATCH 3/5] cache/disk: evicted last_read counters follow the access state too Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- cache/disk/census.go | 16 +++++++--------- cache/disk/census_test.go | 5 +++++ 2 files changed, 12 insertions(+), 9 deletions(-) diff --git a/cache/disk/census.go b/cache/disk/census.go index 61b20ab..746605f 100644 --- a/cache/disk/census.go +++ b/cache/disk/census.go @@ -425,7 +425,6 @@ func (r *censusRecorder) OnEvict(key string, value lruItem) { if now > value.addedAt { createdAge = now - value.addedAt } - neverRead := value.accessState == itemStateNew lastReadIdx := censusBucketIdx(lastReadAge) createdIdx := censusBucketIdx(createdAge) @@ -455,16 +454,15 @@ func (r *censusRecorder) OnEvict(key string, value lruItem) { r.mu.Unlock() sz := float64(value.sizeOnDisk) - // Never-read entries have no last read: they are counted in the - // never-read counter and the created-age buckets only, so the - // last_read series stays a pure harm signal (bytes someone actually - // used, evicted anyway). This also gives thrash alerts built-in - // restart tolerance: boot-scanned entries look never-read until first - // touched, so a roll cannot fabricate still-live evictions. - if neverRead { + // Only touched entries have a last read, so the last_read series stays + // a pure harm signal (bytes someone actually used, evicted anyway) and + // a roll cannot fabricate still-live evictions out of boot-scanned + // entries; those count in the created-age buckets only. + switch value.accessState { + case itemStateNew: r.evictedNeverReadBytes.WithLabelValues(tool).Add(sz) r.evictedNeverReadEntries.WithLabelValues(tool).Inc() - } else { + case itemStateTouched: r.evictedBytesByAge.WithLabelValues("last_read", censusBucketLabels[lastReadIdx], tool).Add(sz) r.evictedEntriesByAge.WithLabelValues("last_read", censusBucketLabels[lastReadIdx], tool).Inc() } diff --git a/cache/disk/census_test.go b/cache/disk/census_test.go index cd649f4..5743507 100644 --- a/cache/disk/census_test.go +++ b/cache/disk/census_test.go @@ -426,6 +426,11 @@ func TestCensusBootScannedIsUnknownNotNeverRead(t *testing.T) { t.Errorf("EvictedByLastReadAge[%d] = %d, want 0", i, v) } } + for _, label := range censusBucketLabels { + if got := testutil.ToFloat64(rec.evictedEntriesByAge.WithLabelValues("last_read", label, "unknown")); got != 0 { + t.Errorf("evictedEntriesByAge last_read %s = %v, want 0", label, got) + } + } if rows[0].ResidentUnknownBytesByClass[class] != 0 || rows[0].ResidentBytesByClass[class] != 50 { t.Errorf("resident unknown/total = %d/%d, want 0/50", rows[0].ResidentUnknownBytesByClass[class], rows[0].ResidentBytesByClass[class]) From 8ed21f2782de950f07cb650edb55b66e8b214f6c Mon Sep 17 00:00:00 2001 From: "shreyas@blacksmith.sh" Date: Mon, 2 Nov 2026 15:33:05 +0000 Subject: [PATCH 4/5] cache/disk: persist the index's read history across restarts On shutdown, after the servers have stopped, write each known entry's access state and last read to /access-state.v1; the next boot scan consumes the file so pre-existing blobs come back as new/touched with their real recency instead of boot-scanned (unknown), and the file is deleted so a stale copy can never be replayed. Blobs created after the file was written, and anything without a record, stay on file timestamps. Steady state is untouched; the cost is one parallel walk of the index at shutdown (~0.3 s / 2M entries, 21 B per entry) and one map load at boot. Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- cache/disk/BUILD.bazel | 2 + cache/disk/access_state.go | 256 ++++++++++++++++++ cache/disk/access_state_test.go | 165 +++++++++++ cache/disk/disk.go | 3 + cache/disk/load.go | 21 +- main.go | 8 +- server/grpc_bytestream_write_contract_test.go | 2 + server/grpc_test.go | 2 + 8 files changed, 455 insertions(+), 4 deletions(-) create mode 100644 cache/disk/access_state.go create mode 100644 cache/disk/access_state_test.go diff --git a/cache/disk/BUILD.bazel b/cache/disk/BUILD.bazel index df77cb7..3cd59f2 100644 --- a/cache/disk/BUILD.bazel +++ b/cache/disk/BUILD.bazel @@ -3,6 +3,7 @@ load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test") go_library( name = "go_default_library", srcs = [ + "access_state.go", "census.go", "census_clickhouse.go", "disk.go", @@ -39,6 +40,7 @@ go_library( go_test( name = "go_default_test", srcs = [ + "access_state_test.go", "census_clickhouse_test.go", "census_test.go", "disk_test.go", diff --git a/cache/disk/access_state.go b/cache/disk/access_state.go new file mode 100644 index 0000000..dfd0627 --- /dev/null +++ b/cache/disk/access_state.go @@ -0,0 +1,256 @@ +package disk + +import ( + "bufio" + "encoding/binary" + "errors" + "fmt" + "hash/fnv" + "io" + "log" + "os" + "path" + "runtime" + "sync" + "time" +) + +// The access-state file carries each entry's read history across a restart. +// Without it the boot scan only has file timestamps, and on a noatime mount +// those cannot tell a blob that was read a minute ago from one nobody ever +// asked for, so every pre-existing entry re-enters the index as +// itemStateBootScanned (read history unknown) and the tenant census reports +// its bytes as unknown until they are touched again. +// +// The file is written once, after the servers have stopped and no index +// mutation can follow, and consumed (then deleted) once, by the next boot +// scan. It is advisory only: a missing, truncated or stale file leaves the +// scan on file timestamps for the entries it cannot vouch for. +// +// Layout (little endian): magic, uint32 written-at unix seconds, uint64 +// record count, then one record per entry: 16-byte FNV-1a hash of the +// lookup key, uint32 lastAccess, uint8 access state. Only entries whose +// state is known (itemStateNew or itemStateTouched) are written; a +// boot-scanned entry that was never touched would come back as boot-scanned +// anyway. +const ( + accessStateFileName = "access-state.v1" + accessStateTempName = accessStateFileName + ".tmp" + accessStateMagic = "BRAS\x01" + accessStateHdrLen = len(accessStateMagic) + 4 + 8 + accessStateRecLen = 16 + 4 + 1 +) + +type accessStateKey [16]byte + +type accessStateRecord struct { + lastAccess uint32 + state uint8 +} + +// accessStateIndex is the consumed file, held only for the duration of the +// boot scan. +type accessStateIndex struct { + writtenAt uint32 + records map[accessStateKey]accessStateRecord +} + +func accessStateKeyFor(lookupKey string) accessStateKey { + h := fnv.New128a() + _, _ = io.WriteString(h, lookupKey) + var k accessStateKey + h.Sum(k[:0]) + return k +} + +// lookup returns the persisted state for an entry, or false when the file +// has nothing trustworthy for it: no record, or the blob was created after +// the file was written (a blob's mtime is its creation time; the record +// cannot be about this file). +func (idx *accessStateIndex) lookup(lookupKey string, addedAt uint32) (accessStateRecord, bool) { + if idx == nil || addedAt > idx.writtenAt { + return accessStateRecord{}, false + } + rec, ok := idx.records[accessStateKeyFor(lookupKey)] + if !ok { + return accessStateRecord{}, false + } + if rec.lastAccess < addedAt { + rec.lastAccess = addedAt + } + return rec, true +} + +// SaveAccessState writes the index's read history to the cache dir so the +// next boot scan can restore it. Call it only once no request can mutate +// the index any more (after the servers have stopped): it walks the whole +// index under the cache mutex. Hashing the keys is the dominant cost, so +// the walk hands batches of entries to one encoder per CPU and a single +// writer appends their output; record order is irrelevant to the reader. +func (c *diskCache) SaveAccessState() error { + start := time.Now() + tmp := path.Join(c.dir, accessStateTempName) + final := path.Join(c.dir, accessStateFileName) + + f, err := os.OpenFile(tmp, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0644) + if err != nil { + return err + } + + const batchSize = 1 << 14 + batches := make(chan []*entry, runtime.NumCPU()*2) + encoded := make(chan []byte, runtime.NumCPU()*2) + + var encoders sync.WaitGroup + for i := 0; i < runtime.NumCPU(); i++ { + encoders.Add(1) + go func() { + defer encoders.Done() + for batch := range batches { + buf := make([]byte, 0, len(batch)*accessStateRecLen) + for _, kv := range batch { + k := accessStateKeyFor(kv.key) + buf = append(buf, k[:]...) + buf = binary.LittleEndian.AppendUint32(buf, kv.value.lastAccess) + buf = append(buf, kv.value.accessState) + } + encoded <- buf + } + }() + } + + var writeErr error + writerDone := make(chan struct{}) + go func() { + defer close(writerDone) + w := bufio.NewWriterSize(f, 4<<20) + // Header first; the record count is patched in once known. + hdr := make([]byte, accessStateHdrLen) + copy(hdr, accessStateMagic) + _, writeErr = w.Write(hdr) + for buf := range encoded { + if writeErr == nil { + _, writeErr = w.Write(buf) + } + } + if writeErr == nil { + writeErr = w.Flush() + } + }() + + c.mu.Lock() + total := c.lru.Len() + writtenAt := c.lru.nowFn() + var count uint64 + batch := make([]*entry, 0, batchSize) + for e := c.lru.ll.Front(); e != nil; e = e.Next() { + kv := e.Value.(*entry) + if kv.value.accessState == itemStateBootScanned { + continue + } + batch = append(batch, kv) + count++ + if len(batch) == batchSize { + batches <- batch + batch = make([]*entry, 0, batchSize) + } + } + if len(batch) > 0 { + batches <- batch + } + close(batches) + encoders.Wait() + close(encoded) + <-writerDone + c.mu.Unlock() + + err = writeErr + if err == nil { + hdr := make([]byte, accessStateHdrLen) + copy(hdr, accessStateMagic) + binary.LittleEndian.PutUint32(hdr[len(accessStateMagic):], writtenAt) + binary.LittleEndian.PutUint64(hdr[len(accessStateMagic)+4:], count) + _, err = f.WriteAt(hdr, 0) + } + if err == nil { + err = f.Sync() + } + if cerr := f.Close(); err == nil { + err = cerr + } + if err == nil { + err = os.Rename(tmp, final) + } + if err != nil { + _ = os.Remove(tmp) + return fmt.Errorf("failed to save access state: %w", err) + } + log.Printf("Saved access state for %d of %d index entries in %s", count, total, time.Since(start).Round(time.Millisecond)) + return nil +} + +// consumeAccessState reads and deletes the access-state file left by the +// previous process, if any. Errors are logged, not returned: the file is +// advisory and the boot scan proceeds on file timestamps without it. +func consumeAccessState(dir string) *accessStateIndex { + _ = os.Remove(path.Join(dir, accessStateTempName)) + final := path.Join(dir, accessStateFileName) + f, err := os.Open(final) + if err != nil { + if !errors.Is(err, os.ErrNotExist) { + log.Printf("Ignoring access state file: %v", err) + } + return nil + } + start := time.Now() + idx, err := readAccessState(f) + _ = f.Close() + // Consumed exactly once: a later crash must not replay a stale file. + if rmErr := os.Remove(final); rmErr != nil { + log.Printf("Failed to remove consumed access state file: %v", rmErr) + } + if err != nil { + log.Printf("Ignoring access state file: %v", err) + return nil + } + log.Printf("Loaded access state for %d entries (written %s) in %s", + len(idx.records), time.Unix(int64(idx.writtenAt), 0).UTC().Format(time.RFC3339), + time.Since(start).Round(time.Millisecond)) + return idx +} + +func readAccessState(r io.Reader) (*accessStateIndex, error) { + br := bufio.NewReaderSize(r, 4<<20) + hdr := make([]byte, accessStateHdrLen) + if _, err := io.ReadFull(br, hdr); err != nil { + return nil, fmt.Errorf("short header: %w", err) + } + if string(hdr[:len(accessStateMagic)]) != accessStateMagic { + return nil, errors.New("bad magic") + } + idx := &accessStateIndex{ + writtenAt: binary.LittleEndian.Uint32(hdr[len(accessStateMagic):]), + } + count := binary.LittleEndian.Uint64(hdr[len(accessStateMagic)+4:]) + if count > 1<<32 { + return nil, fmt.Errorf("implausible record count %d", count) + } + idx.records = make(map[accessStateKey]accessStateRecord, int(count)) + rec := make([]byte, accessStateRecLen) + for i := uint64(0); i < count; i++ { + if _, err := io.ReadFull(br, rec); err != nil { + return nil, fmt.Errorf("truncated after %d of %d records: %w", i, count, err) + } + state := rec[20] + if state != itemStateNew && state != itemStateTouched { + return nil, fmt.Errorf("record %d has unknown state %d", i, state) + } + var k accessStateKey + copy(k[:], rec[:16]) + idx.records[k] = accessStateRecord{ + lastAccess: binary.LittleEndian.Uint32(rec[16:]), + state: state, + } + } + return idx, nil +} diff --git a/cache/disk/access_state_test.go b/cache/disk/access_state_test.go new file mode 100644 index 0000000..2ed9e0b --- /dev/null +++ b/cache/disk/access_state_test.go @@ -0,0 +1,165 @@ +package disk + +import ( + "bytes" + "context" + "os" + "path" + "testing" + "time" + + "github.com/buchgr/bazel-remote/v2/cache" + testutils "github.com/buchgr/bazel-remote/v2/utils" +) + +func putBlob(t *testing.T, c Cache, size int64) string { + t.Helper() + data, hash := testutils.RandomDataAndHash(size) + if err := c.Put(context.Background(), cache.CAS, hash, size, bytes.NewReader(data)); err != nil { + t.Fatal(err) + } + return hash +} + +func peekState(t *testing.T, c Cache, hash string) lruItem { + t.Helper() + dc := c.(*diskCache) + dc.mu.Lock() + defer dc.mu.Unlock() + item, ok := dc.lru.Peek(cache.LookupKey(cache.CAS, hash)) + if !ok { + t.Fatalf("%s not in index", hash) + } + return item +} + +func TestAccessStateSurvivesRestart(t *testing.T) { + cacheDir := testutils.TempDir(t) + defer func() { _ = os.RemoveAll(cacheDir) }() + + c, err := New(cacheDir, 1024*1024, WithAccessLogger(testutils.NewSilentLogger())) + if err != nil { + t.Fatal(err) + } + // The boot scan trusts a record only for blobs created (file mtime) + // no later than the file was written, so the index clock must stay + // real time; the read is placed well after the writes. + dc := c.(*diskCache) + clock := uint32(time.Now().Unix()) + dc.lru.nowFn = func() uint32 { return clock } + + read := putBlob(t, c, 100) + unread := putBlob(t, c, 100) + clock += 1000 + readAt := clock + rc, _, err := c.Get(context.Background(), cache.CAS, read, 100, 0) + if err != nil { + t.Fatal(err) + } + _ = rc.Close() + + if err := c.SaveAccessState(); err != nil { + t.Fatal(err) + } + if _, err := os.Stat(path.Join(cacheDir, accessStateFileName)); err != nil { + t.Fatal(err) + } + + // A blob written after the file was saved has no record and is + // genuinely unknown to the next process. + later := putBlob(t, c, 100) + + restarted, err := New(cacheDir, 1024*1024, WithAccessLogger(testutils.NewSilentLogger())) + if err != nil { + t.Fatal(err) + } + if _, err := os.Stat(path.Join(cacheDir, accessStateFileName)); !os.IsNotExist(err) { + t.Errorf("access state file not consumed: %v", err) + } + + if got := peekState(t, restarted, read); got.accessState != itemStateTouched || got.lastAccess != readAt { + t.Errorf("read blob: state %d lastAccess %d, want touched/%d", got.accessState, got.lastAccess, readAt) + } + if got := peekState(t, restarted, unread); got.accessState != itemStateNew { + t.Errorf("unread blob: state %d, want new", got.accessState) + } + if got := peekState(t, restarted, later); got.accessState != itemStateBootScanned { + t.Errorf("later blob: state %d, want boot-scanned", got.accessState) + } + + // The census baseline reflects the restored states: only the later blob + // is unknown, only the unread one is never-read. + rdc := restarted.(*diskCache) + rdc.census.mu.Lock() + acc := rdc.census.tenant("") + class := censusClassIdx(cache.CAS, 100) + unknown := acc.residentUnknownBytesByClass[class] + neverRead := acc.residentNeverReadBytesByClass[class] + rdc.census.mu.Unlock() + laterItem := peekState(t, restarted, later) + unreadItem := peekState(t, restarted, unread) + if unknown != laterItem.sizeOnDisk || neverRead != unreadItem.sizeOnDisk { + t.Errorf("census unknown/never-read = %d/%d, want %d/%d", unknown, neverRead, laterItem.sizeOnDisk, unreadItem.sizeOnDisk) + } + + // Restored recency orders the LRU: the blob read 1000 s after the + // others were written is the most recent entry, not a timestamp tie. + rdc.mu.Lock() + frontKey := rdc.lru.ll.Front().Value.(*entry).key + rdc.mu.Unlock() + if frontKey != cache.LookupKey(cache.CAS, read) { + t.Errorf("LRU front = %q, want the read blob", frontKey) + } +} + +func TestAccessStateIgnoresCorruptFile(t *testing.T) { + cacheDir := testutils.TempDir(t) + defer func() { _ = os.RemoveAll(cacheDir) }() + + c, err := New(cacheDir, 1024*1024, WithAccessLogger(testutils.NewSilentLogger())) + if err != nil { + t.Fatal(err) + } + hash := putBlob(t, c, 100) + + if err := os.WriteFile(path.Join(cacheDir, accessStateFileName), []byte("garbage"), 0644); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(path.Join(cacheDir, accessStateTempName), []byte("partial"), 0644); err != nil { + t.Fatal(err) + } + + restarted, err := New(cacheDir, 1024*1024, WithAccessLogger(testutils.NewSilentLogger())) + if err != nil { + t.Fatal(err) + } + for _, name := range []string{accessStateFileName, accessStateTempName} { + if _, err := os.Stat(path.Join(cacheDir, name)); !os.IsNotExist(err) { + t.Errorf("%s not removed: %v", name, err) + } + } + if got := peekState(t, restarted, hash); got.accessState != itemStateBootScanned { + t.Errorf("state %d, want boot-scanned", got.accessState) + } +} + +func TestAccessStateRoundTrip(t *testing.T) { + idx := &accessStateIndex{writtenAt: 100, records: map[accessStateKey]accessStateRecord{}} + key := cache.LookupKey(cache.AC, testHashA) + idx.records[accessStateKeyFor(key)] = accessStateRecord{lastAccess: 50, state: itemStateTouched} + + if _, ok := idx.lookup(key, 101); ok { + t.Error("record applied to a blob created after the file was written") + } + rec, ok := idx.lookup(key, 80) + if !ok || rec.lastAccess != 80 || rec.state != itemStateTouched { + t.Errorf("lookup = %+v/%v, want lastAccess clamped to addedAt 80", rec, ok) + } + if _, ok := idx.lookup(cache.LookupKey(cache.AC, testHashB), 10); ok { + t.Error("unexpected record for an unknown key") + } + var nilIdx *accessStateIndex + if _, ok := nilIdx.lookup(key, 10); ok { + t.Error("nil index returned a record") + } +} diff --git a/cache/disk/disk.go b/cache/disk/disk.go index c8d28b0..83dd3a6 100644 --- a/cache/disk/disk.go +++ b/cache/disk/disk.go @@ -50,6 +50,9 @@ type Cache interface { PutZstd(ctx context.Context, hash string, size int64, r io.Reader) error Contains(ctx context.Context, kind cache.EntryKind, hash string, size int64) (bool, int64) FindMissingCasBlobs(ctx context.Context, blobs []*pb.Digest) ([]*pb.Digest, error) + // SaveAccessState persists the index's read history for the next boot + // scan (see access_state.go). Call once the servers have stopped. + SaveAccessState() error MaxSize() int64 Stats() (totalSize int64, reservedSize int64, numItems int, uncompressedSize int64) diff --git a/cache/disk/load.go b/cache/disk/load.go index 65f9bb4..7beb08f 100644 --- a/cache/disk/load.go +++ b/cache/disk/load.go @@ -13,6 +13,7 @@ import ( "strconv" "strings" "sync" + "sync/atomic" "time" "github.com/buchgr/bazel-remote/v2/cache" @@ -362,7 +363,8 @@ var incompleteFilesRemoved = promauto.NewCounter(prometheus.CounterOpts{ Help: "Files the boot scan removed because they were still marked in progress (partially written, or stored as received and not yet verified) when the previous process exited.", }) -func (c *diskCache) scanDir() (scanResult, error) { +func (c *diskCache) scanDir(bootAccess *accessStateIndex) (scanResult, error) { + var restored atomic.Int64 numWorkers := runtime.NumCPU() if numWorkers < 4 { @@ -529,6 +531,15 @@ func (c *diskCache) scanDir() (scanResult, error) { if item[n].lastAccess < item[n].addedAt { item[n].lastAccess = item[n].addedAt } + // The previous process's read history, when it left + // one, beats file timestamps: real state for the + // census and real recency for the LRU order. + if rec, ok := bootAccess.lookup(metadata[n].lookupKey, item[n].addedAt); ok { + item[n].lastAccess = rec.lastAccess + item[n].accessState = rec.state + metadata[n].ts = time.Unix(int64(rec.lastAccess), 0) + restored.Add(1) + } n++ } @@ -561,6 +572,9 @@ func (c *diskCache) scanDir() (scanResult, error) { if strings.ToLower(name) == lowercaseDSStoreFile { continue } + if rootRel == "" && (name == accessStateFileName || name == accessStateTempName) { + continue + } return fmt.Errorf("Unexpected file: %s", path.Join(rootRel, name)) } @@ -665,6 +679,9 @@ func (c *diskCache) scanDir() (scanResult, error) { <-received + if bootAccess != nil { + log.Printf("Restored access state for %d index entries", restored.Load()) + } return finalScanResult, nil } @@ -674,7 +691,7 @@ func (c *diskCache) scanDir() (scanResult, error) { func (c *diskCache) loadExistingFiles(maxSizeBytes int64, cc CacheConfig) error { log.Printf("Loading existing files in %s.\n", c.dir) - result, err := c.scanDir() + result, err := c.scanDir(consumeAccessState(c.dir)) if err != nil { log.Printf("Failed to scan cache dir: %s", err.Error()) return err diff --git a/main.go b/main.go index 5c1363a..b5827e7 100644 --- a/main.go +++ b/main.go @@ -324,8 +324,12 @@ func run(ctx *cli.Context) error { } err = servers.Wait() - // Both servers have stopped accepting requests, so no new observations - // can arrive: flush what remains and wait for in-flight uploads. + // Both servers have stopped accepting requests, so the index is final: + // persist its read history for the next boot scan, then flush what + // remains of the LRU observations and wait for in-flight uploads. + if saveErr := diskCache.SaveAccessState(); saveErr != nil { + log.Println(saveErr) + } if lruFlusher != nil { log.Println("Draining buffered LRU observation artifacts") lruFlusher.Drain() diff --git a/server/grpc_bytestream_write_contract_test.go b/server/grpc_bytestream_write_contract_test.go index b022c83..9c8f828 100644 --- a/server/grpc_bytestream_write_contract_test.go +++ b/server/grpc_bytestream_write_contract_test.go @@ -70,6 +70,8 @@ func (c *consumptionTrackingCache) GetZstd(ctx context.Context, hash string, siz panic("not used by Write") } +func (c *consumptionTrackingCache) SaveAccessState() error { return nil } + func (c *consumptionTrackingCache) FindMissingCasBlobs(ctx context.Context, blobs []*pb.Digest) ([]*pb.Digest, error) { panic("not used by Write") } diff --git a/server/grpc_test.go b/server/grpc_test.go index dc58ab2..c19a378 100644 --- a/server/grpc_test.go +++ b/server/grpc_test.go @@ -2888,6 +2888,8 @@ func (c *StubCache) Contains(ctx context.Context, kind cache.EntryKind, hash str return false, -1 } +func (c *StubCache) SaveAccessState() error { return nil } + func (c *StubCache) FindMissingCasBlobs(ctx context.Context, blobs []*pb.Digest) ([]*pb.Digest, error) { return blobs, nil } From d73ddde43140d39f8a1117c72e1fcba0647abfc2 Mon Sep 17 00:00:00 2001 From: "shreyas@blacksmith.sh" Date: Mon, 2 Nov 2026 15:41:09 +0000 Subject: [PATCH 5/5] cache/disk: leftover access-state files never fail the boot scan Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- cache/disk/access_state_test.go | 21 +++++++++++++++++++++ cache/disk/load.go | 5 +++++ 2 files changed, 26 insertions(+) diff --git a/cache/disk/access_state_test.go b/cache/disk/access_state_test.go index 2ed9e0b..b951454 100644 --- a/cache/disk/access_state_test.go +++ b/cache/disk/access_state_test.go @@ -163,3 +163,24 @@ func TestAccessStateRoundTrip(t *testing.T) { t.Error("nil index returned a record") } } + +func TestAccessStateFilesNeverFailTheScan(t *testing.T) { + cacheDir := testutils.TempDir(t) + defer func() { _ = os.RemoveAll(cacheDir) }() + + c, err := New(cacheDir, 1024*1024, WithAccessLogger(testutils.NewSilentLogger())) + if err != nil { + t.Fatal(err) + } + putBlob(t, c, 100) + for _, name := range []string{accessStateFileName, accessStateTempName} { + if err := os.WriteFile(path.Join(cacheDir, name), []byte("x"), 0644); err != nil { + t.Fatal(err) + } + } + // The scan itself must tolerate the files, whether or not + // consumeAccessState managed to delete them first. + if _, err := c.(*diskCache).scanDir(nil); err != nil { + t.Fatalf("scanDir with leftover access-state files: %v", err) + } +} diff --git a/cache/disk/load.go b/cache/disk/load.go index 7beb08f..33cadd4 100644 --- a/cache/disk/load.go +++ b/cache/disk/load.go @@ -646,6 +646,11 @@ func (c *diskCache) scanDir(bootAccess *accessStateIndex) (scanResult, error) { if strings.ToLower(name) == lowercaseDSStoreFile { continue } + // consumeAccessState deletes these before the scan; if the + // unlink failed they are still only advisory, never fatal. + if name == accessStateFileName || name == accessStateTempName { + continue + } return scanResult{}, fmt.Errorf("unexpected file: %s", name) }