fix: check backend directly in blobStore.Put to avoid stale cache

Co-Authored-By: Claude Opus 4.6 <noreply@anthropicxxxxxx.com>
Signed-off-by: luochenhui1 <luochenhui1@kingsoft.com>

registry/storage/blobstore.go

@@ -59,31 +59,34 @@ func (bs *blobStore) Open(ctx context.Context, dgst digest.Digest) (io.ReadSeekC
 
 // Put stores the content p in the blob store, calculating the digest. If the
 // content is already present, only the digest will be returned. This should
-// only be used for small objects, such as manifests. This implemented as a convenience for other Put implementations
+// only be used for small objects, such as manifests.
 func (bs *blobStore) Put(ctx context.Context, mediaType string, p []byte) (v1.Descriptor, error) {
 	dgst := digest.FromBytes(p)
-	desc, err := bs.statter.Stat(ctx, dgst)
-	if err == nil {
-		// content already present
-		return desc, nil
-	} else if err != distribution.ErrBlobUnknown {
-		dcontext.GetLogger(ctx).Errorf("blobStore: error stating content (%v): %v", dgst, err)
-		// real error, return it
-		return v1.Descriptor{}, err
-	}
 
 	bp, err := bs.path(dgst)
 	if err != nil {
 		return v1.Descriptor{}, err
 	}
 
-	// TODO(stevvooe): Write out mediatype here, as well.
-	return v1.Descriptor{
-		Size: int64(len(p)),
+	// Check backend directly, NOT via the cached statter —
+	// the cache may report stale "exists" after GC, causing
+	// Put to skip writing and leading to MANIFEST_UNKNOWN.
+	fi, err := bs.driver.Stat(ctx, bp)
+	if err != nil {
+		if _, ok := err.(driver.PathNotFoundError); !ok {
+			dcontext.GetLogger(ctx).Warnf("blobStore: stat %s before put: %v (writing anyway)", dgst, err)
+		}
+	}
+	if err == nil && !fi.IsDir() {
+		return v1.Descriptor{
+			Size:      fi.Size(),
+			MediaType: "application/octet-stream",
+			Digest:    dgst,
+		}, nil
+	}
 
-		// NOTE(stevvooe): The central blob store firewalls media types from
-		// other users. The caller should look this up and override the value
-		// for the specific repository.
+	return v1.Descriptor{
+		Size:      int64(len(p)),
 		MediaType: "application/octet-stream",
 		Digest:    dgst,
 	}, bs.driver.PutContent(ctx, bp, p)
@@ -133,9 +136,11 @@ func (bs *blobStore) path(dgst digest.Digest) (string, error) {
 // link links the path to the provided digest by writing the digest into the
 // target file. Caller must ensure that the blob actually exists.
 func (bs *blobStore) link(ctx context.Context, path string, dgst digest.Digest) error {
-	// The contents of the "link" file are the exact string contents of the
-	// digest, which is specified in that package.
-	return bs.driver.PutContent(ctx, path, []byte(dgst))
+	err := bs.driver.PutContent(ctx, path, []byte(dgst))
+	if err != nil {
+		dcontext.GetLogger(ctx).Warnf("failed to link blob %s at path %s: %v", dgst, path, err)
+	}
+	return err
 }
 
 // readlink returns the linked digest at path.

registry/storage/blobstore_test.go

@@ -0,0 +1,141 @@
+package storage
+
+import (
+	"context"
+	"testing"
+
+	"github.com/distribution/distribution/v3/registry/storage/cache"
+	"github.com/distribution/distribution/v3/registry/storage/cache/memory"
+	"github.com/distribution/distribution/v3/registry/storage/driver"
+	"github.com/distribution/distribution/v3/registry/storage/driver/inmemory"
+	"github.com/opencontainers/go-digest"
+)
+
+// TestBlobStorePutSkipsExistingContent verifies that Put skips writing when
+// the blob genuinely exists in the backend, even with a cachedBlobStatter.
+func TestBlobStorePutSkipsExistingContent(t *testing.T) {
+	ctx := context.Background()
+	drvr := inmemory.New()
+
+	cacheProvider := memory.NewInMemoryBlobDescriptorCacheProvider(memory.UnlimitedSize)
+	repoCache, err := cacheProvider.RepositoryScoped("test/repo")
+	if err != nil {
+		t.Fatalf("RepositoryScoped: %v", err)
+	}
+
+	rawStatter := &blobStatter{driver: drvr}
+	cachedStatter := cache.NewCachedBlobStatter(repoCache, rawStatter)
+
+	bs := &blobStore{
+		driver:  drvr,
+		statter: cachedStatter,
+	}
+
+	content := []byte("existing-blob-content")
+	dgst := digest.FromBytes(content)
+
+	if _, err := bs.Put(ctx, "application/octet-stream", content); err != nil {
+		t.Fatalf("Put: %v", err)
+	}
+
+	bp, _ := bs.path(dgst)
+	fi1, err := drvr.Stat(ctx, bp)
+	if err != nil {
+		t.Fatalf("Stat: %v", err)
+	}
+
+	if _, err := bs.Put(ctx, "application/octet-stream", content); err != nil {
+		t.Fatalf("second Put: %v", err)
+	}
+
+	fi2, _ := drvr.Stat(ctx, bp)
+	if fi1.ModTime() != fi2.ModTime() {
+		t.Fatal("Put rewrote data that already existed")
+	}
+}
+
+func isPathNotFound(err error) bool {
+	_, ok := err.(driver.PathNotFoundError)
+	return ok
+}
+
+// TestBlobStorePutAfterGCWithStaleCache verifies that Put writes data back
+// after GC deletes blob data from the backend, even when the cachedBlobStatter
+// still has a stale cache entry reporting the blob as present. This is the
+// actual bug scenario: after GC, the cache says "exists" but the backend data
+// is gone, so Put must bypass the cache and check the backend directly.
+func TestBlobStorePutAfterGCWithStaleCache(t *testing.T) {
+	ctx := context.Background()
+	drvr := inmemory.New()
+
+	cacheProvider := memory.NewInMemoryBlobDescriptorCacheProvider(memory.UnlimitedSize)
+	repoCache, err := cacheProvider.RepositoryScoped("test/repo")
+	if err != nil {
+		t.Fatalf("RepositoryScoped: %v", err)
+	}
+
+	rawStatter := &blobStatter{driver: drvr}
+	cachedStatter := cache.NewCachedBlobStatter(repoCache, rawStatter)
+
+	bs := &blobStore{
+		driver:  drvr,
+		statter: cachedStatter,
+	}
+
+	content := []byte("test-blob-content")
+	dgst := digest.FromBytes(content)
+
+	// Step 1: Put the blob. This also populates the cache.
+	desc, err := bs.Put(ctx, "application/octet-stream", content)
+	if err != nil {
+		t.Fatalf("initial Put: %v", err)
+	}
+	if desc.Digest != dgst {
+		t.Fatalf("digest: got %q, want %q", desc.Digest, dgst)
+	}
+
+	// Step 2: Verify the cached statter reports the blob exists.
+	_, err = cachedStatter.Stat(ctx, dgst)
+	if err != nil {
+		t.Fatalf("cached statter should report blob exists after Put: %v", err)
+	}
+
+	// Step 3: Simulate GC — delete blob data from backend.
+	bp, _ := bs.path(dgst)
+	if err := drvr.Delete(ctx, bp); err != nil {
+		t.Fatalf("delete (GC simulation): %v", err)
+	}
+
+	// Step 4: Confirm backend has no blob.
+	if _, err := drvr.Stat(ctx, bp); !isPathNotFound(err) {
+		t.Fatalf("expected PathNotFoundError after GC, got: %v", err)
+	}
+
+	// Step 5: Confirm cached statter STILL reports blob exists (stale cache).
+	cachedDesc, err := cachedStatter.Stat(ctx, dgst)
+	if err != nil {
+		t.Fatalf("cached statter should still report blob exists (stale entry): %v", err)
+	}
+	if cachedDesc.Digest != dgst {
+		t.Fatalf("stale cache digest: got %q, want %q", cachedDesc.Digest, dgst)
+	}
+
+	// Step 6: Re-put should succeed because Put bypasses the stale cache
+	// and checks the backend directly.
+	desc, err = bs.Put(ctx, "application/octet-stream", content)
+	if err != nil {
+		t.Fatalf("re-Put after GC with stale cache: %v", err)
+	}
+	if desc.Digest != dgst {
+		t.Fatalf("digest after re-Put: got %q, want %q", desc.Digest, dgst)
+	}
+
+	// Step 7: Verify the content is actually readable.
+	got, err := bs.Get(ctx, dgst)
+	if err != nil {
+		t.Fatalf("Get after re-Put: %v", err)
+	}
+	if string(got) != string(content) {
+		t.Fatalf("content: got %q, want %q", got, content)
+	}
+}