feat(ecr): reclaim manifest objects + GC orphaned blobs on digest delete (#390)

DELETE /v2/{name}/manifests/{digest} was a 202 no-op. Now delete-by-digest
removes the manifest record, deletes the manifest object, and sweeps the
account-wide blob pool: each child blob no longer referenced by any live
manifest is reclaimed, while shared blobs survive.

- DeleteImage (digest path) captures ChildDigests, deletes the manifest object,
  then reclaimManifest() drops orphaned blobs via referencedDigests() (a
  mark-and-sweep over the live manifest set across the account's repos).
- OCI deleteManifest by-digest wired to DeleteImage (real work + 202; 404 on
  unknown). Tag delete stays KV-only so untagged images persist (AWS-faithful).
- Predastore reclaim is best-effort: a failure leaks bytes but never undoes the
  logical delete.

spinifex/gateway/ecr/registry.go

@@ -574,8 +574,18 @@ func (reg *Registry) putManifest(w http.ResponseWriter, r *http.Request, name, r
 }
 
 func (reg *Registry) deleteManifest(w http.ResponseWriter, name, reference string) {
-	// Tag pointer deletion only; digest GC is out of scope.
+	// Delete by digest removes the manifest record + object and reclaims any
+	// orphaned blobs; delete by tag removes only the tag pointer (the untagged
+	// image persists, matching AWS).
 	if ecr.ValidateDigest(reference) {
+		if _, err := reg.DeleteImage(reg.AccountID, name, "", reference); err != nil {
+			if errors.Is(err, ErrImageNotFound) {
+				WriteError(w, http.StatusNotFound, "MANIFEST_UNKNOWN", "manifest unknown")
+				return
+			}
+			reg.internal(w, "delete manifest", err)
+			return
+		}
 		w.WriteHeader(http.StatusAccepted)
 		return
 	}

spinifex/gateway/ecr/registry_image.go

@@ -4,6 +4,7 @@ import (
 	"encoding/hex"
 	"errors"
 	"io"
+	"log/slog"
 	"net/http"
 	"slices"
 	"strings"
@@ -210,7 +211,8 @@ func (reg *Registry) DeleteImage(account, repo, tag, digest string) (string, err
 		return resolved, nil
 	}
 
-	if _, err := scoped.Meta.GetManifestMeta(account, repo, digest); err != nil {
+	meta, err := scoped.Meta.GetManifestMeta(account, repo, digest)
+	if err != nil {
 		if errors.Is(err, ecr.ErrNotFound) {
 			return "", ErrImageNotFound
 		}
@@ -237,9 +239,76 @@ func (reg *Registry) DeleteImage(account, repo, tag, digest string) (string, err
 	if err := scoped.Meta.DeleteManifestMeta(account, repo, digest); err != nil {
 		return "", err
 	}
+
+	// The KV record is gone, so the image is logically deleted. Predastore
+	// reclaim is best-effort: a failure leaks bytes (a capacity nuisance) but
+	// does not undo the delete.
+	scoped.reclaimManifest(account, repo, digest, meta.ChildDigests)
 	return digest, nil
 }
 
+// reclaimManifest deletes a manifest's predastore object and any child blob no
+// longer referenced by a live manifest in the account pool. It must be called
+// after the manifest's KV meta is removed, so referencedDigests excludes it.
+func (reg *Registry) reclaimManifest(account, repo, digest string, children []string) {
+	if _, err := reg.Store.DeleteObject(&s3.DeleteObjectInput{
+		Bucket: aws.String(ecr.AccountBucket(account)),
+		Key:    aws.String(ecr.ManifestKey(repo, digest)),
+	}); err != nil {
+		slog.Warn("ECR/GC: manifest object delete failed", "repo", repo, "digest", digest, "err", err)
+	}
+
+	if len(children) == 0 {
+		return
+	}
+	referenced, err := reg.referencedDigests(account)
+	if err != nil {
+		slog.Warn("ECR/GC: reference scan failed, skipping blob reclaim", "account", account, "err", err)
+		return
+	}
+	for _, child := range children {
+		if referenced[child] {
+			continue
+		}
+		if _, err := reg.Store.DeleteObject(&s3.DeleteObjectInput{
+			Bucket: aws.String(ecr.AccountBucket(account)),
+			Key:    aws.String(ecr.BlobKey(child)),
+		}); err != nil {
+			slog.Warn("ECR/GC: blob delete failed", "digest", child, "err", err)
+		}
+	}
+}
+
+// referencedDigests returns the set of child digests referenced by every live
+// manifest across the account's repos. The account blob pool is shared, so the
+// reference set spans all repos, not just the one being mutated.
+func (reg *Registry) referencedDigests(account string) (map[string]bool, error) {
+	repos, err := reg.Meta.ListRepos(account)
+	if err != nil {
+		return nil, err
+	}
+	referenced := make(map[string]bool)
+	for _, repo := range repos {
+		digests, err := reg.Meta.ListManifests(account, repo)
+		if err != nil {
+			return nil, err
+		}
+		for _, key := range digests {
+			meta, err := reg.Meta.GetManifestMeta(account, repo, key)
+			if err != nil {
+				if errors.Is(err, ecr.ErrNotFound) {
+					continue
+				}
+				return nil, err
+			}
+			for _, child := range meta.ChildDigests {
+				referenced[child] = true
+			}
+		}
+	}
+	return referenced, nil
+}
+
 // mediaTypeAccepted reports whether mediaType is in the accepted set.
 func mediaTypeAccepted(accepted []string, mediaType string) bool {
 	return slices.Contains(accepted, mediaType)

spinifex/gateway/ecr/registry_image_test.go

@@ -6,6 +6,8 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/s3"
 	"github.com/mulgadc/spinifex/spinifex/handlers/ecr"
 	"github.com/mulgadc/spinifex/spinifex/objectstore"
 	"github.com/mulgadc/spinifex/spinifex/testutil"
@@ -13,6 +15,59 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+// objectExists reports whether a predastore key is present in the test account
+// bucket.
+func objectExists(reg *Registry, key string) bool {
+	_, err := reg.Store.HeadObject(&s3.HeadObjectInput{
+		Bucket: aws.String(ecr.AccountBucket(testAccount)),
+		Key:    aws.String(key),
+	})
+	return err == nil
+}
+
+// storeLayersManifest stores a manifest referencing the given layer blobs under
+// repo:tag, returning its digest.
+func storeLayersManifest(t *testing.T, reg *Registry, repo, tag string, layers ...string) string {
+	t.Helper()
+	refs := make([]string, len(layers))
+	for i, l := range layers {
+		refs[i] = fmt.Sprintf(`{"digest":"%s"}`, l)
+	}
+	body := fmt.Appendf(nil,
+		`{"schemaVersion":2,"mediaType":"%s","layers":[%s]}`,
+		mediaTypeDockerManifest, strings.Join(refs, ","))
+	digest, err := reg.StoreManifest(testAccount, repo, tag, mediaTypeDockerManifest, body)
+	require.NoError(t, err)
+	return digest
+}
+
+func TestDeleteImage_ReclaimsBlobsKeepsShared(t *testing.T) {
+	reg := newTestRegistry()
+	repo := "team/app"
+	shared := pushBlob(t, reg, repo, []byte("shared-layer-bytes"))
+	exclusive := pushBlob(t, reg, repo, []byte("exclusive-layer-bytes"))
+
+	digA := storeLayersManifest(t, reg, repo, "a", shared, exclusive)
+	digB := storeLayersManifest(t, reg, repo, "b", shared)
+
+	require.True(t, objectExists(reg, ecr.ManifestKey(repo, digA)))
+	require.True(t, objectExists(reg, ecr.BlobKey(exclusive)))
+
+	got, err := reg.DeleteImage(testAccount, repo, "", digA)
+	require.NoError(t, err)
+	assert.Equal(t, digA, got)
+
+	// Manifest A object + its exclusive blob reclaimed.
+	assert.False(t, objectExists(reg, ecr.ManifestKey(repo, digA)), "manifest A object should be gone")
+	assert.False(t, objectExists(reg, ecr.BlobKey(exclusive)), "exclusive blob should be reclaimed")
+
+	// Shared blob + manifest B survive (B still references the shared blob).
+	assert.True(t, objectExists(reg, ecr.BlobKey(shared)), "shared blob must survive")
+	assert.True(t, objectExists(reg, ecr.ManifestKey(repo, digB)), "manifest B object must survive")
+	_, _, _, err = reg.GetManifest(testAccount, repo, digB, nil)
+	require.NoError(t, err)
+}
+
 // seedImage pushes a config + layer blob and stores an image manifest tagged
 // `tag`, returning the manifest bytes and digest.
 func seedImage(t *testing.T, reg *Registry, repo, tag string) ([]byte, string) {

spinifex/gateway/ecr/registry_test.go

@@ -271,6 +271,30 @@ func TestManifest_Delete_Tag(t *testing.T) {
 	assert.Equal(t, http.StatusNotFound, w.Code)
 }
 
+// TestManifest_Delete_ByDigest proves delete-by-digest is no longer a no-op: it
+// removes the manifest record + object, and a subsequent GET is a 404.
+func TestManifest_Delete_ByDigest(t *testing.T) {
+	reg := newTestRegistry()
+	repo := "team/app"
+	cfg := pushBlob(t, reg, repo, []byte("cfg-bytes"))
+	manifest := fmt.Appendf(nil,
+		`{"mediaType":"%s","config":{"digest":"%s"},"layers":[]}`, mediaTypeDockerManifest, cfg)
+	mdg := digestOf(manifest)
+	do(reg, http.MethodPut, "/v2/"+repo+"/manifests/v1", manifest,
+		map[string]string{"Content-Type": mediaTypeDockerManifest})
+
+	w := do(reg, http.MethodDelete, "/v2/"+repo+"/manifests/"+mdg, nil, nil)
+	assert.Equal(t, http.StatusAccepted, w.Code)
+
+	// The record is gone: GET by digest now 404s, and the object was reclaimed.
+	w = do(reg, http.MethodGet, "/v2/"+repo+"/manifests/"+mdg, nil, nil)
+	assert.Equal(t, http.StatusNotFound, w.Code)
+
+	// Deleting again -> 404 (record already gone).
+	w = do(reg, http.MethodDelete, "/v2/"+repo+"/manifests/"+mdg, nil, nil)
+	assert.Equal(t, http.StatusNotFound, w.Code)
+}
+
 func TestTagsList(t *testing.T) {
 	reg := newTestRegistry()
 	repo := "team/app"