From 3ffb0669f1c7c699daae6483dcfbb3bd7f583581 Mon Sep 17 00:00:00 2001 From: Shreyas Kalyan Date: Sun, 2 Aug 2026 21:46:02 -0400 Subject: [PATCH 1/5] s3proxy: owed-upload ledger, sweeper, and enqueue coalescing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Restore the L1 ⊆ MinIO subset invariant that lossy write-through shedding violated: FindMissingBlobs answers local-first, so a blob whose backend upload was shed is reported present forever and never reaches MinIO. - Queue-full sheds and terminal upload failures (failed PUT, breaker refusal) record the upload identity in a bounded, snapshotted per-backend owed ledger instead of being silently lost. - A background sweeper re-enqueues owed items through the normal queue (same workers, breaker, outcome accounting), yielding to live traffic: breaker closed + queue under half capacity only. Locally-evicted owed blobs settle as void (honestly missing everywhere; the client re-upload path heals both tiers). - Duplicate enqueues coalesce against an in-flight identity set (kind+hash+prefix+bucket): matrix fan-out re-delivers the same blob from many hosts, MinIO create-if-absent 412s all but one, and every queued duplicate pins an open FD. - Standalone-only: wired via WithOwedLedgerDir + a BlobSource back-reference injected after disk-cache construction. The host-side fallback proxy keeps best-effort semantics. - New series: owed_uploads gauge, owed_sweep_total{result}, owed_rejected_total, upload_queue_coalesced_total. Co-authored-by: Cursor --- cache/disk/owed.go | 42 ++++ cache/disk/owed_test.go | 95 +++++++++ cache/s3proxy/breaker.go | 6 +- cache/s3proxy/owed.go | 407 +++++++++++++++++++++++++++++++++++++ cache/s3proxy/owed_test.go | 253 +++++++++++++++++++++++ cache/s3proxy/s3proxy.go | 60 +++++- config/proxy.go | 11 +- main.go | 11 + 8 files changed, 877 insertions(+), 8 deletions(-) create mode 100644 cache/disk/owed.go create mode 100644 cache/disk/owed_test.go create mode 100644 cache/s3proxy/owed.go create mode 100644 cache/s3proxy/owed_test.go diff --git a/cache/disk/owed.go b/cache/disk/owed.go new file mode 100644 index 0000000..2bcaf9c --- /dev/null +++ b/cache/disk/owed.go @@ -0,0 +1,42 @@ +package disk + +import ( + "context" + "errors" + "io" + "os" + "path" + + "github.com/buchgr/bazel-remote/v2/cache" +) + +var errOwedBlobUnavailable = errors.New("owed blob not present in local cache") + +// OpenOwedBlob implements s3proxy.BlobSource: it reopens a committed blob's +// raw on-disk representation (header + compression included — the same bytes +// Put streams to the proxy) for a deferred backend upload, returning the +// reader and the on-disk size. ctx must carry the entry's request-scoped +// storage prefix when it had one, because both the lookup key and the file +// location are prefix-scoped. +// +// An evicted or unreadable blob returns an error. Callers treat that as +// debt-settled: once the blob is gone locally, FindMissingBlobs reports it +// honestly missing and the normal client re-upload path restores both tiers. +func (c *diskCache) OpenOwedBlob(ctx context.Context, kind cache.EntryKind, hash string) (io.ReadCloser, int64, error) { + key := cache.LookupKeyForContext(ctx, kind, hash) + + c.mu.Lock() + item, listElem := c.lru.Get(key) + if listElem == nil { + c.mu.Unlock() + return nil, -1, errOwedBlobUnavailable + } + blobPath := path.Join(c.dir, c.FileLocationForContext(ctx, kind, item.legacy, hash, item.size, item.random)) + f, err := os.Open(blobPath) + c.mu.Unlock() + + if err != nil { + return nil, -1, errOwedBlobUnavailable + } + return f, item.sizeOnDisk, nil +} diff --git a/cache/disk/owed_test.go b/cache/disk/owed_test.go new file mode 100644 index 0000000..69c168a --- /dev/null +++ b/cache/disk/owed_test.go @@ -0,0 +1,95 @@ +package disk + +import ( + "bytes" + "context" + "io" + "os" + "testing" + + "github.com/buchgr/bazel-remote/v2/cache" + testutils "github.com/buchgr/bazel-remote/v2/utils" +) + +func TestOpenOwedBlobReturnsRawOnDiskBytes(t *testing.T) { + ctx := context.Background() + + cacheDir := tempDir(t) + defer func() { _ = os.RemoveAll(cacheDir) }() + + testCacheI, err := New(cacheDir, 1024*1024, WithAccessLogger(testutils.NewSilentLogger())) + if err != nil { + t.Fatal(err) + } + testCache := testCacheI.(*diskCache) + + data, hash := testutils.RandomDataAndHash(256) + if err := testCache.Put(ctx, cache.CAS, hash, 256, io.NopCloser(bytes.NewReader(data))); err != nil { + t.Fatal(err) + } + + rc, sizeOnDisk, err := testCache.OpenOwedBlob(ctx, cache.CAS, hash) + if err != nil { + t.Fatalf("OpenOwedBlob on a committed blob: %v", err) + } + raw, err := io.ReadAll(rc) + _ = rc.Close() + if err != nil { + t.Fatal(err) + } + if int64(len(raw)) != sizeOnDisk { + t.Fatalf("read %d raw bytes, want reported sizeOnDisk %d", len(raw), sizeOnDisk) + } + + // The raw representation must be exactly the committed file's content — + // the same bytes disk.Put streams to the proxy. + key := cache.LookupKey(cache.CAS, hash) + testCache.mu.Lock() + item, el := testCache.lru.Get(key) + testCache.mu.Unlock() + if el == nil { + t.Fatal("blob missing from LRU") + } + onDisk, err := os.ReadFile(cacheDir + "/" + testCache.FileLocation(cache.CAS, item.legacy, hash, item.size, item.random)) + if err != nil { + t.Fatal(err) + } + if !bytes.Equal(raw, onDisk) { + t.Fatal("OpenOwedBlob bytes differ from the committed on-disk file") + } + + // Unknown hash behaves like an evicted blob. + _, unknownHash := testutils.RandomDataAndHash(64) + if _, _, err := testCache.OpenOwedBlob(ctx, cache.CAS, unknownHash); err == nil { + t.Fatal("OpenOwedBlob on an absent blob returned nil error") + } +} + +func TestOpenOwedBlobIsStoragePrefixScoped(t *testing.T) { + cacheDir := tempDir(t) + defer func() { _ = os.RemoveAll(cacheDir) }() + + testCacheI, err := New(cacheDir, 1024*1024, WithAccessLogger(testutils.NewSilentLogger())) + if err != nil { + t.Fatal(err) + } + testCache := testCacheI.(*diskCache) + + prefixCtx := cache.WithStoragePrefix(context.Background(), "tenant-a/prod") + data, hash := testutils.RandomDataAndHash(128) + if err := testCache.Put(prefixCtx, cache.CAS, hash, 128, io.NopCloser(bytes.NewReader(data))); err != nil { + t.Fatal(err) + } + + // The prefix travels in the context, exactly as the sweeper rebuilds it. + rc, _, err := testCache.OpenOwedBlob(prefixCtx, cache.CAS, hash) + if err != nil { + t.Fatalf("OpenOwedBlob with matching prefix ctx: %v", err) + } + _ = rc.Close() + + // Without the prefix the lookup key differs: no cross-tenant leakage. + if _, _, err := testCache.OpenOwedBlob(context.Background(), cache.CAS, hash); err == nil { + t.Fatal("OpenOwedBlob without the storage prefix found a prefix-scoped blob") + } +} diff --git a/cache/s3proxy/breaker.go b/cache/s3proxy/breaker.go index 1f5cd45..0747963 100644 --- a/cache/s3proxy/breaker.go +++ b/cache/s3proxy/breaker.go @@ -62,9 +62,9 @@ var breakerTimeout = 15 * time.Second // Breaker states. The numeric values are the wire contract of the // bazel_remote_s3_breaker_state gauge. const ( - breakerClosed int32 = iota // 0 - breakerHalfOpen // 1 - breakerOpen // 2 + breakerClosed int32 = iota // 0 + breakerHalfOpen // 1 + breakerOpen // 2 ) func breakerStateName(state int32) string { diff --git a/cache/s3proxy/owed.go b/cache/s3proxy/owed.go new file mode 100644 index 0000000..e890281 --- /dev/null +++ b/cache/s3proxy/owed.go @@ -0,0 +1,407 @@ +package s3proxy + +// Owed-upload machinery: restores the "L1 contents ⊆ MinIO contents" +// invariant that lossy write-through shedding silently violates. +// +// Why this exists: the disk cache answers FindMissingBlobs local-first, so a +// blob that landed on the L1 but whose backend upload was shed is reported +// "present" to clients — nothing ever re-uploads it, and MinIO permanently +// lacks an object the footprint accounting may already have counted. Shedding +// therefore must be a DEFERRAL, not a loss: +// +// - Put() overflow and terminal upload failures record the item in a +// per-backend owed ledger (bounded, snapshotted to disk so restarts +// don't forget the debt). +// - A background sweeper re-enqueues owed items through the normal upload +// queue — the same worker pool, breaker (ExecuteNoProbe), and outcome +// accounting govern the retry. The sweeper yields to live traffic: it +// only runs while the breaker is closed and the queue is under half +// full, so convergence never competes with builds. +// - Upload success (created or already_exists) settles the debt. +// +// The inflight set doubles as upload coalescing: cross-host matrix fan-out +// makes the same missing blob arrive from every host at once, and MinIO +// would reject the duplicates with 412 anyway (create-if-absent), so the +// duplicate enqueue is dropped at the door and its reader (an open FD — the +// queue's real cost) is released immediately. + +import ( + "context" + "encoding/json" + "io" + "os" + "path/filepath" + "strings" + "sync" + "time" + + "github.com/buchgr/bazel-remote/v2/cache" + "github.com/buchgr/bazel-remote/v2/utils/backendproxy" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" +) + +const ( + // maxOwedEntries bounds one backend's ledger. At ~200 bytes/entry this + // is ~50 MiB of memory and snapshot. When the ledger is full, new debts + // are REJECTED (metered) rather than evicting older ones: older entries + // are closer to LRU eviction, which self-resolves them, while a + // saturated ledger is an alertable capacity signal either way. + maxOwedEntries = 1 << 18 + + // sweepInterval paces the background sweeper. Convergence within + // minutes-to-hours is the goal; this is deliberately unhurried. + sweepInterval = 15 * time.Second + + // sweepBatch caps how many owed items one pass re-enqueues. + sweepBatch = 256 + + // sweepQueueHeadroom: the sweeper only injects work while the live + // queue is under this fraction of capacity, so deferred uploads never + // compete with current build traffic for queue slots or workers. + sweepQueueHeadroom = 0.5 +) + +var ( + uploadQueueCoalesced = promauto.NewCounterVec(prometheus.CounterOpts{ + Name: "bazel_remote_s3_upload_queue_coalesced_total", + Help: "Backend uploads dropped at enqueue because an identical upload (kind+hash+prefix+bucket) was already queued or in flight.", + }, []string{"backend"}) + owedBacklog = promauto.NewGaugeVec(prometheus.GaugeOpts{ + Name: "bazel_remote_s3_owed_uploads", + Help: "Blobs present locally but owed to the S3 backend (shed or failed write-throughs awaiting the sweeper).", + }, []string{"backend"}) + owedSweep = promauto.NewCounterVec(prometheus.CounterOpts{ + Name: "bazel_remote_s3_owed_sweep_total", + Help: "Owed-upload sweeper dispositions (requeued, blob_evicted).", + }, []string{"backend", "result"}) + owedRejected = promauto.NewCounterVec(prometheus.CounterOpts{ + Name: "bazel_remote_s3_owed_rejected_total", + Help: "Owed-upload records rejected because the ledger is at capacity (permanent MinIO gap until the blob is evicted and re-uploaded).", + }, []string{"backend"}) +) + +// BlobSource reopens a committed blob's raw on-disk representation (exactly +// the bytes the original write-through would have streamed) for a deferred +// upload. ctx carries the entry's storage prefix when it was request-scoped, +// because the disk layout and lookup key are prefix-scoped. An evicted or +// missing blob returns an error; callers settle the debt (eviction makes the +// blob honestly absent everywhere, so FindMissingBlobs heals it the normal +// way). +type BlobSource interface { + OpenOwedBlob(ctx context.Context, kind cache.EntryKind, hash string) (io.ReadCloser, int64, error) +} + +// BlobSourceSetter is implemented by the proxies returned from New/NewMulti. +// The disk cache is constructed AFTER the proxy (it takes the proxy as an +// argument), so the back-reference is injected post-construction. +type BlobSourceSetter interface { + SetBlobSource(src BlobSource) +} + +// uploadKey identifies one logical backend upload. Two uploads with the same +// key write the same object to the same place; queueing both is pure waste. +type uploadKey struct { + Kind cache.EntryKind `json:"kind"` + Hash string `json:"hash"` + Prefix string `json:"prefix"` + Bucket string `json:"bucket"` +} + +// owedEntry is everything needed to reconstruct an UploadReq later. Sizes are +// re-derived from disk at sweep time (the authoritative copy), but +// LogicalSize is kept for observer accounting parity. +type owedEntry struct { + Key uploadKey `json:"key"` + LogicalSize int64 `json:"logical_size"` + RequestScopedStoragePrefix bool `json:"request_scoped_prefix"` + RequireStoragePrefix bool `json:"require_prefix"` +} + +// inflightSet tracks uploads that are queued or being uploaded right now. +// Methods are safe on a nil receiver (hand-built caches in tests): a nil set +// admits everything and coalesces nothing. +type inflightSet struct { + mu sync.Mutex + m map[uploadKey]struct{} +} + +func newInflightSet() *inflightSet { + return &inflightSet{m: make(map[uploadKey]struct{})} +} + +// tryAdd returns false when the key is already inflight (coalesce case). +func (s *inflightSet) tryAdd(k uploadKey) bool { + if s == nil { + return true + } + s.mu.Lock() + defer s.mu.Unlock() + if _, dup := s.m[k]; dup { + return false + } + s.m[k] = struct{}{} + return true +} + +func (s *inflightSet) remove(k uploadKey) { + if s == nil { + return + } + s.mu.Lock() + delete(s.m, k) + s.mu.Unlock() +} + +// owedLedger is the bounded, snapshotted record of uploads the backend is +// still owed. All methods are safe on a nil receiver (feature disabled). +type owedLedger struct { + backendKey string + path string + + mu sync.Mutex + entries map[uploadKey]owedEntry + dirty bool +} + +// newOwedLedger loads any snapshot left by a previous process. A corrupt or +// unreadable snapshot starts empty and logs: the ledger is a convergence +// accelerator, never a correctness gate worth failing startup for. +func newOwedLedger(dir, backendKey string, errorLogger cache.Logger) *owedLedger { + l := &owedLedger{ + backendKey: backendKey, + path: filepath.Join(dir, "owed-uploads-"+sanitizeForFilename(backendKey)+".json"), + entries: make(map[uploadKey]owedEntry), + } + if err := os.MkdirAll(dir, 0o755); err != nil { + errorLogger.Printf("owed ledger: cannot create %s (%v); continuing without restart persistence", dir, err) + l.path = "" + return l + } + data, err := os.ReadFile(l.path) + if err == nil { + var snapshot []owedEntry + if jerr := json.Unmarshal(data, &snapshot); jerr != nil { + errorLogger.Printf("owed ledger: corrupt snapshot %s (%v); starting empty", l.path, jerr) + } else { + for _, e := range snapshot { + if len(l.entries) >= maxOwedEntries { + break + } + l.entries[e.Key] = e + } + } + } + owedBacklog.WithLabelValues(backendKey).Set(float64(len(l.entries))) + return l +} + +func (l *owedLedger) add(e owedEntry) { + if l == nil { + return + } + l.mu.Lock() + defer l.mu.Unlock() + if _, exists := l.entries[e.Key]; !exists && len(l.entries) >= maxOwedEntries { + owedRejected.WithLabelValues(l.backendKey).Inc() + return + } + l.entries[e.Key] = e + l.dirty = true + owedBacklog.WithLabelValues(l.backendKey).Set(float64(len(l.entries))) +} + +func (l *owedLedger) settle(k uploadKey) { + if l == nil { + return + } + l.mu.Lock() + defer l.mu.Unlock() + if _, ok := l.entries[k]; !ok { + return + } + delete(l.entries, k) + l.dirty = true + owedBacklog.WithLabelValues(l.backendKey).Set(float64(len(l.entries))) +} + +// batch returns up to n entries for a sweep pass, in map order (unordered — +// fairness across passes comes from settled entries leaving the map). +func (l *owedLedger) batch(n int) []owedEntry { + if l == nil { + return nil + } + l.mu.Lock() + defer l.mu.Unlock() + out := make([]owedEntry, 0, n) + for _, e := range l.entries { + out = append(out, e) + if len(out) == n { + break + } + } + return out +} + +// snapshotIfDirty atomically rewrites the on-disk snapshot. Called from the +// sweeper loop; a crash loses at most one sweepInterval of ledger changes, +// which the next drop or eviction re-discovers (documented, acceptable). +func (l *owedLedger) snapshotIfDirty(errorLogger cache.Logger) { + if l == nil || l.path == "" { + return + } + l.mu.Lock() + if !l.dirty { + l.mu.Unlock() + return + } + snapshot := make([]owedEntry, 0, len(l.entries)) + for _, e := range l.entries { + snapshot = append(snapshot, e) + } + l.dirty = false + l.mu.Unlock() + + data, err := json.Marshal(snapshot) + if err != nil { + errorLogger.Printf("owed ledger: marshal failed: %v", err) + return + } + tmp := l.path + ".tmp" + if err := os.WriteFile(tmp, data, 0o644); err != nil { + errorLogger.Printf("owed ledger: snapshot write failed: %v", err) + return + } + if err := os.Rename(tmp, l.path); err != nil { + errorLogger.Printf("owed ledger: snapshot rename failed: %v", err) + } +} + +func sanitizeForFilename(s string) string { + return strings.Map(func(r rune) rune { + switch { + case r >= 'a' && r <= 'z', r >= 'A' && r <= 'Z', r >= '0' && r <= '9', r == '-', r == '.': + return r + default: + return '_' + } + }, s) +} + +// resolveUploadIdentity mirrors UploadFile's prefix/bucket resolution so the +// identity computed at enqueue time matches the one settled at outcome time. +func (c *s3Cache) resolveUploadIdentity(item backendproxy.UploadReq) uploadKey { + prefix := item.StoragePrefix + if item.Kind == cache.RAW || prefix == "" { + prefix = c.prefix + } + bucket := item.S3Backend.Bucket + if bucket == "" { + bucket = c.bucket + } + return uploadKey{Kind: item.Kind, Hash: item.Hash, Prefix: prefix, Bucket: bucket} +} + +func (c *s3Cache) owedEntryForItem(key uploadKey, item backendproxy.UploadReq) owedEntry { + requestScoped := item.RequestScopedStoragePrefix + if item.Kind == cache.RAW { + requestScoped = false + } + return owedEntry{ + Key: key, + LogicalSize: item.LogicalSize, + RequestScopedStoragePrefix: requestScoped, + RequireStoragePrefix: item.RequireStoragePrefix && requestScoped, + } +} + +// SetBlobSource injects the disk cache back-reference and starts the sweeper. +// Called once, after the disk cache is constructed. Without a blob source the +// ledger still records and persists debts (visible in metrics and settled by +// live re-uploads), but nothing can proactively drain it. +func (c *s3Cache) SetBlobSource(src BlobSource) { + if c.owed == nil || src == nil { + return + } + c.blobSourceOnce.Do(func() { + c.blobSource = src + go c.sweepOwedLoop() + }) +} + +func (m *multiS3Cache) SetBlobSource(src BlobSource) { + for _, backend := range m.backends { + backend.SetBlobSource(src) + } +} + +func (c *s3Cache) sweepOwedLoop() { + ticker := time.NewTicker(sweepInterval) + defer ticker.Stop() + for range ticker.C { + c.sweepOwedOnce() + c.owed.snapshotIfDirty(c.errorLogger) + } +} + +// sweepOwedOnce re-enqueues one batch of owed uploads if the backend is +// healthy and the queue has headroom. Every re-enqueued item flows through +// the normal worker/breaker/outcome path; settlement happens there. +func (c *s3Cache) sweepOwedOnce() { + if c.owed == nil || c.blobSource == nil || c.uploadQueue == nil { + return + } + if c.breaker != nil && c.breaker.State() != breakerClosed { + return + } + if len(c.uploadQueue) >= int(sweepQueueHeadroom*float64(cap(c.uploadQueue))) { + return + } + + for _, entry := range c.owed.batch(sweepBatch) { + if len(c.uploadQueue) >= int(sweepQueueHeadroom*float64(cap(c.uploadQueue))) { + return + } + if !c.inflight.tryAdd(entry.Key) { + continue // already queued or uploading; its outcome will settle it + } + + ctx := context.Background() + if entry.RequestScopedStoragePrefix { + ctx = cache.WithStoragePrefix(ctx, entry.Key.Prefix) + } + rc, sizeOnDisk, err := c.blobSource.OpenOwedBlob(ctx, entry.Key.Kind, entry.Key.Hash) + if err != nil { + // Evicted (or unreadable): the blob is honestly absent + // everywhere now, so FindMissingBlobs will report it missing + // and a future client upload recreates both copies. Debt void. + c.inflight.remove(entry.Key) + c.owed.settle(entry.Key) + owedSweep.WithLabelValues(c.key, "blob_evicted").Inc() + continue + } + + req := backendproxy.UploadReq{ + Hash: entry.Key.Hash, + LogicalSize: entry.LogicalSize, + SizeOnDisk: sizeOnDisk, + Kind: entry.Key.Kind, + Rc: rc, + StoragePrefix: entry.Key.Prefix, + RequestScopedStoragePrefix: entry.RequestScopedStoragePrefix, + RequireStoragePrefix: entry.RequireStoragePrefix, + S3Backend: cache.S3BackendSelection{Bucket: entry.Key.Bucket}, + } + select { + case c.uploadQueue <- req: + owedSweep.WithLabelValues(c.key, "requeued").Inc() + uploadQueueDepth.WithLabelValues(c.key).Set(float64(len(c.uploadQueue))) + default: + // Lost the headroom race; stay owed, try next pass. + c.inflight.remove(entry.Key) + _ = rc.Close() + return + } + } +} diff --git a/cache/s3proxy/owed_test.go b/cache/s3proxy/owed_test.go new file mode 100644 index 0000000..d5b32d5 --- /dev/null +++ b/cache/s3proxy/owed_test.go @@ -0,0 +1,253 @@ +package s3proxy + +import ( + "bytes" + "context" + "errors" + "io" + stdlog "log" + "net/http/httptest" + "os" + "path/filepath" + "strconv" + "strings" + "testing" + + "github.com/buchgr/bazel-remote/v2/cache" + "github.com/buchgr/bazel-remote/v2/utils/backendproxy" + "github.com/johannesboyne/gofakes3" + "github.com/johannesboyne/gofakes3/backend/s3mem" + "github.com/prometheus/client_golang/prometheus/testutil" +) + +// fakeBlobSource serves owed blobs from a map; missing keys error like an +// evicted blob. +type fakeBlobSource struct { + blobs map[string][]byte + calls int +} + +func (f *fakeBlobSource) OpenOwedBlob(_ context.Context, _ cache.EntryKind, hash string) (io.ReadCloser, int64, error) { + f.calls++ + data, ok := f.blobs[hash] + if !ok { + return nil, -1, errors.New("evicted") + } + return io.NopCloser(bytes.NewReader(data)), int64(len(data)), nil +} + +func owedTestCache(t *testing.T, name string, queueCap int) (*s3Cache, chan backendproxy.UploadReq) { + t.Helper() + queue := make(chan backendproxy.UploadReq, queueCap) + logger := stdlog.New(&bytes.Buffer{}, "", 0) + c := &s3Cache{ + key: name, + bucket: "test-bucket", + breaker: newBreaker(name, nil), + objectKey: objectKeyV1, + readDeadline: readDeadline, + accessLogger: logger, + errorLogger: logger, + inflight: newInflightSet(), + owed: newOwedLedger(t.TempDir(), name, logger), + uploadQueue: queue, + } + return c, queue +} + +func TestDuplicateEnqueueCoalesced(t *testing.T) { + c, queue := owedTestCache(t, "owed-coalesce-test", 4) + + before := testutil.ToFloat64(uploadQueueCoalesced.WithLabelValues(c.key)) + c.Put(context.Background(), cache.CAS, testHash, 4, 4, io.NopCloser(strings.NewReader("blob"))) + c.Put(context.Background(), cache.CAS, testHash, 4, 4, io.NopCloser(strings.NewReader("blob"))) + + if got := len(queue); got != 1 { + t.Fatalf("queue depth after duplicate Put = %d, want 1", got) + } + if got := testutil.ToFloat64(uploadQueueCoalesced.WithLabelValues(c.key)) - before; got != 1 { + t.Fatalf("coalesced counter delta = %v, want 1", got) + } + + // A different hash is not coalesced. + otherHash := strings.Repeat("ab", 32) + c.Put(context.Background(), cache.CAS, otherHash, 4, 4, io.NopCloser(strings.NewReader("blob"))) + if got := len(queue); got != 2 { + t.Fatalf("queue depth after distinct Put = %d, want 2", got) + } +} + +func TestQueueFullRecordsOwedAndSweeperRepays(t *testing.T) { + backend := s3mem.New() + if err := backend.CreateBucket("test-bucket"); err != nil { + t.Fatal(err) + } + ts := httptest.NewServer(gofakes3.New(backend).Server()) + t.Cleanup(ts.Close) + + c, _ := owedTestCache(t, "owed-repay-test", 1) + c.mcore = coreFor(t, ts, 1) + + content := []byte("owed-blob-content") + + // Fill the queue, then overflow it with the blob we care about. + fillerHash := strings.Repeat("cd", 32) + c.Put(context.Background(), cache.CAS, fillerHash, 4, 4, io.NopCloser(strings.NewReader("fill"))) + c.Put(context.Background(), cache.CAS, testHash, int64(len(content)), int64(len(content)), + io.NopCloser(bytes.NewReader(content))) + + if got := testutil.ToFloat64(owedBacklog.WithLabelValues(c.key)); got != 1 { + t.Fatalf("owed backlog after shed = %v, want 1", got) + } + + // Sweep into a fresh, roomy queue (the shed queue stays full so the + // sweeper's headroom check would rightly refuse it). + sweepQueue := make(chan backendproxy.UploadReq, 8) + c.uploadQueue = sweepQueue + c.blobSource = &fakeBlobSource{blobs: map[string][]byte{testHash: content}} + c.sweepOwedOnce() + + if got := len(sweepQueue); got != 1 { + t.Fatalf("sweep queue depth = %d, want 1 requeued upload", got) + } + req := <-sweepQueue + if req.Hash != testHash || req.SizeOnDisk != int64(len(content)) { + t.Fatalf("requeued req = %+v, want hash %s size %d", req, testHash, len(content)) + } + + // Run the upload; success must settle the debt and store the object. + c.UploadFile(req) + if got := testutil.ToFloat64(owedBacklog.WithLabelValues(c.key)); got != 0 { + t.Fatalf("owed backlog after successful repay = %v, want 0", got) + } + rc, _, err := c.Get(context.Background(), cache.CAS, testHash, -1) + if err != nil || rc == nil { + t.Fatalf("Get after repay = (%v, %v), want hit", rc, err) + } + stored, err := io.ReadAll(rc) + _ = rc.Close() + if err != nil || !bytes.Equal(stored, content) { + t.Fatalf("stored bytes = %q (err %v), want %q", stored, err, content) + } +} + +func TestUploadFailureRecordsOwedAndBreakerBlocksSweep(t *testing.T) { + c, _ := owedTestCache(t, "owed-failure-test", 4) + + // A breaker-refused upload is a terminal failure: the debt must be + // recorded and the reader closed. + tripBreaker(t, c.breaker) + c.UploadFile(backendproxy.UploadReq{ + Hash: testHash, + Kind: cache.CAS, + SizeOnDisk: 4, + Rc: io.NopCloser(strings.NewReader("blob")), + }) + if got := testutil.ToFloat64(owedBacklog.WithLabelValues(c.key)); got != 1 { + t.Fatalf("owed backlog after breaker-refused upload = %v, want 1", got) + } + + // While the breaker is open the sweeper must not even consult the blob + // source — MinIO is sick; retrying now is the old thundering herd. + src := &fakeBlobSource{blobs: map[string][]byte{}} + c.blobSource = src + c.sweepOwedOnce() + if src.calls != 0 { + t.Fatalf("sweep consulted blob source %d times with breaker open, want 0", src.calls) + } +} + +func TestSweepEvictedBlobSettlesDebt(t *testing.T) { + c, queue := owedTestCache(t, "owed-evicted-test", 4) + c.owed.add(owedEntry{Key: uploadKey{Kind: cache.CAS, Hash: testHash, Bucket: "test-bucket"}}) + + c.blobSource = &fakeBlobSource{blobs: map[string][]byte{}} // nothing local + c.sweepOwedOnce() + + if got := testutil.ToFloat64(owedBacklog.WithLabelValues(c.key)); got != 0 { + t.Fatalf("owed backlog after evicted-blob sweep = %v, want 0 (debt void)", got) + } + if got := len(queue); got != 0 { + t.Fatalf("queue depth after evicted-blob sweep = %d, want 0", got) + } +} + +func TestSweepYieldsToBusyQueue(t *testing.T) { + c, queue := owedTestCache(t, "owed-yield-test", 4) + c.owed.add(owedEntry{Key: uploadKey{Kind: cache.CAS, Hash: testHash, Bucket: "test-bucket"}}) + + // 2 of 4 slots used = at the 50% headroom threshold: sweep must yield. + queue <- backendproxy.UploadReq{} + queue <- backendproxy.UploadReq{} + + src := &fakeBlobSource{blobs: map[string][]byte{testHash: []byte("blob")}} + c.blobSource = src + c.sweepOwedOnce() + + if src.calls != 0 { + t.Fatalf("sweep consulted blob source %d times with a busy queue, want 0", src.calls) + } + if got := testutil.ToFloat64(owedBacklog.WithLabelValues(c.key)); got != 1 { + t.Fatalf("owed backlog after yielded sweep = %v, want 1 (still owed)", got) + } +} + +func TestOwedLedgerSnapshotRoundtrip(t *testing.T) { + dir := t.TempDir() + logger := stdlog.New(&bytes.Buffer{}, "", 0) + + l := newOwedLedger(dir, "snap-test", logger) + entry := owedEntry{ + Key: uploadKey{Kind: cache.CAS, Hash: testHash, Prefix: "tenant-a", Bucket: "b"}, + LogicalSize: 42, + RequestScopedStoragePrefix: true, + RequireStoragePrefix: true, + } + l.add(entry) + l.snapshotIfDirty(logger) + + reloaded := newOwedLedger(dir, "snap-test", logger) + got := reloaded.batch(10) + if len(got) != 1 || got[0] != entry { + t.Fatalf("reloaded ledger = %+v, want [%+v]", got, entry) + } + + // A corrupt snapshot must start empty without failing. + if err := os.WriteFile(filepath.Join(dir, "owed-uploads-snap-test.json"), []byte("{not json"), 0o644); err != nil { + t.Fatal(err) + } + corrupt := newOwedLedger(dir, "snap-test", logger) + if got := corrupt.batch(10); len(got) != 0 { + t.Fatalf("ledger from corrupt snapshot = %+v, want empty", got) + } +} + +func TestOwedLedgerCapacityRejectsNewDebt(t *testing.T) { + logger := stdlog.New(&bytes.Buffer{}, "", 0) + l := newOwedLedger(t.TempDir(), "cap-test", logger) + + before := testutil.ToFloat64(owedRejected.WithLabelValues("cap-test")) + for i := 0; i < maxOwedEntries; i++ { + l.entries[uploadKey{Hash: strconv.Itoa(i)}] = owedEntry{} + } + + newKey := uploadKey{Kind: cache.CAS, Hash: testHash} + l.add(owedEntry{Key: newKey}) + if _, ok := l.entries[newKey]; ok { + t.Fatal("full ledger accepted new debt, want rejection") + } + if got := testutil.ToFloat64(owedRejected.WithLabelValues("cap-test")) - before; got != 1 { + t.Fatalf("owed rejected counter delta = %v, want 1", got) + } + + // Updating an EXISTING key must still succeed at capacity. + var existing uploadKey + for k := range l.entries { + existing = k + break + } + l.add(owedEntry{Key: existing, LogicalSize: 99}) + if l.entries[existing].LogicalSize != 99 { + t.Fatal("full ledger refused update of existing debt, want acceptance") + } +} diff --git a/cache/s3proxy/s3proxy.go b/cache/s3proxy/s3proxy.go index bd07b4c..e47eca2 100644 --- a/cache/s3proxy/s3proxy.go +++ b/cache/s3proxy/s3proxy.go @@ -66,6 +66,14 @@ type s3Cache struct { readDeadline time.Duration objectKey func(prefix string, hash string, kind cache.EntryKind) string observer cache.OperationObserver + // inflight coalesces duplicate enqueues; owed records shed/failed + // uploads for the background sweeper; blobSource reopens local blobs + // for deferred uploads. See owed.go. + inflight *inflightSet + owed *owedLedger + owedDir string + blobSource BlobSource + blobSourceOnce sync.Once } type Option func(*s3Cache) @@ -76,6 +84,16 @@ func WithOperationObserver(observer cache.OperationObserver) Option { } } +// WithOwedLedgerDir enables the owed-upload ledger (see owed.go), persisting +// per-backend snapshots under dir. Without this option, shed uploads are +// dropped exactly as before — embedders that don't own durable storage (the +// host-side fallback proxy) keep the old best-effort semantics. +func WithOwedLedgerDir(dir string) Option { + return func(c *s3Cache) { + c.owedDir = dir + } +} + var ( cacheHits = promauto.NewCounterVec(prometheus.CounterOpts{ Name: "bazel_remote_s3_cache_hits", @@ -399,6 +417,11 @@ func newBackend(spec BackendSpec, updateTimestamps bool, connRecycleInterval tim c.objectKey = objectKeyV1 } + c.inflight = newInflightSet() + if c.owedDir != "" { + c.owed = newOwedLedger(c.owedDir, key, errorLogger) + } + c.uploadQueue = backendproxy.StartUploaders(c, numUploaders, maxQueuedUploads) return c, nil @@ -596,6 +619,19 @@ func (c *s3Cache) UploadFile(item backendproxy.UploadReq) { } uploadOutcomes.WithLabelValues(c.key, status, reason).Inc() c.observeUpload(context.Background(), item, status, reason) + + // Settle or record the owed-ledger debt for this upload identity. Any + // success (created, or already_exists — someone else stored it) clears + // the debt; any terminal failure (failed PUT, breaker refusal) records + // it so the sweeper retries once the backend is healthy again. + identity := c.resolveUploadIdentity(item) + if status == "error" { + c.owed.add(c.owedEntryForItem(identity, item)) + } else { + c.owed.settle(identity) + } + c.inflight.remove(identity) + uploadQueueDepth.WithLabelValues(c.key).Set(float64(len(c.uploadQueue))) _ = item.Rc.Close() @@ -639,8 +675,7 @@ func (c *s3Cache) Put(ctx context.Context, kind cache.EntryKind, hash string, lo // decision — but the pair travels together.) selection, _ := cache.S3BackendFromContext(ctx) - select { - case c.uploadQueue <- backendproxy.UploadReq{ + item := backendproxy.UploadReq{ Hash: hash, LogicalSize: logicalSize, SizeOnDisk: sizeOnDisk, @@ -651,11 +686,30 @@ func (c *s3Cache) Put(ctx context.Context, kind cache.EntryKind, hash string, lo RequireStoragePrefix: requirePrefix, S3Backend: selection, MetricsLabels: labels, - }: + } + + // Coalesce duplicate uploads: matrix fan-out delivers the same missing + // blob from many hosts at once, MinIO's create-if-absent would 412 all + // but one anyway, and every queued duplicate pins an open FD. The + // in-flight copy's terminal outcome settles any owed debt for this key. + key := c.resolveUploadIdentity(item) + if !c.inflight.tryAdd(key) { + uploadQueueCoalesced.WithLabelValues(c.key).Inc() + _ = rc.Close() + return + } + + select { + case c.uploadQueue <- item: uploadQueueDepth.WithLabelValues(c.key).Set(float64(len(c.uploadQueue))) default: + c.inflight.remove(key) c.errorLogger.Printf("too many uploads queued for S3 backend %s\n", c.key) uploadQueueDropped.WithLabelValues(c.key).Inc() + // Shedding is a deferral, not a loss: FindMissingBlobs answers + // local-first, so nothing would ever re-upload this blob. Record + // the debt; the sweeper repays it when the queue has headroom. + c.owed.add(c.owedEntryForItem(key, item)) cache.ObserveOperation(ctx, c.observer, cache.OperationOutcome{ Method: "backend_upload", Kind: kind.String(), diff --git a/config/proxy.go b/config/proxy.go index d532ffb..4bdf23d 100644 --- a/config/proxy.go +++ b/config/proxy.go @@ -8,6 +8,7 @@ import ( "fmt" "net/http" "os" + "path/filepath" "syscall" "github.com/buchgr/bazel-remote/v2/cache/azblobproxy" @@ -187,7 +188,12 @@ func (c *Config) setProxy() error { c.S3CloudStorage.ConnRecycleInterval, c.StorageMode, c.AccessLogger, c.ErrorLogger, numUploaders, maxQueuedUploads, s3proxy.PrometheusMetrics(), - s3proxy.WithReadDeadline(c.S3CloudStorage.ReadTimeout)) + s3proxy.WithReadDeadline(c.S3CloudStorage.ReadTimeout), + // Standalone deployments own durable local storage, so shed + // or failed write-throughs become owed uploads (snapshotted + // beside the cache) instead of silent losses. See + // s3proxy/owed.go for the invariant this preserves. + s3proxy.WithOwedLedgerDir(filepath.Join(c.Dir, "s3-owed"))) if err != nil { return err } @@ -220,7 +226,8 @@ func (c *Config) setProxy() error { c.S3CloudStorage.ConnRecycleInterval, c.StorageMode, c.AccessLogger, c.ErrorLogger, c.NumUploaders, c.MaxQueuedUploads, s3proxy.PrometheusMetrics(), - s3proxy.WithReadDeadline(c.S3CloudStorage.ReadTimeout)) + s3proxy.WithReadDeadline(c.S3CloudStorage.ReadTimeout), + s3proxy.WithOwedLedgerDir(filepath.Join(c.Dir, "s3-owed"))) return nil } diff --git a/main.go b/main.go index e8d5166..53b681f 100644 --- a/main.go +++ b/main.go @@ -16,6 +16,7 @@ import ( auth "github.com/abbot/go-http-auth" "github.com/buchgr/bazel-remote/v2/cache/disk" + "github.com/buchgr/bazel-remote/v2/cache/s3proxy" "github.com/buchgr/bazel-remote/v2/config" "github.com/buchgr/bazel-remote/v2/ldap" @@ -177,6 +178,16 @@ func run(ctx *cli.Context) error { } diskCache.RegisterMetrics() + // The owed-upload sweeper (s3proxy/owed.go) reopens local blobs for + // deferred backend uploads. The disk cache is constructed after the + // proxy (it takes the proxy as an option), so the back-reference is + // injected here, which also starts the per-backend sweepers. + if setter, ok := c.ProxyBackend.(s3proxy.BlobSourceSetter); ok { + if source, ok := diskCache.(s3proxy.BlobSource); ok { + setter.SetBlobSource(source) + } + } + servers := new(errgroup.Group) var htpasswdSecrets auth.SecretProvider From efe6c7323154541ce849ce0957b8c530088cd8ae Mon Sep 17 00:00:00 2001 From: Shreyas Kalyan Date: Sun, 2 Aug 2026 22:04:53 -0400 Subject: [PATCH 2/5] =?UTF-8?q?s3proxy:=20owed-ledger=20review=20fixes=20?= =?UTF-8?q?=E2=80=94=20boot-crash=20on=20ledger=20dir,=20void-settle=20rac?= =?UTF-8?q?e?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Self-review findings on the owed-upload piece: - P0: the ledger dir lived inside the cache root, and the disk cache's startup scan hard-fails on unexpected top-level dirs — every standalone boot with an S3 proxy crashed with 'unexpected dir: s3-owed'. The dir name is now a shared cache.OwedLedgerDirName constant and both scan paths skip it like lost+found; regression test constructs the disk cache over a populated root. - P2: the sweeper's evicted-blob void-settle could race a concurrent client re-Put whose upload coalesced against the sweeper's own in-flight claim, erasing the only record that MinIO lacks the object. Owed entries carry a per-add seq; coalesced Puts re-record their debt, and void settlement is seq-conditional (settleVoid) so a stale observation no-ops. Upload-success settlement stays unconditional (proof the object exists discharges any seq). - P2: OpenOwedBlob no longer holds the global disk mutex across os.Open (a sweep batch is hundreds of opens; losing the eviction race surfaces as the open error callers already treat as void). - P3: snapshot filenames append a hash of the raw backend key (sanitization is lossy; colliding keys would clobber snapshots). - P3: a failed snapshot write re-dirties the ledger so the next pass retries instead of stretching the 15s staleness window forever. Co-authored-by: Cursor --- cache/cache.go | 8 ++++ cache/disk/load.go | 4 +- cache/disk/owed.go | 7 +++- cache/disk/owed_test.go | 30 +++++++++++++++ cache/s3proxy/owed.go | 76 +++++++++++++++++++++++++++++++++----- cache/s3proxy/owed_test.go | 42 +++++++++++++++++++-- cache/s3proxy/s3proxy.go | 8 +++- config/proxy.go | 10 +++-- 8 files changed, 164 insertions(+), 21 deletions(-) diff --git a/cache/cache.go b/cache/cache.go index 7a8a007..fd62545 100644 --- a/cache/cache.go +++ b/cache/cache.go @@ -104,6 +104,14 @@ func TransformActionCacheKey(key, instance string, logger Logger) string { return newKey } +// OwedLedgerDirName is the directory (under the disk cache root) holding +// owed-upload ledger snapshots (see s3proxy/owed.go). Defined here because +// two packages that must not import each other need to agree on it: the +// proxy config wires it as the snapshot location, and the disk cache's +// startup scan must skip it like lost+found rather than fail the boot on +// an unexpected directory. +const OwedLedgerDirName = "s3-owed" + func LookupKey(kind EntryKind, hash string) string { return kind.String() + "/" + hash } diff --git a/cache/disk/load.go b/cache/disk/load.go index 30c155b..2fefe99 100644 --- a/cache/disk/load.go +++ b/cache/disk/load.go @@ -532,7 +532,7 @@ func (c *diskCache) scanDir() (scanResult, error) { if name == lostAndFound { continue } - if rootRel == "" && storagePrefixDRE.MatchString(name) { + if rootRel == "" && (name == cache.OwedLedgerDirName || storagePrefixDRE.MatchString(name)) { continue } @@ -600,7 +600,7 @@ func (c *diskCache) scanDir() (scanResult, error) { return scanResult{}, fmt.Errorf("unexpected file: %s", name) } - if name == lostAndFound { + if name == lostAndFound || name == cache.OwedLedgerDirName { continue } diff --git a/cache/disk/owed.go b/cache/disk/owed.go index 2bcaf9c..ac8d7b3 100644 --- a/cache/disk/owed.go +++ b/cache/disk/owed.go @@ -32,9 +32,14 @@ func (c *diskCache) OpenOwedBlob(ctx context.Context, kind cache.EntryKind, hash return nil, -1, errOwedBlobUnavailable } blobPath := path.Join(c.dir, c.FileLocationForContext(ctx, kind, item.legacy, hash, item.size, item.random)) - f, err := os.Open(blobPath) c.mu.Unlock() + // Open outside the global cache mutex, like the rest of this package: + // c.mu serializes every Get/Put/eviction on the node, and a sweep batch + // is hundreds of opens that may block on disk I/O. Losing the race with + // eviction surfaces as an open error, which callers already treat as + // debt-settled-void. + f, err := os.Open(blobPath) if err != nil { return nil, -1, errOwedBlobUnavailable } diff --git a/cache/disk/owed_test.go b/cache/disk/owed_test.go index 69c168a..a217278 100644 --- a/cache/disk/owed_test.go +++ b/cache/disk/owed_test.go @@ -65,6 +65,36 @@ func TestOpenOwedBlobReturnsRawOnDiskBytes(t *testing.T) { } } +// TestDiskCacheLoadsWithOwedLedgerDirPresent pins the wiring contract: the +// proxy config creates /s3-owed BEFORE disk.New scans the cache +// root (setProxy runs during config load), and the startup scan must skip it +// like lost+found instead of failing the boot with "unexpected dir". +func TestDiskCacheLoadsWithOwedLedgerDirPresent(t *testing.T) { + cacheDir := tempDir(t) + defer func() { _ = os.RemoveAll(cacheDir) }() + + if err := os.MkdirAll(cacheDir+"/"+cache.OwedLedgerDirName, 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(cacheDir+"/"+cache.OwedLedgerDirName+"/owed-uploads-test-deadbeef.json", []byte("[]"), 0o644); err != nil { + t.Fatal(err) + } + + testCacheI, err := New(cacheDir, 1024*1024, WithAccessLogger(testutils.NewSilentLogger())) + if err != nil { + t.Fatalf("disk.New with %s present: %v", cache.OwedLedgerDirName, err) + } + + // And again with content in the cache, exercising the populated-scan path. + data, hash := testutils.RandomDataAndHash(64) + if err := testCacheI.Put(context.Background(), cache.CAS, hash, 64, io.NopCloser(bytes.NewReader(data))); err != nil { + t.Fatal(err) + } + if _, err := New(cacheDir, 1024*1024, WithAccessLogger(testutils.NewSilentLogger())); err != nil { + t.Fatalf("disk.New rescan with %s present: %v", cache.OwedLedgerDirName, err) + } +} + func TestOpenOwedBlobIsStoragePrefixScoped(t *testing.T) { cacheDir := tempDir(t) defer func() { _ = os.RemoveAll(cacheDir) }() diff --git a/cache/s3proxy/owed.go b/cache/s3proxy/owed.go index e890281..b01053a 100644 --- a/cache/s3proxy/owed.go +++ b/cache/s3proxy/owed.go @@ -27,7 +27,9 @@ package s3proxy import ( "context" + "crypto/sha256" "encoding/json" + "fmt" "io" "os" "path/filepath" @@ -117,6 +119,15 @@ type owedEntry struct { LogicalSize int64 `json:"logical_size"` RequestScopedStoragePrefix bool `json:"request_scoped_prefix"` RequireStoragePrefix bool `json:"require_prefix"` + + // seq is bumped on every add for this key. The sweeper's evicted-blob + // settlement is conditional on it (see settleVoid): between the sweeper + // observing "not on disk" and settling, a client can re-Put the blob — + // whose upload the sweeper's own in-flight claim then coalesces away — + // so an unconditional settle would erase the only record that MinIO + // still lacks the object. Not persisted: within one process lifetime is + // the only window the race exists in. + seq uint64 } // inflightSet tracks uploads that are queued or being uploaded right now. @@ -163,16 +174,23 @@ type owedLedger struct { mu sync.Mutex entries map[uploadKey]owedEntry dirty bool + nextSeq uint64 } // newOwedLedger loads any snapshot left by a previous process. A corrupt or // unreadable snapshot starts empty and logs: the ledger is a convergence // accelerator, never a correctness gate worth failing startup for. func newOwedLedger(dir, backendKey string, errorLogger cache.Logger) *owedLedger { + // The filename carries a hash of the RAW key alongside the sanitized + // form: sanitization is lossy (everything outside [a-zA-Z0-9.-] maps + // to '_'), and two backend keys colliding onto one snapshot file would + // clobber each other's debts and load the union on restart. + keyDigest := sha256.Sum256([]byte(backendKey)) l := &owedLedger{ backendKey: backendKey, - path: filepath.Join(dir, "owed-uploads-"+sanitizeForFilename(backendKey)+".json"), - entries: make(map[uploadKey]owedEntry), + path: filepath.Join(dir, fmt.Sprintf("owed-uploads-%s-%x.json", + sanitizeForFilename(backendKey), keyDigest[:4])), + entries: make(map[uploadKey]owedEntry), } if err := os.MkdirAll(dir, 0o755); err != nil { errorLogger.Printf("owed ledger: cannot create %s (%v); continuing without restart persistence", dir, err) @@ -207,11 +225,16 @@ func (l *owedLedger) add(e owedEntry) { owedRejected.WithLabelValues(l.backendKey).Inc() return } + l.nextSeq++ + e.seq = l.nextSeq l.entries[e.Key] = e l.dirty = true owedBacklog.WithLabelValues(l.backendKey).Set(float64(len(l.entries))) } +// settle removes the debt unconditionally: callers hold proof that the +// object now exists in the backend (created or already_exists), which +// discharges the debt no matter how many times it was re-recorded. func (l *owedLedger) settle(k uploadKey) { if l == nil { return @@ -226,6 +249,28 @@ func (l *owedLedger) settle(k uploadKey) { owedBacklog.WithLabelValues(l.backendKey).Set(float64(len(l.entries))) } +// settleVoid removes the debt only if it has not been re-recorded since the +// caller observed it (seq match). Used by the sweeper's evicted-blob path, +// where the proof is "the blob is gone locally" — a fact that a concurrent +// client re-Put (coalesced away by the sweeper's own in-flight claim) can +// invalidate between observation and settlement. A bumped seq means the +// debt is live again; keep it for the next pass. Returns whether it settled. +func (l *owedLedger) settleVoid(k uploadKey, seq uint64) bool { + if l == nil { + return false + } + l.mu.Lock() + defer l.mu.Unlock() + e, ok := l.entries[k] + if !ok || e.seq != seq { + return false + } + delete(l.entries, k) + l.dirty = true + owedBacklog.WithLabelValues(l.backendKey).Set(float64(len(l.entries))) + return true +} + // batch returns up to n entries for a sweep pass, in map order (unordered — // fairness across passes comes from settled entries leaving the map). func (l *owedLedger) batch(n int) []owedEntry { @@ -269,12 +314,18 @@ func (l *owedLedger) snapshotIfDirty(errorLogger cache.Logger) { return } tmp := l.path + ".tmp" - if err := os.WriteFile(tmp, data, 0o644); err != nil { - errorLogger.Printf("owed ledger: snapshot write failed: %v", err) - return + err = os.WriteFile(tmp, data, 0o644) + if err == nil { + err = os.Rename(tmp, l.path) } - if err := os.Rename(tmp, l.path); err != nil { - errorLogger.Printf("owed ledger: snapshot rename failed: %v", err) + if err != nil { + errorLogger.Printf("owed ledger: snapshot failed: %v", err) + // Re-dirty so the next pass retries: without this, a ledger that + // goes quiet after a transient write error would never snapshot + // again, stretching the documented ≤15s staleness indefinitely. + l.mu.Lock() + l.dirty = true + l.mu.Unlock() } } @@ -375,10 +426,15 @@ func (c *s3Cache) sweepOwedOnce() { if err != nil { // Evicted (or unreadable): the blob is honestly absent // everywhere now, so FindMissingBlobs will report it missing - // and a future client upload recreates both copies. Debt void. + // and a future client upload recreates both copies. Debt void — + // but only if it wasn't re-recorded since this batch was taken + // (a concurrent re-Put coalesced against OUR in-flight claim + // re-adds the debt; see settleVoid). Release the claim first so + // a re-add racing this line coalesces at most once. + if c.owed.settleVoid(entry.Key, entry.seq) { + owedSweep.WithLabelValues(c.key, "blob_evicted").Inc() + } c.inflight.remove(entry.Key) - c.owed.settle(entry.Key) - owedSweep.WithLabelValues(c.key, "blob_evicted").Inc() continue } diff --git a/cache/s3proxy/owed_test.go b/cache/s3proxy/owed_test.go index d5b32d5..e48ddfb 100644 --- a/cache/s3proxy/owed_test.go +++ b/cache/s3proxy/owed_test.go @@ -8,7 +8,6 @@ import ( stdlog "log" "net/http/httptest" "os" - "path/filepath" "strconv" "strings" "testing" @@ -208,12 +207,14 @@ func TestOwedLedgerSnapshotRoundtrip(t *testing.T) { reloaded := newOwedLedger(dir, "snap-test", logger) got := reloaded.batch(10) - if len(got) != 1 || got[0] != entry { + if len(got) != 1 || got[0].Key != entry.Key || got[0].LogicalSize != entry.LogicalSize || + got[0].RequestScopedStoragePrefix != entry.RequestScopedStoragePrefix || + got[0].RequireStoragePrefix != entry.RequireStoragePrefix { t.Fatalf("reloaded ledger = %+v, want [%+v]", got, entry) } // A corrupt snapshot must start empty without failing. - if err := os.WriteFile(filepath.Join(dir, "owed-uploads-snap-test.json"), []byte("{not json"), 0o644); err != nil { + if err := os.WriteFile(reloaded.path, []byte("{not json"), 0o644); err != nil { t.Fatal(err) } corrupt := newOwedLedger(dir, "snap-test", logger) @@ -222,6 +223,41 @@ func TestOwedLedgerSnapshotRoundtrip(t *testing.T) { } } +// TestVoidSettleYieldsToConcurrentReAdd pins the coalesce/void race: the +// sweeper batches an entry, finds the blob evicted, and — before it settles — +// a client re-Puts the blob, whose upload coalesces against the sweeper's own +// in-flight claim and re-records the debt. The stale void-settle must no-op. +func TestVoidSettleYieldsToConcurrentReAdd(t *testing.T) { + c, queue := owedTestCache(t, "owed-void-race-test", 8) + key := uploadKey{Kind: cache.CAS, Hash: testHash, Bucket: "test-bucket"} + c.owed.add(owedEntry{Key: key}) + + batched := c.owed.batch(1)[0] + + // The re-Put arrives between batch and settle: coalesced (the sweeper + // would hold the in-flight claim), debt re-recorded with a bumped seq. + if !c.inflight.tryAdd(key) { + t.Fatal("test setup: in-flight claim failed") + } + c.Put(context.Background(), cache.CAS, testHash, 4, 4, io.NopCloser(strings.NewReader("blob"))) + if got := len(queue); got != 0 { + t.Fatalf("re-Put was not coalesced: queue depth %d, want 0", got) + } + + if c.owed.settleVoid(key, batched.seq) { + t.Fatal("stale void-settle succeeded, want no-op after re-add") + } + if got := testutil.ToFloat64(owedBacklog.WithLabelValues(c.key)); got != 1 { + t.Fatalf("owed backlog after raced void-settle = %v, want 1 (debt live)", got) + } + + // With no interleaving re-add, void settlement works. + fresh := c.owed.batch(1)[0] + if !c.owed.settleVoid(key, fresh.seq) { + t.Fatal("clean void-settle failed, want success") + } +} + func TestOwedLedgerCapacityRejectsNewDebt(t *testing.T) { logger := stdlog.New(&bytes.Buffer{}, "", 0) l := newOwedLedger(t.TempDir(), "cap-test", logger) diff --git a/cache/s3proxy/s3proxy.go b/cache/s3proxy/s3proxy.go index e47eca2..b96cba2 100644 --- a/cache/s3proxy/s3proxy.go +++ b/cache/s3proxy/s3proxy.go @@ -691,10 +691,16 @@ func (c *s3Cache) Put(ctx context.Context, kind cache.EntryKind, hash string, lo // Coalesce duplicate uploads: matrix fan-out delivers the same missing // blob from many hosts at once, MinIO's create-if-absent would 412 all // but one anyway, and every queued duplicate pins an open FD. The - // in-flight copy's terminal outcome settles any owed debt for this key. + // coalesced-away upload still records its debt: usually the in-flight + // copy's success settles it moments later, but the in-flight claim can + // also be the SWEEPER holding a blob it just found evicted — about to + // settle the debt as void while this Put proves the blob is back. The + // seq bump from this add makes that void-settle a no-op (settleVoid), + // and the next sweep pass repays it for real. key := c.resolveUploadIdentity(item) if !c.inflight.tryAdd(key) { uploadQueueCoalesced.WithLabelValues(c.key).Inc() + c.owed.add(c.owedEntryForItem(key, item)) _ = rc.Close() return } diff --git a/config/proxy.go b/config/proxy.go index 4bdf23d..54aed74 100644 --- a/config/proxy.go +++ b/config/proxy.go @@ -11,6 +11,7 @@ import ( "path/filepath" "syscall" + "github.com/buchgr/bazel-remote/v2/cache" "github.com/buchgr/bazel-remote/v2/cache/azblobproxy" "github.com/buchgr/bazel-remote/v2/cache/gcsproxy" "github.com/buchgr/bazel-remote/v2/cache/grpcproxy" @@ -191,9 +192,10 @@ func (c *Config) setProxy() error { s3proxy.WithReadDeadline(c.S3CloudStorage.ReadTimeout), // Standalone deployments own durable local storage, so shed // or failed write-throughs become owed uploads (snapshotted - // beside the cache) instead of silent losses. See - // s3proxy/owed.go for the invariant this preserves. - s3proxy.WithOwedLedgerDir(filepath.Join(c.Dir, "s3-owed"))) + // under the cache dir — the disk scan skips this entry by + // name) instead of silent losses. See s3proxy/owed.go for + // the invariant this preserves. + s3proxy.WithOwedLedgerDir(filepath.Join(c.Dir, cache.OwedLedgerDirName))) if err != nil { return err } @@ -227,7 +229,7 @@ func (c *Config) setProxy() error { c.StorageMode, c.AccessLogger, c.ErrorLogger, c.NumUploaders, c.MaxQueuedUploads, s3proxy.PrometheusMetrics(), s3proxy.WithReadDeadline(c.S3CloudStorage.ReadTimeout), - s3proxy.WithOwedLedgerDir(filepath.Join(c.Dir, "s3-owed"))) + s3proxy.WithOwedLedgerDir(filepath.Join(c.Dir, cache.OwedLedgerDirName))) return nil } From cb7ad518e28f20ecbe54219bbbcb811d63cddb2e Mon Sep 17 00:00:00 2001 From: Shreyas Kalyan Date: Tue, 4 Aug 2026 08:43:34 -0400 Subject: [PATCH 3/5] s3proxy: sweeper probes the breaker so idle nodes repay owed uploads MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Breaker recovery relied on read-path probes ("calls arrive constantly"), but an idle node has none: after a MinIO outage the breaker stayed open forever, sweepOwedOnce returned early at its breaker gate, and the owed ledger never drained — recovery of the subset invariant was traffic-gated (observed live on staging-2, 2026-08-04 chaos drill: 12 debts frozen for 200s+ post-recovery until an unrelated read probed the breaker closed). When the sweeper holds debts and finds the breaker non-closed it now volunteers a bucket-existence check as the half-open recovery probe, bounded by sweepProbeTimeout. allow() still enforces the open-window cooldown, so this costs at most one cheap RPC per sweep pass while the backend stays dark, and the ledger drains on the pass after recovery. Co-authored-by: Cursor --- cache/s3proxy/owed.go | 42 +++++++++++++++++++++++++++++ cache/s3proxy/owed_test.go | 55 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 97 insertions(+) diff --git a/cache/s3proxy/owed.go b/cache/s3proxy/owed.go index b01053a..061fcc1 100644 --- a/cache/s3proxy/owed.go +++ b/cache/s3proxy/owed.go @@ -63,6 +63,11 @@ const ( // queue is under this fraction of capacity, so deferred uploads never // compete with current build traffic for queue slots or workers. sweepQueueHeadroom = 0.5 + + // sweepProbeTimeout bounds the sweeper's own breaker probe (a bucket + // existence check). Generous enough for a slow-but-alive backend, + // bounded so a black-holed one costs at most this per sweep pass. + sweepProbeTimeout = 5 * time.Second ) var ( @@ -232,6 +237,17 @@ func (l *owedLedger) add(e owedEntry) { owedBacklog.WithLabelValues(l.backendKey).Set(float64(len(l.entries))) } +// size reports the current number of owed entries. Nil-safe like every +// other method (feature disabled -> permanently zero). +func (l *owedLedger) size() int { + if l == nil { + return 0 + } + l.mu.Lock() + defer l.mu.Unlock() + return len(l.entries) +} + // settle removes the debt unconditionally: callers hold proof that the // object now exists in the backend (created or already_exists), which // discharges the debt no matter how many times it was re-recorded. @@ -387,6 +403,20 @@ func (m *multiS3Cache) SetBlobSource(src BlobSource) { } } +// probeBackendForSweep offers a bucket existence check as the breaker's +// half-open recovery probe. The call is cheap (one metadata RPC), bounded by +// sweepProbeTimeout, and classified by breakerReadOutcome — the backend +// merely answering (even "no such bucket") proves it is reachable again. +// Success closes the breaker; the next sweep pass drains the ledger. +func (c *s3Cache) probeBackendForSweep() { + ctx, cancel := context.WithTimeout(context.Background(), sweepProbeTimeout) + defer cancel() + _ = c.breaker.Execute(func() breakerOutcome { + _, err := c.mcore.BucketExists(ctx, c.bucket) + return breakerReadOutcome(ctx, err) + }) +} + func (c *s3Cache) sweepOwedLoop() { ticker := time.NewTicker(sweepInterval) defer ticker.Stop() @@ -404,6 +434,18 @@ func (c *s3Cache) sweepOwedOnce() { return } if c.breaker != nil && c.breaker.State() != breakerClosed { + // The breaker normally recovers through read-path probes (see + // ExecuteNoProbe), but that makes ledger convergence traffic-gated: + // an idle node has no reads, so after an outage the breaker stays + // open forever and the debts never drain (observed on staging, + // 2026-08-04 chaos drill). When we hold debts, volunteer a cheap + // bounded call as the recovery probe. allow() still enforces the + // open-window cooldown, so this is a no-op until the window + // elapses and costs at most one RPC per sweep pass while the + // backend stays dark. + if c.owed.size() > 0 { + c.probeBackendForSweep() + } return } if len(c.uploadQueue) >= int(sweepQueueHeadroom*float64(cap(c.uploadQueue))) { diff --git a/cache/s3proxy/owed_test.go b/cache/s3proxy/owed_test.go index e48ddfb..cf10081 100644 --- a/cache/s3proxy/owed_test.go +++ b/cache/s3proxy/owed_test.go @@ -11,6 +11,7 @@ import ( "strconv" "strings" "testing" + "time" "github.com/buchgr/bazel-remote/v2/cache" "github.com/buchgr/bazel-remote/v2/utils/backendproxy" @@ -156,6 +157,60 @@ func TestUploadFailureRecordsOwedAndBreakerBlocksSweep(t *testing.T) { } } +// TestSweepProbesBreakerWhenIdle pins the idle-node recovery contract: with +// debts on the ledger, an open breaker past its cooldown, and NO read +// traffic to volunteer as the half-open probe, the sweeper itself must +// probe the backend and close the breaker, then drain the ledger on the +// following pass. Without the sweeper probe this deadlocks — the breaker +// waits for traffic, the sweeper waits for the breaker — and an idle node +// never repays its debts (observed on staging, 2026-08-04 chaos drill). +func TestSweepProbesBreakerWhenIdle(t *testing.T) { + backend := s3mem.New() + if err := backend.CreateBucket("test-bucket"); err != nil { + t.Fatal(err) + } + ts := httptest.NewServer(gofakes3.New(backend).Server()) + t.Cleanup(ts.Close) + + c, queue := owedTestCache(t, "owed-idle-probe-test", 4) + c.mcore = coreFor(t, ts, 1) + + content := []byte("idle-owed-blob") + src := &fakeBlobSource{blobs: map[string][]byte{testHash: content}} + c.blobSource = src + + c.owed.add(owedEntry{Key: uploadKey{Kind: cache.CAS, Hash: testHash}}) + tripBreaker(t, c.breaker) + + // Inside the cooldown window the probe must be refused: the pass is a + // complete no-op (no blob source reads, nothing queued, still open). + c.sweepOwedOnce() + if c.breaker.State() != breakerOpen || src.calls != 0 || len(queue) != 0 { + t.Fatalf("sweep inside cooldown: state=%v srcCalls=%d queued=%d, want open/0/0", + c.breaker.State(), src.calls, len(queue)) + } + + // Past the cooldown the sweeper's probe is admitted as the half-open + // probe; the healthy fake closes the breaker. Recovery deliberately + // takes one extra pass: this one only probes. + c.breaker.mu.Lock() + c.breaker.openedAt = time.Now().Add(-2 * breakerTimeout) + c.breaker.mu.Unlock() + c.sweepOwedOnce() + if got := c.breaker.State(); got != breakerClosed { + t.Fatalf("breaker state after idle sweep probe = %v, want closed", got) + } + if src.calls != 0 || len(queue) != 0 { + t.Fatalf("probe pass touched work: srcCalls=%d queued=%d, want 0/0", src.calls, len(queue)) + } + + // Next pass drains the ledger through the normal requeue path. + c.sweepOwedOnce() + if got := len(queue); got != 1 { + t.Fatalf("queue depth after post-recovery sweep = %d, want 1", got) + } +} + func TestSweepEvictedBlobSettlesDebt(t *testing.T) { c, queue := owedTestCache(t, "owed-evicted-test", 4) c.owed.add(owedEntry{Key: uploadKey{Kind: cache.CAS, Hash: testHash, Bucket: "test-bucket"}}) From 54ebcc6b13986e7a409bc03ad30649f373c7abb1 Mon Sep 17 00:00:00 2001 From: Shreyas Kalyan Date: Tue, 4 Aug 2026 08:55:55 -0400 Subject: [PATCH 4/5] s3proxy: probe breaker recovery via sentinel StatObject, not BucketExists MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit BucketExists issues HEAD-bucket, which prefix-scoped MinIO policies reject with 403 — a permanent probe failure that re-opened the breaker every sweep pass on staging-2 (6 half-open -> open flaps in 3 minutes) while the ledger stayed frozen. StatObject on a sentinel key is the read path's exact call shape, so it needs only permissions the deployed credentials demonstrably hold, and its NoSuchKey answer already counts as success in breakerReadOutcome. Probe failures now log the underlying error so a policy-shaped failure is visible in the journal. Co-authored-by: Cursor --- cache/s3proxy/owed.go | 27 ++++++++++++++++++++------- 1 file changed, 20 insertions(+), 7 deletions(-) diff --git a/cache/s3proxy/owed.go b/cache/s3proxy/owed.go index 061fcc1..215f7cb 100644 --- a/cache/s3proxy/owed.go +++ b/cache/s3proxy/owed.go @@ -40,6 +40,8 @@ import ( "github.com/buchgr/bazel-remote/v2/cache" "github.com/buchgr/bazel-remote/v2/utils/backendproxy" + "github.com/minio/minio-go/v7" + "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/promauto" ) @@ -403,17 +405,28 @@ func (m *multiS3Cache) SetBlobSource(src BlobSource) { } } -// probeBackendForSweep offers a bucket existence check as the breaker's -// half-open recovery probe. The call is cheap (one metadata RPC), bounded by -// sweepProbeTimeout, and classified by breakerReadOutcome — the backend -// merely answering (even "no such bucket") proves it is reachable again. -// Success closes the breaker; the next sweep pass drains the ledger. +// probeBackendForSweep offers a sentinel StatObject as the breaker's +// half-open recovery probe: the exact call shape of the read path +// (Contains), so it exercises only permissions the deployed credentials +// are known to hold — NOT BucketExists/HEAD-bucket, which prefix-scoped +// MinIO policies reject with 403, a permanent probe failure that would +// re-open the breaker forever (observed on staging-2, 2026-08-04). The +// call is cheap, bounded by sweepProbeTimeout, and classified by +// breakerReadOutcome: the expected NoSuchKey is a success (the backend +// answered). Success closes the breaker; the next sweep pass drains the +// ledger. Failures are logged — a probe that fails for a non-connectivity +// reason (credentials, policy) is otherwise invisible. func (c *s3Cache) probeBackendForSweep() { ctx, cancel := context.WithTimeout(context.Background(), sweepProbeTimeout) defer cancel() _ = c.breaker.Execute(func() breakerOutcome { - _, err := c.mcore.BucketExists(ctx, c.bucket) - return breakerReadOutcome(ctx, err) + _, err := c.mcore.StatObject(ctx, c.bucket, "sweep-breaker-probe-sentinel", + minio.StatObjectOptions{}) + outcome := breakerReadOutcome(ctx, err) + if outcome == outcomeFailure && c.errorLogger != nil { + c.errorLogger.Printf("owed sweeper: breaker recovery probe failed: %v", err) + } + return outcome }) } From a3ccdcb53b12116710bcce04ddd7361e37053d43 Mon Sep 17 00:00:00 2001 From: Shreyas Kalyan Date: Tue, 4 Aug 2026 09:17:32 -0400 Subject: [PATCH 5/5] s3proxy: probe breaker recovery against a real owed key MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A synthetic sentinel key fails the same way HEAD-bucket did: prefix-scoped MinIO policies answer Access Denied for any path outside the tenant prefixes this node serves, so the probe re-opened the breaker every sweep pass (staging-2, 2026-08-04, "probe failed: Access Denied"). Stat the object key of an owed entry instead — by construction a path these credentials can write and one the Contains path HEADs routinely — so the probe only fails when the backend is genuinely sick. Co-authored-by: Cursor --- cache/s3proxy/owed.go | 42 ++++++++++++++++++++++++++++-------------- 1 file changed, 28 insertions(+), 14 deletions(-) diff --git a/cache/s3proxy/owed.go b/cache/s3proxy/owed.go index 215f7cb..7ae23c5 100644 --- a/cache/s3proxy/owed.go +++ b/cache/s3proxy/owed.go @@ -405,26 +405,40 @@ func (m *multiS3Cache) SetBlobSource(src BlobSource) { } } -// probeBackendForSweep offers a sentinel StatObject as the breaker's -// half-open recovery probe: the exact call shape of the read path -// (Contains), so it exercises only permissions the deployed credentials -// are known to hold — NOT BucketExists/HEAD-bucket, which prefix-scoped -// MinIO policies reject with 403, a permanent probe failure that would -// re-open the breaker forever (observed on staging-2, 2026-08-04). The -// call is cheap, bounded by sweepProbeTimeout, and classified by -// breakerReadOutcome: the expected NoSuchKey is a success (the backend -// answered). Success closes the breaker; the next sweep pass drains the -// ledger. Failures are logged — a probe that fails for a non-connectivity -// reason (credentials, policy) is otherwise invisible. +// probeBackendForSweep offers a StatObject on a REAL owed entry's object +// key as the breaker's half-open recovery probe. The key choice is the +// whole point: prefix-scoped MinIO policies 403 anything outside the +// tenant paths this node actually serves — BucketExists (HEAD-bucket) and +// a synthetic sentinel key both probe-failed forever with Access Denied on +// staging-2 (2026-08-04), re-opening the breaker every sweep pass while +// the ledger sat frozen. An owed entry's key is by construction a path +// these credentials can write, and the read path (Contains) HEADs such +// keys routinely, so the only way this probe fails is genuine backend +// sickness. It is cheap, bounded by sweepProbeTimeout, and classified by +// breakerReadOutcome: NoSuchKey (not yet repaid) and found (someone else +// uploaded it) are both "the backend answered" successes. Success closes +// the breaker; the next sweep pass drains the ledger. Failures are logged +// — a probe failing for a non-connectivity reason (credentials, policy) +// is otherwise invisible. func (c *s3Cache) probeBackendForSweep() { + entries := c.owed.batch(1) + if len(entries) == 0 { + return + } + entry := entries[0] + bucket := entry.Key.Bucket + if bucket == "" { + bucket = c.bucket + } + objectKey := c.objectKeyForPrefix(entry.Key.Prefix, entry.Key.Hash, entry.Key.Kind) ctx, cancel := context.WithTimeout(context.Background(), sweepProbeTimeout) defer cancel() _ = c.breaker.Execute(func() breakerOutcome { - _, err := c.mcore.StatObject(ctx, c.bucket, "sweep-breaker-probe-sentinel", - minio.StatObjectOptions{}) + _, err := c.mcore.StatObject(ctx, bucket, objectKey, minio.StatObjectOptions{}) outcome := breakerReadOutcome(ctx, err) if outcome == outcomeFailure && c.errorLogger != nil { - c.errorLogger.Printf("owed sweeper: breaker recovery probe failed: %v", err) + c.errorLogger.Printf("owed sweeper: breaker recovery probe (HEAD %s/%s) failed: %v", + bucket, objectKey, err) } return outcome })