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 new file mode 100644 index 0000000..ac8d7b3 --- /dev/null +++ b/cache/disk/owed.go @@ -0,0 +1,47 @@ +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)) + 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 + } + 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..a217278 --- /dev/null +++ b/cache/disk/owed_test.go @@ -0,0 +1,125 @@ +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") + } +} + +// 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) }() + + 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..7ae23c5 --- /dev/null +++ b/cache/s3proxy/owed.go @@ -0,0 +1,532 @@ +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" + "crypto/sha256" + "encoding/json" + "fmt" + "io" + "os" + "path/filepath" + "strings" + "sync" + "time" + + "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" +) + +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 + + // 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 ( + 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"` + + // 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. +// 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 + 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, 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) + 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.nextSeq++ + e.seq = l.nextSeq + l.entries[e.Key] = e + l.dirty = true + 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. +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))) +} + +// 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 { + 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" + err = os.WriteFile(tmp, data, 0o644) + if err == nil { + err = os.Rename(tmp, l.path) + } + 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() + } +} + +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) + } +} + +// 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, bucket, objectKey, minio.StatObjectOptions{}) + outcome := breakerReadOutcome(ctx, err) + if outcome == outcomeFailure && c.errorLogger != nil { + c.errorLogger.Printf("owed sweeper: breaker recovery probe (HEAD %s/%s) failed: %v", + bucket, objectKey, err) + } + return outcome + }) +} + +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 { + // 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))) { + 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 — + // 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) + 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..cf10081 --- /dev/null +++ b/cache/s3proxy/owed_test.go @@ -0,0 +1,344 @@ +package s3proxy + +import ( + "bytes" + "context" + "errors" + "io" + stdlog "log" + "net/http/httptest" + "os" + "strconv" + "strings" + "testing" + "time" + + "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) + } +} + +// 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"}}) + + 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].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(reloaded.path, []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) + } +} + +// 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) + + 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..b96cba2 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,36 @@ 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 + // 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 + } + + 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..54aed74 100644 --- a/config/proxy.go +++ b/config/proxy.go @@ -8,8 +8,10 @@ import ( "fmt" "net/http" "os" + "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" @@ -187,7 +189,13 @@ 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 + // 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 } @@ -220,7 +228,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, cache.OwedLedgerDirName))) 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