bazel l1 [fork 6]: owed-upload ledger — shedding becomes deferral, not loss

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
 }

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
 		}
 

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
+}

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 <cache dir>/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")
+	}
+}

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 {