diff --git a/metadata/db.go b/metadata/db.go
index 8241930a9..a187439ce 100644
--- a/metadata/db.go
+++ b/metadata/db.go
@@ -325,6 +325,8 @@ func (m *DB) publishEvents(events []namespacedEvent) {
 			ctx := namespaces.WithNamespace(ctx, ne.namespace)
 			var topic string
 			switch ne.event.(type) {
+			case *eventstypes.ImageDelete:
+				topic = "/images/delete"
 			case *eventstypes.SnapshotRemove:
 				topic = "/snapshot/remove"
 			default:
diff --git a/metadata/db_test.go b/metadata/db_test.go
index fd0b12aad..c9f8d132f 100644
--- a/metadata/db_test.go
+++ b/metadata/db_test.go
@@ -655,6 +655,13 @@ func create(obj object, tx *bolt.Tx, db *DB, cs content.Store, sn snapshots.Snap
 		if err != nil {
 			return nil, fmt.Errorf("failed to create image: %w", err)
 		}
+		if !obj.removed {
+			node = &gc.Node{
+				Type:      ResourceImage,
+				Namespace: namespace,
+				Key:       image.Name,
+			}
+		}
 	case testContainer:
 		container := containers.Container{
 			ID:          v.id,
diff --git a/metadata/gc.go b/metadata/gc.go
index 0cfbd5b33..855bcab51 100644
--- a/metadata/gc.go
+++ b/metadata/gc.go
@@ -41,6 +41,8 @@ const (
 	ResourceContainer
 	// ResourceTask specifies a task resource
 	ResourceTask
+	// ResourceImage specifies an image
+	ResourceImage
 	// ResourceLease specifies a lease
 	ResourceLease
 	// ResourceIngest specifies a content ingest
@@ -54,6 +56,7 @@ const (
 const (
 	resourceContentFlat  = ResourceContent | 0x20
 	resourceSnapshotFlat = ResourceSnapshot | 0x20
+	resourceImageFlat    = ResourceImage | 0x20
 )
 
 var (
@@ -61,8 +64,22 @@ var (
 	labelGCRef        = []byte("containerd.io/gc.ref.")
 	labelGCSnapRef    = []byte("containerd.io/gc.ref.snapshot.")
 	labelGCContentRef = []byte("containerd.io/gc.ref.content")
-	labelGCExpire     = []byte("containerd.io/gc.expire")
-	labelGCFlat       = []byte("containerd.io/gc.flat")
+	labelGCImageRef   = []byte("containerd.io/gc.ref.image")
+
+	// labelGCExpire indicates that an object is collectible after the
+	// provided time. For image objects, this makes them available to
+	// garbage collect when expired, when not provided, image objects
+	// are root objects that never expire. For non-root objects such
+	// as content or snapshots, these objects will be treated like
+	// root objects before their expiration.
+	// Expected format is RFC 3339
+	labelGCExpire = []byte("containerd.io/gc.expire")
+
+	// labelGCFlat indicates that a lease is flat and only intends to
+	// lease the referenced objects, not their references. This can be
+	// used to avoid leasing an entire tree of objects when only the root
+	// object is needed.
+	labelGCFlat = []byte("containerd.io/gc.flat")
 )
 
 // CollectionContext manages a resource collection during a single run of
@@ -137,6 +154,19 @@ func startGCContext(ctx context.Context, collectors map[gc.ResourceType]Collecto
 				fn(gcnode(ResourceSnapshot, ns, fmt.Sprintf("%s/%s", snapshotter, v)))
 			},
 		},
+		{
+			key: labelGCImageRef,
+			fn: func(ns string, k, v []byte, fn func(gc.Node)) {
+				if ks := string(k); ks != string(labelGCImageRef) {
+					// Allow reference naming separated by . or /, ignore names
+					if ks[len(labelGCImageRef)] != '.' && ks[len(labelGCImageRef)] != '/' {
+						return
+					}
+				}
+
+				fn(gcnode(ResourceImage, ns, string(v)))
+			},
+		},
 	}
 	if len(collectors) > 0 {
 		contexts = map[gc.ResourceType]CollectionContext{}
@@ -166,7 +196,7 @@ func startGCContext(ctx context.Context, collectors map[gc.ResourceType]Collecto
 			}
 			contexts[rt] = c
 		}
-		// Sort labelHandlers to ensure key seeking is always forwardS
+		// Sort labelHandlers to ensure key seeking is always forward
 		sort.Slice(labelHandlers, func(i, j int) bool {
 			return bytes.Compare(labelHandlers[i].key, labelHandlers[j].key) < 0
 		})
@@ -324,6 +354,21 @@ func (c *gcContext) scanRoots(ctx context.Context, tx *bolt.Tx, nc chan<- gc.Nod
 					}
 				}
 
+				itype := ResourceImage
+				if flat {
+					itype = resourceImageFlat
+				}
+
+				ibkt = libkt.Bucket(bucketKeyObjectImages)
+				if ibkt != nil {
+					if err := ibkt.ForEach(func(k, v []byte) error {
+						fn(gcnode(itype, ns, string(k)))
+						return nil
+					}); err != nil {
+						return err
+					}
+				}
+
 				c.leased(ns, string(k), fn)
 
 				return nil
@@ -339,12 +384,10 @@ func (c *gcContext) scanRoots(ctx context.Context, tx *bolt.Tx, nc chan<- gc.Nod
 					return nil
 				}
 
-				target := ibkt.Bucket(k).Bucket(bucketKeyTarget)
-				if target != nil {
-					contentKey := string(target.Get(bucketKeyDigest))
-					fn(gcnode(ResourceContent, ns, contentKey))
+				if !isExpiredImage(ctx, k, ibkt.Bucket(k), expThreshold) {
+					fn(gcnode(ResourceImage, ns, string(k)))
 				}
-				return c.sendLabelRefs(ns, ibkt.Bucket(k), fn)
+				return nil
 			}); err != nil {
 				return err
 			}
@@ -482,6 +525,31 @@ func (c *gcContext) references(ctx context.Context, tx *bolt.Tx, node gc.Node, f
 		}
 
 		return c.sendLabelRefs(node.Namespace, bkt, fn)
+
+	case ResourceImage, resourceImageFlat:
+		bkt := getBucket(tx, bucketKeyVersion, []byte(node.Namespace), bucketKeyObjectImages, []byte(node.Key))
+		if bkt == nil {
+			// Node may be created from dead edge
+			return nil
+		}
+		target := bkt.Bucket(bucketKeyTarget)
+		if target != nil {
+			ctype := ResourceContent
+			if node.Type == resourceImageFlat {
+				// For flat leases, keep the target content only
+				ctype = resourceContentFlat
+			}
+			contentKey := string(target.Get(bucketKeyDigest))
+			fn(gcnode(ctype, node.Namespace, contentKey))
+		}
+
+		// Do not send labeled references for flat image refs
+		if node.Type == resourceImageFlat {
+			return nil
+		}
+
+		return c.sendLabelRefs(node.Namespace, bkt, fn)
+
 	case ResourceIngest:
 		// Send expected value
 		bkt := getBucket(tx, bucketKeyVersion, []byte(node.Namespace), bucketKeyObjectContent, bucketKeyObjectIngests, []byte(node.Key))
@@ -576,6 +644,19 @@ func (c *gcContext) scanAll(ctx context.Context, tx *bolt.Tx, fn func(ctx contex
 				}
 			}
 		}
+
+		ibkt := nbkt.Bucket(bucketKeyObjectImages)
+		if ibkt != nil {
+			if err := ibkt.ForEach(func(k, v []byte) error {
+				if v != nil {
+					return nil
+				}
+				node := gcnode(ResourceImage, ns, string(k))
+				return fn(ctx, node)
+			}); err != nil {
+				return err
+			}
+		}
 	}
 
 	c.all(func(n gc.Node) {
@@ -627,6 +708,13 @@ func (c *gcContext) remove(ctx context.Context, tx *bolt.Tx, node gc.Node) (inte
 				}, ssbkt.DeleteBucket([]byte(key))
 			}
 		}
+	case ResourceImage:
+		ibkt := nsbkt.Bucket(bucketKeyObjectImages)
+		if ibkt != nil {
+			return &eventstypes.ImageDelete{
+				Name: node.Key,
+			}, ibkt.DeleteBucket([]byte(node.Key))
+		}
 	case ResourceLease:
 		lbkt := nsbkt.Bucket(bucketKeyObjectLeases)
 		if lbkt != nil {
@@ -680,6 +768,22 @@ func isRootRef(bkt *bolt.Bucket) bool {
 	return false
 }
 
+func isExpiredImage(ctx context.Context, k []byte, bkt *bolt.Bucket, expTheshold time.Time) bool {
+	lbkt := bkt.Bucket(bucketKeyObjectLabels)
+	if lbkt != nil {
+		el := lbkt.Get(labelGCExpire)
+		if el != nil {
+			exp, err := time.Parse(time.RFC3339, string(el))
+			if err != nil {
+				log.G(ctx).WithError(err).WithField("image", string(k)).Infof("ignoring invalid expiration value %q", string(el))
+				return false
+			}
+			return expTheshold.After(exp)
+		}
+	}
+	return false
+}
+
 func gcnode(t gc.ResourceType, ns, key string) gc.Node {
 	return gc.Node{
 		Type:      t,
diff --git a/metadata/gc_test.go b/metadata/gc_test.go
index 26508d328..689f8481e 100644
--- a/metadata/gc_test.go
+++ b/metadata/gc_test.go
@@ -17,12 +17,15 @@
 package metadata
 
 import (
+	"bytes"
 	"context"
+	"fmt"
 	"io"
 	"math/rand"
 	"path/filepath"
 	"sort"
 	"testing"
+	"text/tabwriter"
 	"time"
 
 	"github.com/containerd/containerd/gc"
@@ -49,7 +52,12 @@ func TestGCRoots(t *testing.T) {
 	alters := []alterFunc{
 		addImage("ns1", "image1", dgst(1), nil),
 		addImage("ns1", "image2", dgst(2), labelmap(string(labelGCSnapRef)+"overlay", "sn2")),
-		addImage("ns2", "image3", dgst(10), labelmap(string(labelGCContentRef), dgst(11).String())),
+		addImage("ns2", "image3", dgst(10), labelmap(
+			string(labelGCContentRef), dgst(11).String(),
+			string(labelGCImageRef), "image4",
+		)),
+		addImage("ns2", "image4", dgst(12), labelmap(string(labelGCExpire), time.Now().Format(time.RFC3339))),
+		addImage("ns2", "image5", dgst(13), labelmap(string(labelGCExpire), time.Now().Format(time.RFC3339))),
 		addContainer("ns1", "container1", "overlay", "sn4", nil),
 		addContainer("ns1", "container2", "overlay", "sn5", labelmap(string(labelGCSnapRef)+"overlay", "sn6")),
 		addContainer("ns1", "container3", "overlay", "sn7", labelmap(
@@ -89,17 +97,20 @@ func TestGCRoots(t *testing.T) {
 		addLease("ns2", "l3", labelmap(string(labelGCExpire), time.Now().Add(time.Hour).Format(time.RFC3339))),
 		addLeaseContent("ns2", "l3", dgst(6)),
 		addLeaseSnapshot("ns2", "l3", "overlay", "sn7"),
+		addLeaseImage("ns2", "l3", "image5"),
 		addLeaseIngest("ns2", "l3", "ingest-4"),
 		addLeaseIngest("ns2", "l3", "ingest-5"),
 		addLease("ns2", "l4", labelmap(string(labelGCExpire), time.Now().Format(time.RFC3339))),
 		addLeaseContent("ns2", "l4", dgst(7)),
 		addLeaseSnapshot("ns2", "l4", "overlay", "sn8"),
+		addLeaseImage("ns2", "l4", "image4"),
 		addLeaseIngest("ns2", "l4", "ingest-6"),
 		addLeaseIngest("ns2", "l4", "ingest-7"),
 
 		addLease("ns3", "l1", labelmap(string(labelGCFlat), time.Now().Add(time.Hour).Format(time.RFC3339))),
 		addLeaseContent("ns3", "l1", dgst(1)),
 		addLeaseSnapshot("ns3", "l1", "overlay", "sn1"),
+		addLeaseImage("ns3", "l1", "image1"),
 		addLeaseIngest("ns3", "l1", "ingest-1"),
 
 		addSandbox("ns3", "sandbox1", nil),
@@ -107,8 +118,6 @@ func TestGCRoots(t *testing.T) {
 	}
 
 	expected := []gc.Node{
-		gcnode(ResourceContent, "ns1", dgst(1).String()),
-		gcnode(ResourceContent, "ns1", dgst(2).String()),
 		gcnode(ResourceContent, "ns1", dgst(7).String()),
 		gcnode(ResourceContent, "ns1", dgst(8).String()),
 		gcnode(ResourceContent, "ns1", dgst(9).String()),
@@ -116,9 +125,6 @@ func TestGCRoots(t *testing.T) {
 		gcnode(ResourceContent, "ns2", dgst(4).String()),
 		gcnode(ResourceContent, "ns2", dgst(5).String()),
 		gcnode(ResourceContent, "ns2", dgst(6).String()),
-		gcnode(ResourceContent, "ns2", dgst(10).String()),
-		gcnode(ResourceContent, "ns2", dgst(11).String()),
-		gcnode(ResourceSnapshot, "ns1", "overlay/sn2"),
 		gcnode(ResourceSnapshot, "ns1", "overlay/sn3"),
 		gcnode(ResourceSnapshot, "ns1", "overlay/sn4"),
 		gcnode(ResourceSnapshot, "ns1", "overlay/sn5"),
@@ -129,6 +135,11 @@ func TestGCRoots(t *testing.T) {
 		gcnode(ResourceSnapshot, "ns2", "overlay/sn5"),
 		gcnode(ResourceSnapshot, "ns2", "overlay/sn6"),
 		gcnode(ResourceSnapshot, "ns2", "overlay/sn7"),
+		gcnode(ResourceSnapshot, "ns4", "overlay/sn1"),
+		gcnode(ResourceImage, "ns1", "image1"),
+		gcnode(ResourceImage, "ns1", "image2"),
+		gcnode(ResourceImage, "ns2", "image3"),
+		gcnode(ResourceImage, "ns2", "image5"),
 		gcnode(ResourceLease, "ns2", "l1"),
 		gcnode(ResourceLease, "ns2", "l2"),
 		gcnode(ResourceLease, "ns2", "l3"),
@@ -139,7 +150,7 @@ func TestGCRoots(t *testing.T) {
 		gcnode(ResourceIngest, "ns3", "ingest-1"),
 		gcnode(resourceContentFlat, "ns3", dgst(1).String()),
 		gcnode(resourceSnapshotFlat, "ns3", "overlay/sn1"),
-		gcnode(ResourceSnapshot, "ns4", "overlay/sn1"),
+		gcnode(resourceImageFlat, "ns3", "image1"),
 	}
 
 	if err := db.Update(func(tx *bolt.Tx) error {
@@ -199,6 +210,8 @@ func TestGCRemove(t *testing.T) {
 		gcnode(ResourceSnapshot, "ns1", "overlay/sn3"),
 		gcnode(ResourceSnapshot, "ns1", "overlay/sn4"),
 		gcnode(ResourceSnapshot, "ns2", "overlay/sn1"),
+		gcnode(ResourceImage, "ns1", "image1"),
+		gcnode(ResourceImage, "ns1", "image2"),
 		gcnode(ResourceLease, "ns1", "l1"),
 		gcnode(ResourceLease, "ns2", "l2"),
 		gcnode(ResourceIngest, "ns1", "ingest-1"),
@@ -269,6 +282,10 @@ func TestGCRefs(t *testing.T) {
 		addContent("ns1", dgst(7), labelmap(string(labelGCContentRef)+"/anything-1", dgst(2).String(), string(labelGCContentRef)+"/anything-2", dgst(3).String())),
 		addContent("ns2", dgst(1), nil),
 		addContent("ns2", dgst(2), nil),
+		addImage("ns1", "image1", dgst(3), nil),
+		addImage("ns1", "image2", dgst(4), labelmap(
+			string(labelGCImageRef)+".anything", "image1",
+			string(labelGCContentRef)+".anotherimage", dgst(5).String())),
 		addIngest("ns1", "ingest-1", "", nil),
 		addIngest("ns2", "ingest-2", dgst(8), nil),
 		addSnapshot("ns1", "overlay", "sn1", "", nil),
@@ -334,6 +351,14 @@ func TestGCRefs(t *testing.T) {
 			gcnode(ResourceContent, "ns2", dgst(1).String()),
 			gcnode(ResourceContent, "ns2", dgst(6).String()),
 		},
+		gcnode(ResourceImage, "ns1", "image1"): {
+			gcnode(ResourceContent, "ns1", dgst(3).String()),
+		},
+		gcnode(ResourceImage, "ns1", "image2"): {
+			gcnode(ResourceContent, "ns1", dgst(4).String()),
+			gcnode(ResourceContent, "ns1", dgst(5).String()),
+			gcnode(ResourceImage, "ns1", "image1"),
+		},
 		gcnode(ResourceIngest, "ns1", "ingest-1"): nil,
 		gcnode(ResourceIngest, "ns2", "ingest-2"): {
 			gcnode(ResourceContent, "ns2", dgst(8).String()),
@@ -350,6 +375,12 @@ func TestGCRefs(t *testing.T) {
 			gcnode(ResourceSnapshot, "ns3", "btrfs/sn1"),
 			gcnode(ResourceSnapshot, "ns3", "overlay/sn1"),
 		},
+		gcnode(resourceImageFlat, "ns1", "image1"): {
+			gcnode(resourceContentFlat, "ns1", dgst(3).String()),
+		},
+		gcnode(resourceImageFlat, "ns1", "image2"): {
+			gcnode(resourceContentFlat, "ns1", dgst(4).String()),
+		},
 	}
 
 	if err := db.Update(func(tx *bolt.Tx) error {
@@ -410,17 +441,19 @@ func TestCollectibleResources(t *testing.T) {
 	all := []gc.Node{
 		gcnode(ResourceContent, "ns1", dgst(1).String()),
 		gcnode(ResourceContent, "ns1", dgst(2).String()),
+		gcnode(ResourceImage, "ns1", "image1"),
+		gcnode(ResourceImage, "ns1", "image2"),
 		gcnode(ResourceLease, "ns1", "lease1"),
 		gcnode(ResourceLease, "ns1", "lease2"),
 		gcnode(testResource, "ns1", "test1"),
-		gcnode(testResource, "ns1", "test2"), // 5: Will be removed
+		gcnode(testResource, "ns1", "test2"), // 7: Will be removed
 		gcnode(testResource, "ns1", "test3"),
 		gcnode(testResource, "ns1", "test4"),
 	}
-	removeIndex := 5
+	removeIndex := 7
 	roots := []gc.Node{
-		gcnode(ResourceContent, "ns1", dgst(1).String()),
-		gcnode(ResourceContent, "ns1", dgst(2).String()),
+		gcnode(ResourceImage, "ns1", "image1"),
+		gcnode(ResourceImage, "ns1", "image2"),
 		gcnode(ResourceLease, "ns1", "lease1"),
 		gcnode(testResource, "ns1", "test1"),
 		gcnode(testResource, "ns1", "test3"),
@@ -612,16 +645,63 @@ func checkNodesEqual(t *testing.T, n1, n2 []gc.Node) {
 	sort.Sort(nodeList(n2))
 
 	if len(n1) != len(n2) {
-		t.Fatalf("Nodes do not match\n\tExpected:\n\t%v\n\tActual:\n\t%v", n2, n1)
+		buf := bytes.NewBuffer(nil)
+		tw := tabwriter.NewWriter(buf, 8, 4, 1, ' ', 0)
+		max := len(n1)
+		if len(n2) > max {
+			max = len(n2)
+		}
+		fmt.Fprintln(tw, "Expected:\tActual:")
+		for i := 0; i < max; i++ {
+			var left, right string
+			if i < len(n1) {
+				right = printNode(n1[i])
+			}
+			if i < len(n2) {
+				left = printNode(n2[i])
+			}
+			fmt.Fprintln(tw, left+"\t"+right)
+		}
+		tw.Flush()
+		t.Fatal("Nodes do not match\n" + buf.String())
 	}
 
 	for i := range n1 {
 		if n1[i] != n2[i] {
-			t.Errorf("[%d] root does not match expected: expected %v, got %v", i, n2[i], n1[i])
+			t.Errorf("[%d] root does not match expected: expected %v, got %v", i, printNode(n2[i]), printNode(n1[i]))
 		}
 	}
 }
 
+func printNode(n gc.Node) string {
+	var t string
+	switch n.Type {
+	case ResourceContent:
+		t = "content"
+	case ResourceSnapshot:
+		t = "snapshot"
+	case ResourceContainer:
+		t = "container"
+	case ResourceTask:
+		t = "task"
+	case ResourceImage:
+		t = "image"
+	case ResourceLease:
+		t = "lease"
+	case ResourceIngest:
+		t = "ingest"
+	case resourceContentFlat:
+		t = "content-flat"
+	case resourceSnapshotFlat:
+		t = "snapshot-flat"
+	case resourceImageFlat:
+		t = "image-flat"
+	default:
+		return fmt.Sprintf("%v", n)
+	}
+	return fmt.Sprintf("%s(%s/%s)", t, n.Namespace, n.Key)
+}
+
 type nodeList []gc.Node
 
 func (nodes nodeList) Len() int {
@@ -738,6 +818,16 @@ func addLeaseContent(ns, lid string, dgst digest.Digest) alterFunc {
 	}
 }
 
+func addLeaseImage(ns, lid, image string) alterFunc {
+	return func(bkt *bolt.Bucket) error {
+		cbkt, err := createBuckets(bkt, ns, string(bucketKeyObjectLeases), lid, string(bucketKeyObjectImages))
+		if err != nil {
+			return err
+		}
+		return cbkt.Put([]byte(image), nil)
+	}
+}
+
 func addLeaseIngest(ns, lid, ref string) alterFunc {
 	return func(bkt *bolt.Bucket) error {
 		cbkt, err := createBuckets(bkt, ns, string(bucketKeyObjectLeases), lid, string(bucketKeyObjectIngests))
diff --git a/metadata/images.go b/metadata/images.go
index ff5b624cc..b51d0d610 100644
--- a/metadata/images.go
+++ b/metadata/images.go
@@ -136,6 +136,10 @@ func (s *imageStore) Create(ctx context.Context, image images.Image) (images.Ima
 			return err
 		}
 
+		if err := addImageLease(ctx, tx, image.Name, image.Labels); err != nil {
+			return err
+		}
+
 		ibkt, err := bkt.CreateBucket([]byte(image.Name))
 		if err != nil {
 			if err != bolt.ErrBucketExists {
@@ -236,6 +240,11 @@ func (s *imageStore) Update(ctx context.Context, image images.Image, fieldpaths
 			return err
 		}
 
+		// Collectible label may be added, if so add to lease
+		if err := addImageLease(ctx, tx, updated.Name, updated.Labels); err != nil {
+			return err
+		}
+
 		updated.CreatedAt = createdat
 		if tm := epoch.FromContext(ctx); tm != nil {
 			updated.UpdatedAt = tm.UTC()
@@ -263,6 +272,10 @@ func (s *imageStore) Delete(ctx context.Context, name string, opts ...images.Del
 			return fmt.Errorf("image %q: %w", name, errdefs.ErrNotFound)
 		}
 
+		if err := removeImageLease(ctx, tx, name); err != nil {
+			return err
+		}
+
 		if err = bkt.DeleteBucket([]byte(name)); err != nil {
 			if err == bolt.ErrBucketNotFound {
 				err = fmt.Errorf("image %q: %w", name, errdefs.ErrNotFound)
diff --git a/metadata/leases.go b/metadata/leases.go
index 03fa75af3..255de3f70 100644
--- a/metadata/leases.go
+++ b/metadata/leases.go
@@ -279,6 +279,20 @@ func (lm *leaseManager) ListResources(ctx context.Context, lease leases.Lease) (
 			}
 		}
 
+		// images resources
+		if ibkt := topbkt.Bucket(bucketKeyObjectImages); ibkt != nil {
+			if err := ibkt.ForEach(func(k, _ []byte) error {
+				rs = append(rs, leases.Resource{
+					ID:   string(k),
+					Type: string(bucketKeyObjectImages),
+				})
+
+				return nil
+			}); err != nil {
+				return err
+			}
+		}
+
 		// ingest resources
 		if lbkt := topbkt.Bucket(bucketKeyObjectIngests); lbkt != nil {
 			if err := lbkt.ForEach(func(k, _ []byte) error {
@@ -461,6 +475,59 @@ func removeIngestLease(ctx context.Context, tx *bolt.Tx, ref string) error {
 	return bkt.Delete([]byte(ref))
 }
 
+func addImageLease(ctx context.Context, tx *bolt.Tx, ref string, labels map[string]string) error {
+	lid, ok := leases.FromContext(ctx)
+	if !ok {
+		return nil
+	}
+
+	// If image doesn't have expiration, it does not need to be leased
+	if _, ok := labels[string(labelGCExpire)]; !ok {
+		return nil
+	}
+
+	namespace, ok := namespaces.Namespace(ctx)
+	if !ok {
+		panic("namespace must already be required")
+	}
+
+	bkt := getBucket(tx, bucketKeyVersion, []byte(namespace), bucketKeyObjectLeases, []byte(lid))
+	if bkt == nil {
+		return fmt.Errorf("lease does not exist: %w", errdefs.ErrNotFound)
+	}
+
+	bkt, err := bkt.CreateBucketIfNotExists(bucketKeyObjectImages)
+	if err != nil {
+		return err
+	}
+
+	if err := bkt.Put([]byte(ref), nil); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func removeImageLease(ctx context.Context, tx *bolt.Tx, ref string) error {
+	lid, ok := leases.FromContext(ctx)
+	if !ok {
+		return nil
+	}
+
+	namespace, ok := namespaces.Namespace(ctx)
+	if !ok {
+		panic("namespace must already be checked")
+	}
+
+	bkt := getBucket(tx, bucketKeyVersion, []byte(namespace), bucketKeyObjectLeases, []byte(lid), bucketKeyObjectImages)
+	if bkt == nil {
+		// Key does not exist so we return nil
+		return nil
+	}
+
+	return bkt.Delete([]byte(ref))
+}
+
 func parseLeaseResource(r leases.Resource) ([]string, string, error) {
 	var (
 		ref  = r.ID
@@ -470,7 +537,8 @@ func parseLeaseResource(r leases.Resource) ([]string, string, error) {
 
 	switch k := keys[0]; k {
 	case string(bucketKeyObjectContent),
-		string(bucketKeyObjectIngests):
+		string(bucketKeyObjectIngests),
+		string(bucketKeyObjectImages):
 
 		if len(keys) != 1 {
 			return nil, "", fmt.Errorf("invalid resource type %s: %w", typ, errdefs.ErrInvalidArgument)
diff --git a/metadata/leases_test.go b/metadata/leases_test.go
index 56aa0d9f0..0cf3198b8 100644
--- a/metadata/leases_test.go
+++ b/metadata/leases_test.go
@@ -315,13 +315,11 @@ func TestLeaseResource(t *testing.T) {
 			err: errdefs.ErrNotImplemented,
 		},
 		{
-			// not allow to reference to image
 			lease: lease,
 			resource: leases.Resource{
 				ID:   "qBUHpWBn03YaCt9cL3PPGKWoxBqTlLfu",
-				Type: "image",
+				Type: "images",
 			},
-			err: errdefs.ErrNotImplemented,
 		},
 		{
 			lease: lease,