diff --git a/registry/storage/blobstore.go b/registry/storage/blobstore.go index c36bd7e9e..3f36cf0a6 100644 --- a/registry/storage/blobstore.go +++ b/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. diff --git a/registry/storage/blobstore_test.go b/registry/storage/blobstore_test.go new file mode 100644 index 000000000..c59fe5d56 --- /dev/null +++ b/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) + } +}