Repository: distribution/distribution

Issue:
Garbage collect deletes blobs in disk but blobs are still in memory and registry does not accept the push twice
### Description

By default conf.yml has the following configuration set, which keeps blob descriptors in memory:

```yaml
storage:
  cache:
    blobdescriptor: inmemory
```
Image tag deletion process deletes blobs and layers from disk, but does not delete in memory descriptors.
Due to in memory descriptors, it's not possible to re-push the same image because registry uses cache layers but such layers are not there actually.

### Reproduce

> reproducable in both latest V3 (`3.0.0-rc.2) and V2; I'll use V3 in this example

When a image is pushed into the registry, deleted and garbage collect is executed, the blobs are fully removed from disk:

```bash
# start a registry
docker run -d -e REGISTRY_STORAGE_DELETE_ENABLED=true -p 5000:5000 --rm --name registry-in-memory-blobs registry:3.0.0-rc.2

# creating a random tag and push it
docker tag registry:3.0.0-rc.2 localhost:5000/registry:3.0.0-rc.2
docker push localhost:5000/registry:3.0.0-rc.2

# then delete the tag and run garbage collect
curl -X DELETE http://localhost:5000/v2/registry/manifests/sha256:776203b7933004f7b6cada731e171469e3e267a1e77912fd8e0fc26c16f896f2
docker exec registry-in-memory-blobs /bin/registry garbage-collect /etc/distribution/config.yml
```

Previous commands successfully deleted blobs from disk, but their descriptions are still in memory.

When the same push takes place again, registry says blobs/layers are already there and does not really accept the push, but uses the cache layers instead of pushing:
```bash
docker push localhost:5000/registry:3.0.0-rc.2
# 2b07afdb0755: Layer already exists 
# 55659e28f40e: Layer already exists 
# 6323af7fd13e: Layer already exists 
# d73b26c48fce: Layer already exists 
# a0904247e36a: Layer already exists 
# 3.0.0-rc.2: digest: sha256:776203b7933004f7b6cada731e171469e3e267a1e77912fd8e0fc26c16f896f2 size: 1364
```

But no new layers were created on disk, and if we try to pull it will not work:
```bash
docker pull localhost:5000/registry:3.0.0-rc.2
# Error response from daemon: manifest for localhost:5000/registry:3.0.0-rc.2 not found: manifest unknown: manifest unknown
```

So, the HTTP DELETE request and garbage collect successfully deleted the blobs and layers from disk, but haven't deleted the in memory blob descriptors. It makes inconsistency between what is in memory and what is in disk.

If the `blobdescriptor: inmemory` is not set, them this behavir does not happen

### Expected behavior

registry allows to push the same image after deleting it

### registry version

3.0.0-rc.2

### Additional Info

_No response_

Hints from issue comments:
@Ollegio: similar issue here https://github.com/distribution/distribution/issues/4525
 they said nothing about cache, but I think it should be the same problem, also possible solution
https://github.com/distribution/distribution/issues/4525#issuecomment-2710688725@milosgajdos: Think we need this to get over the finish line https://github.com/distribution/distribution/pull/3323@milosgajdos: @Ollegio feel free to open a new PR; I'm happy to review it and push it through.@dkvrgic: Hi guys, I am using registry image v3.0.0 official one , this behavior is still happening. I need to restart the registry container after garbage-collect cmd being executed in order to prevent this. Any feedback on this issue ?? @DanielLeone: Confirming this is still a problem for me for both v2 and v3 registry, here's the smallest test script I've got:

```sh
# run a local registry
docker rm -f local-registry
docker run -d --name local-registry -p 6000:5000 -e REGISTRY_STORAGE_CACHE_BLOBDESCRIPTOR=inmemory -v "$(pwd)/private-registry:/var/lib/registry" registry:2.8.3

# create a simple dockerfile with a few shared layers between 2 images
cat > Dockerfile <<'EOF'
FROM busybox:1.36 AS base
WORKDIR /out
RUN printf 'a' > a
RUN printf 'b' > b
RUN printf 'c' > c

FROM base AS image-one
RUN printf 'd-1' > d
CMD ["true"]

FROM base AS image-two
RUN printf 'd-2' > d
CMD ["true"]
EOF

# build both images
docker build -t localhost:6000/test:image-one --target image-one .
docker build -t localhost:6000/test:image-two --target image-two .

# tag, push and pull image one as latest
docker tag localhost:6000/test:image-one localhost:6000/test:latest
docker push localhost:6000/test:latest
docker pull localhost:6000/test:latest

# tag, push and pull image two over the top
docker tag localhost:6000/test:image-two localhost:6000/test:latest
docker push localhost:6000/test:latest
docker pull localhost:6000/test:latest

# run garbage collection
docker exec local-registry registry garbage-collect -m /etc/docker/registry/config.yml

# repush image one back over the top of image two
docker tag localhost:6000/test:image-one localhost:6000/test:latest
docker push localhost:6000/test:latest

# pull is broken, only when the cache is enabled on the registry
docker pull localhost:6000/test:latest
```



v3 exhibits the same failure with the same script, just needs the config path changed


```sh
# run a local registry
docker rm -f local-registry
docker run -d --name local-registry -p 6000:5000 -e REGISTRY_STORAGE_CACHE_BLOBDESCRIPTOR=inmemory -v "$(pwd)/private-registry:/var/lib/registry" registry:3.0.0

# create a simple dockerfile with a few shared layers between 2 images
cat > Dockerfile <<'EOF'
FROM busybox:1.36 AS base
WORKDIR /out
RUN printf 'a' > a
RUN printf 'b' > b
RUN printf 'c' > c

FROM base AS image-one
RUN printf 'd-1' > d
CMD ["true"]

FROM base AS image-two
RUN printf 'd-2' > d
CMD ["true"]
EOF

# build both images
docker build -t localhost:6000/test:image-one --target image-one .
docker build -t localhost:6000/test:image-two --target image-two .

# tag, push and pull image one as latest
docker tag localhost:6000/test:image-one localhost:6000/test:latest
docker push localhost:6000/test:latest
docker pull localhost:6000/test:latest

# tag, push and pull image two over the top
docker tag localhost:6000/test:image-two localhost:6000/test:latest
docker push localhost:6000/test:latest
docker pull localhost:6000/test:latest

# run garbage collection
docker exec local-registry registry garbage-collect -m /etc/distribution/config.yml

# repush image one back over the top of image two
docker tag localhost:6000/test:image-one localhost:6000/test:latest
docker push localhost:6000/test:latest

# pull is broken now for image one only when the cache is enabled
docker pull localhost:6000/test:latest
```


Solution patch:
diff --git a/registry/storage/registry.go b/registry/storage/registry.go
index ab8fb1f9..5b4d6c42 100644
--- a/registry/storage/registry.go
+++ b/registry/storage/registry.go
@@ -261,6 +261,10 @@ func (repo *repository) Manifests(ctx context.Context, options ...distribution.M
 		linkPath:   manifestRevisionLinkPath,
 	}
 
+	if repo.descriptorCache != nil {
+		statter = cache.NewCachedBlobStatter(repo.descriptorCache, statter)
+	}
+
 	if repo.registry.blobDescriptorServiceFactory != nil {
 		statter = repo.registry.blobDescriptorServiceFactory.BlobAccessController(statter)
 	}



Test patch:
diff --git a/registry/storage/manifeststore_test.go b/registry/storage/manifeststore_test.go
index 57c9fa83..1beb78c8 100644
--- a/registry/storage/manifeststore_test.go
+++ b/registry/storage/manifeststore_test.go
@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"errors"
 	"io"
 	"reflect"
 	"testing"
@@ -720,3 +721,154 @@ func ociIndexFromDesriptorsWithMediaType(descriptors []v1.Descriptor, mediaType
 
 	return &d, nil
 }
+
+func TestManifestStorageCache(t *testing.T) {
+	repoName, _ := reference.WithName("foo/bar")
+	cacheProvider := memory.NewInMemoryBlobDescriptorCacheProvider(memory.UnlimitedSize)
+	repositoryScopedCacheProvider, err := cacheProvider.RepositoryScoped(repoName.String())
+	if err != nil {
+		t.Fatalf("unexpected error getting repository scoped cache: %v", err)
+	}
+	env := newManifestStoreTestEnv(t, repoName, "thetag", BlobDescriptorCacheProvider(cacheProvider), EnableDelete)
+	ctx := context.Background()
+	ms, err := env.repository.Manifests(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Push a config, and reference it in the manifest
+	sampleConfig := []byte(`{
+		"architecture": "amd64",
+		"history": [
+		  {
+		    "created": "2015-10-31T22:22:54.690851953Z",
+		    "created_by": "/bin/sh -c #(nop) ADD file:a3bc1e842b69636f9df5256c49c5374fb4eef1e281fe3f282c65fb853ee171c5 in /"
+		  },
+		],
+		"rootfs": {
+		  "diff_ids": [
+		    "sha256:c6f988f4874bb0add23a778f753c65efe992244e148a1d2ec2a8b664fb66bbd1",
+		  ],
+		  "type": "layers"
+		}
+	}`)
+
+	// Build a manifest and store it and its layers in the registry
+
+	blobStore := env.repository.Blobs(ctx)
+	d, err := blobStore.Put(ctx, schema2.MediaTypeImageConfig, sampleConfig)
+	if err != nil {
+		t.Fatal(err)
+	}
+	builder := schema2.NewManifestBuilder(d, sampleConfig)
+
+	m := &schema2.Manifest{
+		Versioned: specs.Versioned{SchemaVersion: 2},
+		MediaType: schema2.MediaTypeManifest,
+		Config: v1.Descriptor{
+			Digest:    digest.FromBytes(sampleConfig),
+			Size:      int64(len(sampleConfig)),
+			MediaType: schema2.MediaTypeImageConfig,
+		},
+		Layers: []v1.Descriptor{},
+	}
+
+	// Build up some test layers and add them to the manifest, saving the
+	// readseekers for upload later.
+	testLayers := map[digest.Digest]io.ReadSeeker{}
+	for i := 0; i < 2; i++ {
+		rs, dgst, err := testutil.CreateRandomTarFile()
+		if err != nil {
+			t.Fatal("unexpected error generating test layer file")
+		}
+
+		testLayers[dgst] = rs
+		layer := v1.Descriptor{
+			Digest:    dgst,
+			Size:      6323,
+			MediaType: schema2.MediaTypeLayer,
+		}
+		m.Layers = append(m.Layers, layer)
+	}
+
+	// Now, upload the layers that were missing!
+	for dgst, rs := range testLayers {
+		wr, err := env.repository.Blobs(env.ctx).Create(env.ctx)
+		if err != nil {
+			t.Fatalf("unexpected error creating test upload: %v", err)
+		}
+
+		if _, err := io.Copy(wr, rs); err != nil {
+			t.Fatalf("unexpected error copying to upload: %v", err)
+		}
+
+		if _, err := wr.Commit(env.ctx, v1.Descriptor{Digest: dgst}); err != nil {
+			t.Fatalf("unexpected error finishing upload: %v", err)
+		}
+		if err := builder.AppendReference(v1.Descriptor{Digest: dgst, MediaType: schema2.MediaTypeLayer}); err != nil {
+			t.Fatalf("unexpected error appending references: %v", err)
+		}
+	}
+
+	sm, err := builder.Build(ctx)
+	if err != nil {
+		t.Fatalf("%s: unexpected error generating manifest: %v", repoName, err)
+	}
+
+	var manifestDigest digest.Digest
+	if manifestDigest, err = ms.Put(ctx, sm); err != nil {
+		t.Fatalf("unexpected error putting manifest: %v", err)
+	}
+
+	_, err = repositoryScopedCacheProvider.Stat(ctx, manifestDigest)
+	if err != nil {
+		t.Errorf("Manifest should be cached")
+	}
+
+	exists, err := ms.Exists(ctx, manifestDigest)
+	if err != nil {
+		t.Fatalf("unexpected error checking manifest existence: %#v", err)
+	}
+
+	if !exists {
+		t.Fatal("manifest should exist")
+	}
+
+	// Test deleting manifests
+	err = ms.Delete(ctx, manifestDigest)
+	if err != nil {
+		t.Fatalf("unexpected an error deleting manifest by digest: %v", err)
+	}
+
+	exists, err = ms.Exists(ctx, manifestDigest)
+	if err != nil {
+		t.Fatal("Error querying manifest existence")
+	}
+	if exists {
+		t.Errorf("Deleted manifest should not exist")
+	}
+
+	deletedManifest, err := ms.Get(ctx, manifestDigest)
+	if err == nil {
+		t.Errorf("Unexpected success getting deleted manifest")
+	}
+	switch err.(type) {
+	case distribution.ErrManifestUnknownRevision:
+		break
+	default:
+		t.Errorf("Unexpected error getting deleted manifest: %s", reflect.ValueOf(err).Type())
+	}
+
+	if deletedManifest != nil {
+		t.Errorf("Deleted manifest get returned non-nil")
+	}
+
+	// Test cache is cleared after manifest delete
+	_, err = repositoryScopedCacheProvider.Stat(ctx, manifestDigest)
+	if err == nil {
+		t.Errorf("Unexpected success getting deleted manifest")
+	}
+	if !errors.Is(err, distribution.ErrBlobUnknown) {
+		t.Errorf("Unexpected error getting cached manifest: %v", err)
+	}
+}