diff --git a/.gitignore b/.gitignore
index 6e359afa..cb4bc0e1 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,2 +1,5 @@
 bin/
 .vscode
+vendor/
+tmp_conv/
+tmp/
\ No newline at end of file
diff --git a/cmd/convertor/builder/builder.go b/cmd/convertor/builder/builder.go
index 7cb651ed..1412672e 100644
--- a/cmd/convertor/builder/builder.go
+++ b/cmd/convertor/builder/builder.go
@@ -134,6 +134,14 @@ func (b *overlaybdBuilder) Build(ctx context.Context) error {
 	alreadyConverted := make([]chan *v1.Descriptor, b.layers)
 	downloaded := make([]chan error, b.layers)
 	converted := make([]chan error, b.layers)
+
+	// check if manifest conversion result is already present in registry, if so, we can avoid conversion.
+	// when errors are encountered fallback to regular conversion
+	if convertedDesc, err := b.engine.CheckForConvertedManifest(ctx); err == nil && convertedDesc.Digest != "" {
+		logrus.Infof("Image found already converted in registry with digest %s", convertedDesc.Digest)
+		return nil
+	}
+
 	// Errgroups will close the context after wait returns so the operations need their own
 	// derived context.
 	g, rctx := errgroup.WithContext(ctx)
@@ -231,6 +239,7 @@ func (b *overlaybdBuilder) Build(ctx context.Context) error {
 	if err := b.engine.UploadImage(ctx); err != nil {
 		return errors.Wrap(err, "failed to upload manifest or config")
 	}
+	b.engine.StoreConvertedManifestDetails(ctx)
 	logrus.Info("convert finished")
 	return nil
 }
diff --git a/cmd/convertor/builder/builder_engine.go b/cmd/convertor/builder/builder_engine.go
index 81f74e91..4b6c5365 100644
--- a/cmd/convertor/builder/builder_engine.go
+++ b/cmd/convertor/builder/builder_engine.go
@@ -51,6 +51,15 @@ type builderEngine interface {
 	// UploadImage upload new manifest and config
 	UploadImage(ctx context.Context) error
 
+	// Cleanup removes workdir
+	Cleanup()
+
+	Deduplicateable
+}
+
+// Deduplicateable provides a number of functions to avoid duplicating work when converting images
+// It is used by the builderEngine to avoid re-converting layers and manifests
+type Deduplicateable interface {
 	// deduplication functions
 	// finds already converted layer in db and validates presence in registry
 	CheckForConvertedLayer(ctx context.Context, idx int) (specs.Descriptor, error)
@@ -58,14 +67,18 @@ type builderEngine interface {
 	// downloads the already converted layer
 	DownloadConvertedLayer(ctx context.Context, idx int, desc specs.Descriptor) error
 
-	// store chainID -> converted layer mapping for deduplication
+	// store chainID -> converted layer mapping for layer deduplication
 	StoreConvertedLayerDetails(ctx context.Context, idx int) error
 
-	// Cleanup removes workdir
-	Cleanup()
+	// store manifest digest -> converted manifest to avoid re-conversion
+	CheckForConvertedManifest(ctx context.Context) (specs.Descriptor, error)
+
+	// store manifest digest -> converted manifest to avoid re-conversion
+	StoreConvertedManifestDetails(ctx context.Context) error
 }
 
 type builderEngineBase struct {
+	resolver     remotes.Resolver
 	fetcher      remotes.Fetcher
 	pusher       remotes.Pusher
 	manifest     specs.Manifest
@@ -77,6 +90,8 @@ type builderEngineBase struct {
 	db           database.ConversionDatabase
 	host         string
 	repository   string
+	inputDesc    specs.Descriptor // original manifest descriptor
+	outputDesc   specs.Descriptor // converted manifest descriptor
 	reserve      bool
 	noUpload     bool
 	dumpManifest bool
@@ -172,6 +187,7 @@ func (e *builderEngineBase) uploadManifestAndConfig(ctx context.Context) error {
 		if err = uploadBytes(ctx, e.pusher, manifestDesc, cbuf); err != nil {
 			return errors.Wrapf(err, "failed to upload manifest")
 		}
+		e.outputDesc = manifestDesc
 		logrus.Infof("manifest uploaded")
 	}
 	if e.dumpManifest {
@@ -181,7 +197,6 @@ func (e *builderEngineBase) uploadManifestAndConfig(ctx context.Context) error {
 		}
 		logrus.Infof("manifest dumped")
 	}
-
 	return nil
 }
 
@@ -203,9 +218,11 @@ func getBuilderEngineBase(ctx context.Context, resolver remotes.Resolver, ref, t
 		return nil, errors.Wrap(err, "failed to fetch manifest and config")
 	}
 	return &builderEngineBase{
-		fetcher:  fetcher,
-		pusher:   pusher,
-		manifest: *manifest,
-		config:   *config,
+		resolver:  resolver,
+		fetcher:   fetcher,
+		pusher:    pusher,
+		manifest:  *manifest,
+		config:    *config,
+		inputDesc: desc,
 	}, nil
 }
diff --git a/cmd/convertor/builder/builder_test.go b/cmd/convertor/builder/builder_test.go
index 3d93e8cd..6118dba6 100644
--- a/cmd/convertor/builder/builder_test.go
+++ b/cmd/convertor/builder/builder_test.go
@@ -118,6 +118,20 @@ func (e *mockFuzzBuilderEngine) StoreConvertedLayerDetails(ctx context.Context,
 	return nil
 }
 
+func (e *mockFuzzBuilderEngine) CheckForConvertedManifest(ctx context.Context) (specs.Descriptor, error) {
+	if e.fixedRand.Float64() < failRate {
+		return specs.Descriptor{}, fmt.Errorf("random error on CheckForConvertedManifest")
+	}
+	return specs.Descriptor{}, nil
+}
+
+func (e *mockFuzzBuilderEngine) StoreConvertedManifestDetails(ctx context.Context) error {
+	if e.fixedRand.Float64() < failRate {
+		return fmt.Errorf("random error on StoreConvertedManifestDetails")
+	}
+	return nil
+}
+
 func (e *mockFuzzBuilderEngine) DownloadConvertedLayer(ctx context.Context, idx int, desc specs.Descriptor) error {
 	if e.fixedRand.Float64() < failRate {
 		return fmt.Errorf("random error on DownloadConvertedLayer")
diff --git a/cmd/convertor/builder/builder_utils.go b/cmd/convertor/builder/builder_utils.go
index 3010d9b8..6ff77684 100644
--- a/cmd/convertor/builder/builder_utils.go
+++ b/cmd/convertor/builder/builder_utils.go
@@ -217,12 +217,7 @@ func uploadBytes(ctx context.Context, pusher remotes.Pusher, desc specs.Descript
 		return err
 	}
 	defer cw.Close()
-
-	err = content.Copy(ctx, cw, bytes.NewReader(data), desc.Size, desc.Digest)
-	if err != nil {
-		return err
-	}
-	return nil
+	return content.Copy(ctx, cw, bytes.NewReader(data), desc.Size, desc.Digest)
 }
 
 func buildArchiveFromFiles(ctx context.Context, target string, compress compression.Compression, files ...string) error {
diff --git a/cmd/convertor/builder/builder_utils_test.go b/cmd/convertor/builder/builder_utils_test.go
index bf8ee9ac..3d38ba31 100644
--- a/cmd/convertor/builder/builder_utils_test.go
+++ b/cmd/convertor/builder/builder_utils_test.go
@@ -291,7 +291,7 @@ func Test_uploadBlob(t *testing.T) {
 	ctx := context.Background()
 	// Create a new inmemory registry to push to
 	reg := testingresources.GetTestRegistry(t, ctx, testingresources.RegistryOptions{
-		InmemoryOnly:              true,
+		InmemoryRegistryOnly:      true,
 		ManifestPushIgnoresLayers: false,
 	})
 
diff --git a/cmd/convertor/builder/overlaybd_builder.go b/cmd/convertor/builder/overlaybd_builder.go
index 28aa35e5..192f69e2 100644
--- a/cmd/convertor/builder/overlaybd_builder.go
+++ b/cmd/convertor/builder/overlaybd_builder.go
@@ -20,6 +20,7 @@ import (
 	"archive/tar"
 	"bufio"
 	"context"
+	"encoding/json"
 	"fmt"
 	"io"
 	"os"
@@ -180,8 +181,8 @@ func (e *overlaybdBuilderEngine) CheckForConvertedLayer(ctx context.Context, idx
 	}
 	chainID := e.overlaybdLayers[idx].chainID
 
-	// try to find in the same repo then check existence on registry
-	entry := e.db.GetEntryForRepo(ctx, e.host, e.repository, chainID)
+	// try to find the layer in the target repo
+	entry := e.db.GetLayerEntryForRepo(ctx, e.host, e.repository, chainID)
 	if entry != nil && entry.ChainID != "" {
 		desc := specs.Descriptor{
 			MediaType: e.mediaTypeImageLayer(),
@@ -197,15 +198,16 @@ func (e *overlaybdBuilderEngine) CheckForConvertedLayer(ctx context.Context, idx
 		}
 		if errdefs.IsNotFound(err) {
 			// invalid record in db, which is not found in registry, remove it
-			err := e.db.DeleteEntry(ctx, e.host, e.repository, chainID)
+			err := e.db.DeleteLayerEntry(ctx, e.host, e.repository, chainID)
 			if err != nil {
 				return specs.Descriptor{}, err
 			}
 		}
 	}
 
+	// fallback to a registry wide search
 	// found record in other repos, try mounting it to the target repo
-	entries := e.db.GetCrossRepoEntries(ctx, e.host, chainID)
+	entries := e.db.GetCrossRepoLayerEntries(ctx, e.host, chainID)
 	for _, entry := range entries {
 		desc := specs.Descriptor{
 			MediaType: e.mediaTypeImageLayer(),
@@ -220,7 +222,7 @@ func (e *overlaybdBuilderEngine) CheckForConvertedLayer(ctx context.Context, idx
 		if errdefs.IsAlreadyExists(err) {
 			desc.Annotations = nil
 
-			if err := e.db.CreateEntry(ctx, e.host, e.repository, entry.ConvertedDigest, chainID, entry.DataSize); err != nil {
+			if err := e.db.CreateLayerEntry(ctx, e.host, e.repository, entry.ConvertedDigest, chainID, entry.DataSize); err != nil {
 				continue // try a different repo if available
 			}
 
@@ -234,11 +236,125 @@ func (e *overlaybdBuilderEngine) CheckForConvertedLayer(ctx context.Context, idx
 	return specs.Descriptor{}, errdefs.ErrNotFound
 }
 
+// If manifest is already converted, avoid conversion. (e.g During tag reuse or cross repo mounts)
+// Note: This is output mediatype sensitive, if the manifest is converted to a different mediatype,
+// we will still convert it normally.
+func (e *overlaybdBuilderEngine) CheckForConvertedManifest(ctx context.Context) (specs.Descriptor, error) {
+	if e.db == nil {
+		return specs.Descriptor{}, errdefs.ErrNotFound
+	}
+
+	// try to find the manifest in the target repo
+	entry := e.db.GetManifestEntryForRepo(ctx, e.host, e.repository, e.mediaTypeManifest(), e.inputDesc.Digest)
+	if entry != nil && entry.ConvertedDigest != "" {
+		convertedDesc := specs.Descriptor{
+			MediaType: e.mediaTypeImageLayer(),
+			Digest:    entry.ConvertedDigest,
+			Size:      entry.DataSize,
+		}
+		rc, err := e.fetcher.Fetch(ctx, convertedDesc)
+		if err == nil {
+			rc.Close()
+			logrus.Infof("manifest %s found in remote with resulting digest %s", e.inputDesc.Digest, convertedDesc.Digest)
+			return convertedDesc, nil
+		}
+		if errdefs.IsNotFound(err) {
+			// invalid record in db, which is not found in registry, remove it
+			err := e.db.DeleteManifestEntry(ctx, e.host, e.repository, e.mediaTypeManifest(), e.inputDesc.Digest)
+			if err != nil {
+				return specs.Descriptor{}, err
+			}
+		}
+	}
+	// fallback to a registry wide search
+	entries := e.db.GetCrossRepoManifestEntries(ctx, e.host, e.mediaTypeManifest(), e.inputDesc.Digest)
+	for _, entry := range entries {
+		convertedDesc := specs.Descriptor{
+			MediaType: e.mediaTypeManifest(),
+			Digest:    entry.ConvertedDigest,
+			Size:      entry.DataSize,
+		}
+		fetcher, err := e.resolver.Fetcher(ctx, fmt.Sprintf("%s/%s@%s", entry.Host, entry.Repository, convertedDesc.Digest.String()))
+		if err != nil {
+			return specs.Descriptor{}, err
+		}
+		manifest, err := fetchManifest(ctx, fetcher, convertedDesc)
+		if err != nil {
+			if errdefs.IsNotFound(err) {
+				// invalid record in db, which is not found in registry, remove it
+				err := e.db.DeleteManifestEntry(ctx, entry.Host, entry.Repository, e.mediaTypeManifest(), e.inputDesc.Digest)
+				if err != nil {
+					return specs.Descriptor{}, err
+				}
+			}
+			continue
+		}
+		if err := e.mountImage(ctx, *manifest, convertedDesc, entry.Repository); err != nil {
+			continue // try a different repo if available
+		}
+		if err := e.db.CreateManifestEntry(ctx, e.host, e.repository, e.mediaTypeManifest(), e.inputDesc.Digest, convertedDesc.Digest, entry.DataSize); err != nil {
+			continue // try a different repo if available
+		}
+		logrus.Infof("manifest %s mount from %s was successful", convertedDesc.Digest, entry.Repository)
+		return convertedDesc, nil
+	}
+
+	logrus.Infof("manifest %s not found already converted in remote", e.inputDesc.Digest)
+	return specs.Descriptor{}, errdefs.ErrNotFound
+}
+
+// mountImage is responsible for mounting a specific manifest from a source repository, this includes
+// mounting all layers + config and then pushing the manifest.
+func (e *overlaybdBuilderEngine) mountImage(ctx context.Context, manifest specs.Manifest, desc specs.Descriptor, mountRepository string) error {
+	// Mount Config Blobs
+	config := manifest.Config
+	config.Annotations = map[string]string{
+		fmt.Sprintf("%s.%s", labelDistributionSource, e.host): mountRepository,
+	}
+	_, err := e.pusher.Push(ctx, config)
+	if errdefs.IsAlreadyExists(err) {
+		logrus.Infof("config blob mount from %s was successful", mountRepository)
+	} else if err != nil {
+		return fmt.Errorf("Failed to mount config blob from %s repository : %w", mountRepository, err)
+	}
+
+	// Mount Layer Blobs
+	for idx, layer := range manifest.Layers {
+		desc := layer
+		desc.Annotations = map[string]string{
+			fmt.Sprintf("%s.%s", labelDistributionSource, e.host): mountRepository,
+		}
+		_, err := e.pusher.Push(ctx, desc)
+		if errdefs.IsAlreadyExists(err) {
+			logrus.Infof("layer %d mount from %s was successful", idx, mountRepository)
+		} else if err != nil {
+			return fmt.Errorf("failed to mount all layers from %s repository : %w", mountRepository, err)
+		}
+	}
+
+	// Push Manifest
+	cbuf, err := json.Marshal(manifest)
+	if err != nil {
+		return err
+	}
+	return uploadBytes(ctx, e.pusher, desc, cbuf)
+}
+
+func (e *overlaybdBuilderEngine) StoreConvertedManifestDetails(ctx context.Context) error {
+	if e.db == nil {
+		return nil
+	}
+	if e.outputDesc.Digest == "" {
+		return errors.New("manifest is not yet converted")
+	}
+	return e.db.CreateManifestEntry(ctx, e.host, e.repository, e.mediaTypeManifest(), e.inputDesc.Digest, e.outputDesc.Digest, e.outputDesc.Size)
+}
+
 func (e *overlaybdBuilderEngine) StoreConvertedLayerDetails(ctx context.Context, idx int) error {
 	if e.db == nil {
 		return nil
 	}
-	return e.db.CreateEntry(ctx, e.host, e.repository, e.overlaybdLayers[idx].desc.Digest, e.overlaybdLayers[idx].chainID, e.overlaybdLayers[idx].desc.Size)
+	return e.db.CreateLayerEntry(ctx, e.host, e.repository, e.overlaybdLayers[idx].desc.Digest, e.overlaybdLayers[idx].chainID, e.overlaybdLayers[idx].desc.Size)
 }
 
 func (e *overlaybdBuilderEngine) DownloadConvertedLayer(ctx context.Context, idx int, desc specs.Descriptor) error {
diff --git a/cmd/convertor/builder/overlaybd_builder_test.go b/cmd/convertor/builder/overlaybd_builder_test.go
index 92c95682..4259e7d4 100644
--- a/cmd/convertor/builder/overlaybd_builder_test.go
+++ b/cmd/convertor/builder/overlaybd_builder_test.go
@@ -23,6 +23,7 @@ import (
 
 	testingresources "github.com/containerd/accelerated-container-image/cmd/convertor/testingresources"
 	"github.com/containerd/containerd/errdefs"
+	"github.com/containerd/containerd/images"
 	_ "github.com/containerd/containerd/pkg/testutil" // Handle custom root flag
 	"github.com/opencontainers/go-digest"
 	v1 "github.com/opencontainers/image-spec/specs-go/v1"
@@ -62,42 +63,163 @@ func Test_overlaybd_builder_CheckForConvertedLayer(t *testing.T) {
 	})
 
 	base.db = db
-
 	t.Run("No Entry in DB", func(t *testing.T) {
 		_, err := e.CheckForConvertedLayer(ctx, 0)
 		testingresources.Assert(t, errdefs.IsNotFound(err), fmt.Sprintf("CheckForConvertedLayer() returned an unexpected Error: %v", err))
 	})
 
-	err := base.db.CreateEntry(ctx, e.host, e.repository, targetDesc.Digest, fakeChainId, targetDesc.Size)
+	err := base.db.CreateLayerEntry(ctx, e.host, e.repository, targetDesc.Digest, fakeChainId, targetDesc.Size)
 	if err != nil {
-		t.Error(err)
+		t.Fatal(err)
 	}
 
-	t.Run("Entry in DB and in Registry", func(t *testing.T) {
+	t.Run("Layer entry in DB and in Registry", func(t *testing.T) {
 		desc, err := e.CheckForConvertedLayer(ctx, 0)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		testingresources.Assert(t, desc.Size == targetDesc.Size, "CheckForConvertedLayer() returned improper size layer")
+		testingresources.Assert(t, desc.Digest == targetDesc.Digest, "CheckForConvertedLayer() returned incorrect digest")
+	})
 
+	// cross repo mount (change target repo)
+	base.repository = "hello-world2"
+	newImageRef := "sample.localstore.io/hello-world2:amd64"
+	e.resolver = testingresources.GetTestResolver(t, ctx)
+	e.pusher = testingresources.GetTestPusherFromResolver(t, ctx, e.resolver, newImageRef)
+
+	t.Run("Cross repo layer entry found in DB mount", func(t *testing.T) {
+		desc, err := e.CheckForConvertedLayer(ctx, 0)
 		if err != nil {
-			t.Error(err)
+			t.Fatal(err)
 		}
 
 		testingresources.Assert(t, desc.Size == targetDesc.Size, "CheckForConvertedLayer() returned improper size layer")
 		testingresources.Assert(t, desc.Digest == targetDesc.Digest, "CheckForConvertedLayer() returned incorrect digest")
+
+		// check that the images can be pulled from the mounted repo
+		fetcher := testingresources.GetTestFetcherFromResolver(t, ctx, e.resolver, newImageRef)
+		rc, err := fetcher.Fetch(ctx, desc)
+		if err != nil {
+			t.Fatal(err)
+		}
+		rc.Close()
 	})
 
 	base.db = testingresources.NewLocalDB() // Reset DB
 	digestNotInRegistry := digest.FromString("Not in reg")
-	err = base.db.CreateEntry(ctx, e.host, e.repository, digestNotInRegistry, fakeChainId, 10)
+	err = base.db.CreateLayerEntry(ctx, e.host, e.repository, digestNotInRegistry, fakeChainId, 10)
 	if err != nil {
-		t.Error(err)
+		t.Fatal(err)
 	}
 
 	t.Run("Entry in DB but not in registry", func(t *testing.T) {
 		_, err := e.CheckForConvertedLayer(ctx, 0)
 		testingresources.Assert(t, errdefs.IsNotFound(err), fmt.Sprintf("CheckForConvertedLayer() returned an unexpected Error: %v", err))
-		entry := base.db.GetEntryForRepo(ctx, e.host, e.repository, fakeChainId)
+		entry := base.db.GetLayerEntryForRepo(ctx, e.host, e.repository, fakeChainId)
 		testingresources.Assert(t, entry == nil, "CheckForConvertedLayer() Invalid entry was not cleaned up")
 	})
-	// TODO: Cross Repo Mount Scenario
+}
+
+func Test_overlaybd_builder_CheckForConvertedManifest(t *testing.T) {
+	ctx := context.Background()
+	db := testingresources.NewLocalDB()
+	resolver := testingresources.GetTestResolver(t, ctx)
+	fetcher := testingresources.GetTestFetcherFromResolver(t, ctx, resolver, testingresources.DockerV2_Manifest_Simple_Ref)
+
+	// Unconverted hello world-image
+	inputDesc := v1.Descriptor{
+		MediaType: images.MediaTypeDockerSchema2Manifest,
+		Digest:    testingresources.DockerV2_Manifest_Simple_Digest,
+		Size:      testingresources.DockerV2_Manifest_Simple_Size,
+	}
+
+	// Converted hello world-image
+	outputDesc := v1.Descriptor{
+		MediaType: v1.MediaTypeImageManifest,
+		Digest:    testingresources.DockerV2_Manifest_Simple_Converted_Digest,
+		Size:      testingresources.DockerV2_Manifest_Simple_Converted_Size,
+	}
+
+	base := &builderEngineBase{
+		fetcher:    fetcher,
+		host:       "sample.localstore.io",
+		repository: "hello-world",
+		inputDesc:  inputDesc,
+		resolver:   resolver,
+		oci:        true,
+	}
+
+	e := &overlaybdBuilderEngine{
+		builderEngineBase: base,
+	}
+
+	t.Run("No DB Present", func(t *testing.T) {
+		_, err := e.CheckForConvertedManifest(ctx)
+		testingresources.Assert(t, errdefs.IsNotFound(err), fmt.Sprintf("CheckForConvertedManifest() returned an unexpected Error: %v", err))
+	})
+
+	base.db = db
+
+	// Store a fake converted manifest in the DB
+	err := base.db.CreateManifestEntry(ctx, e.host, e.repository, outputDesc.MediaType, inputDesc.Digest, outputDesc.Digest, outputDesc.Size)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	t.Run("Entry in DB and in Registry", func(t *testing.T) {
+		desc, err := e.CheckForConvertedManifest(ctx)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		testingresources.Assert(t, desc.Size == outputDesc.Size, "CheckForConvertedManifest() returned incorrect size")
+		testingresources.Assert(t, desc.Digest == outputDesc.Digest, "CheckForConvertedManifest() returned incorrect digest")
+	})
+
+	// cross repo mount (change target repo)
+	base.repository = "hello-world2"
+	newImageRef := "sample.localstore.io/hello-world2:amd64"
+	e.resolver = testingresources.GetTestResolver(t, ctx)
+	e.pusher = testingresources.GetTestPusherFromResolver(t, ctx, e.resolver, newImageRef)
+
+	t.Run("Cross Repo Entry found in DB mount", func(t *testing.T) {
+		_, err := e.CheckForConvertedManifest(ctx)
+		testingresources.Assert(t, err == nil, fmt.Sprintf("CheckForConvertedManifest() returned an unexpected Error: %v", err))
+		// check that the images can be pulled from the mounted repo
+		fetcher := testingresources.GetTestFetcherFromResolver(t, ctx, e.resolver, newImageRef)
+		_, desc, err := e.resolver.Resolve(ctx, newImageRef)
+		if err != nil {
+			t.Fatal(err)
+		}
+		manifest, config, err := fetchManifestAndConfig(ctx, fetcher, desc)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if manifest == nil || config == nil {
+			t.Fatalf("Could not pull mounted manifest or config")
+		}
+		rc, err := fetcher.Fetch(ctx, manifest.Layers[0])
+		if err != nil {
+			t.Fatal(err)
+		}
+		rc.Close()
+	})
+
+	base.db = testingresources.NewLocalDB() // Reset DB
+	digestNotInRegistry := digest.FromString("Not in reg")
+	err = base.db.CreateManifestEntry(ctx, e.host, e.repository, outputDesc.MediaType, inputDesc.Digest, digestNotInRegistry, outputDesc.Size)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	t.Run("Entry in DB but not in registry", func(t *testing.T) {
+		_, err := e.CheckForConvertedManifest(ctx)
+		testingresources.Assert(t, errdefs.IsNotFound(err), fmt.Sprintf("CheckForConvertedManifest() returned an unexpected Error: %v", err))
+		entry := base.db.GetManifestEntryForRepo(ctx, e.host, e.repository, outputDesc.MediaType, inputDesc.Digest)
+		testingresources.Assert(t, entry == nil, "CheckForConvertedManifest() Invalid entry was not cleaned up")
+	})
 }
 
 func Test_overlaybd_builder_StoreConvertedLayerDetails(t *testing.T) {
diff --git a/cmd/convertor/builder/turboOCI_builder.go b/cmd/convertor/builder/turboOCI_builder.go
index 984b442d..cbda5086 100644
--- a/cmd/convertor/builder/turboOCI_builder.go
+++ b/cmd/convertor/builder/turboOCI_builder.go
@@ -214,6 +214,16 @@ func (e *turboOCIBuilderEngine) DownloadConvertedLayer(ctx context.Context, idx
 	return errdefs.ErrNotImplemented
 }
 
+// DownloadConvertedLayer TODO
+func (e *turboOCIBuilderEngine) CheckForConvertedManifest(ctx context.Context) (specs.Descriptor, error) {
+	return specs.Descriptor{}, errdefs.ErrNotImplemented
+}
+
+// DownloadConvertedLayer TODO
+func (e *turboOCIBuilderEngine) StoreConvertedManifestDetails(ctx context.Context) error {
+	return errdefs.ErrNotImplemented
+}
+
 func (e *turboOCIBuilderEngine) Cleanup() {
 	os.RemoveAll(e.workDir)
 }
diff --git a/cmd/convertor/database/database.go b/cmd/convertor/database/database.go
index a9f4fa5c..045390e4 100644
--- a/cmd/convertor/database/database.go
+++ b/cmd/convertor/database/database.go
@@ -23,16 +23,32 @@ import (
 )
 
 type ConversionDatabase interface {
-	GetEntryForRepo(ctx context.Context, host string, repository string, chainID string) *Entry
-	GetCrossRepoEntries(ctx context.Context, host string, chainID string) []*Entry
-	CreateEntry(ctx context.Context, host string, repository string, convertedDigest digest.Digest, chainID string, size int64) error
-	DeleteEntry(ctx context.Context, host string, repository string, chainID string) error
+	// Layer Entries
+	CreateLayerEntry(ctx context.Context, host, repository string, convertedDigest digest.Digest, chainID string, size int64) error
+	GetLayerEntryForRepo(ctx context.Context, host, repository, chainID string) *LayerEntry
+	GetCrossRepoLayerEntries(ctx context.Context, host, chainID string) []*LayerEntry
+	DeleteLayerEntry(ctx context.Context, host, repository, chainID string) error
+
+	// Manifest Entries
+	CreateManifestEntry(ctx context.Context, host, repository, mediatype string, original, convertedDigest digest.Digest, size int64) error
+	GetManifestEntryForRepo(ctx context.Context, host, repository, mediatype string, original digest.Digest) *ManifestEntry
+	GetCrossRepoManifestEntries(ctx context.Context, host, mediatype string, original digest.Digest) []*ManifestEntry
+	DeleteManifestEntry(ctx context.Context, host, repository, mediatype string, original digest.Digest) error
 }
 
-type Entry struct {
+type LayerEntry struct {
 	ConvertedDigest digest.Digest
 	DataSize        int64
 	Repository      string
 	ChainID         string
 	Host            string
 }
+
+type ManifestEntry struct {
+	ConvertedDigest digest.Digest
+	OriginalDigest  digest.Digest
+	DataSize        int64
+	Repository      string
+	Host            string
+	MediaType       string
+}
diff --git a/cmd/convertor/database/mysql.go b/cmd/convertor/database/mysql.go
index 1db4375b..30663371 100644
--- a/cmd/convertor/database/mysql.go
+++ b/cmd/convertor/database/mysql.go
@@ -19,10 +19,9 @@ package database
 import (
 	"context"
 	"database/sql"
+	"fmt"
 
 	"github.com/containerd/containerd/log"
-	"github.com/pkg/errors"
-
 	"github.com/opencontainers/go-digest"
 )
 
@@ -36,18 +35,21 @@ func NewSqlDB(db *sql.DB) ConversionDatabase {
 	}
 }
 
-func (m *sqldb) GetEntryForRepo(ctx context.Context, host string, repository string, chainID string) *Entry {
-	var entry Entry
+func (m *sqldb) CreateLayerEntry(ctx context.Context, host, repository string, convertedDigest digest.Digest, chainID string, size int64) error {
+	_, err := m.db.ExecContext(ctx, "insert into overlaybd_layers(host, repo, chain_id, data_digest, data_size) values(?, ?, ?, ?, ?)", host, repository, chainID, convertedDigest, size)
+	return err
+}
 
+func (m *sqldb) GetLayerEntryForRepo(ctx context.Context, host, repository, chainID string) *LayerEntry {
+	var entry LayerEntry
 	row := m.db.QueryRowContext(ctx, "select host, repo, chain_id, data_digest, data_size from overlaybd_layers where host=? and repo=? and chain_id=?", host, repository, chainID)
 	if err := row.Scan(&entry.Host, &entry.Repository, &entry.ChainID, &entry.ConvertedDigest, &entry.DataSize); err != nil {
 		return nil
 	}
-
 	return &entry
 }
 
-func (m *sqldb) GetCrossRepoEntries(ctx context.Context, host string, chainID string) []*Entry {
+func (m *sqldb) GetCrossRepoLayerEntries(ctx context.Context, host, chainID string) []*LayerEntry {
 	rows, err := m.db.QueryContext(ctx, "select host, repo, chain_id, data_digest, data_size from overlaybd_layers where host=? and chain_id=?", host, chainID)
 	if err != nil {
 		if err == sql.ErrNoRows {
@@ -56,9 +58,9 @@ func (m *sqldb) GetCrossRepoEntries(ctx context.Context, host string, chainID st
 		log.G(ctx).Infof("query error %v", err)
 		return nil
 	}
-	var entries []*Entry
+	var entries []*LayerEntry
 	for rows.Next() {
-		var entry Entry
+		var entry LayerEntry
 		err = rows.Scan(&entry.Host, &entry.Repository, &entry.ChainID, &entry.ConvertedDigest, &entry.DataSize)
 		if err != nil {
 			continue
@@ -69,15 +71,54 @@ func (m *sqldb) GetCrossRepoEntries(ctx context.Context, host string, chainID st
 	return entries
 }
 
-func (m *sqldb) CreateEntry(ctx context.Context, host string, repository string, convertedDigest digest.Digest, chainID string, size int64) error {
-	_, err := m.db.ExecContext(ctx, "insert into overlaybd_layers(host, repo, chain_id, data_digest, data_size) values(?, ?, ?, ?, ?)", host, repository, chainID, convertedDigest, size)
+func (m *sqldb) DeleteLayerEntry(ctx context.Context, host, repository string, chainID string) error {
+	_, err := m.db.Exec("delete from overlaybd_layers where host=? and repo=? and chain_id=?", host, repository, chainID)
+	if err != nil {
+		return fmt.Errorf("failed to remove invalid record in db: %w", err)
+	}
+	return nil
+}
+
+func (m *sqldb) CreateManifestEntry(ctx context.Context, host, repository, mediaType string, original, convertedDigest digest.Digest, size int64) error {
+	_, err := m.db.ExecContext(ctx, "insert into overlaybd_manifests(host, repo, src_digest, out_digest, data_size, mediatype) values(?, ?, ?, ?, ?, ?)", host, repository, original, convertedDigest, size, mediaType)
 	return err
 }
 
-func (m *sqldb) DeleteEntry(ctx context.Context, host string, repository string, chainID string) error {
-	_, err := m.db.Exec("delete from overlaybd_layers where host=? and repo=? and chain_id=?", host, repository, chainID)
+func (m *sqldb) GetManifestEntryForRepo(ctx context.Context, host, repository, mediaType string, original digest.Digest) *ManifestEntry {
+	var entry ManifestEntry
+	row := m.db.QueryRowContext(ctx, "select host, repo, src_digest, out_digest, data_size, mediatype from overlaybd_manifests where host=? and repo=? and src_digest=? and mediatype=?", host, repository, original, mediaType)
+	if err := row.Scan(&entry.Host, &entry.Repository, &entry.OriginalDigest, &entry.ConvertedDigest, &entry.DataSize, &entry.MediaType); err != nil {
+		return nil
+	}
+	return &entry
+}
+
+func (m *sqldb) GetCrossRepoManifestEntries(ctx context.Context, host, mediaType string, original digest.Digest) []*ManifestEntry {
+	rows, err := m.db.QueryContext(ctx, "select host, repo, src_digest, out_digest, data_size, mediatype from overlaybd_manifests where host=? and src_digest=? and mediatype=?", host, original, mediaType)
+	if err != nil {
+		if err == sql.ErrNoRows {
+			return nil
+		}
+		log.G(ctx).Infof("query error %v", err)
+		return nil
+	}
+	var entries []*ManifestEntry
+	for rows.Next() {
+		var entry ManifestEntry
+		err = rows.Scan(&entry.Host, &entry.Repository, &entry.OriginalDigest, &entry.ConvertedDigest, &entry.DataSize, &entry.MediaType)
+		if err != nil {
+			continue
+		}
+		entries = append(entries, &entry)
+	}
+
+	return entries
+}
+
+func (m *sqldb) DeleteManifestEntry(ctx context.Context, host, repository, mediaType string, original digest.Digest) error {
+	_, err := m.db.Exec("delete from overlaybd_manifests where host=? and repo=? and src_digest=? and mediatype=?", host, repository, original, mediaType)
 	if err != nil {
-		return errors.Wrapf(err, "failed to remove invalid record in db")
+		return fmt.Errorf("failed to remove invalid record in db: %w", err)
 	}
 	return nil
 }
diff --git a/cmd/convertor/resources/samples/mysql-db-manifest-cache-sample-workload.sh b/cmd/convertor/resources/samples/mysql-db-manifest-cache-sample-workload.sh
new file mode 100755
index 00000000..50318507
--- /dev/null
+++ b/cmd/convertor/resources/samples/mysql-db-manifest-cache-sample-workload.sh
@@ -0,0 +1,21 @@
+# Validation Examples
+registry=$1 # registry to push to
+username=$2 # username for registry
+password=$3 # password for registry
+sourceImage=$4 # public image to convert
+repository=$5 # repository to push to
+tag=$6 # tag to push to
+mysqldbuser=$7 # mysql user
+mysqldbpassword=$8 # mysql password
+
+oras login $registry -u $username -p $password
+oras cp $sourceImage $registry/$repository:$tag
+# Try one conversion
+./bin/convertor --repository $registry/$repository -u $username:$password --input-tag $tag --oci --overlaybd $tag-obd-cache --db-str "$mysqldbuser:mysqldbpassword@tcp(127.0.0.1:3306)/conversioncache" --db-type mysql
+
+# Retry, result manifest should be cached
+./bin/convertor --repository $registry/$repository -u $username:$password --input-tag $tag --oci --overlaybd $tag-obd-cache-2 --db-str "$mysqldbuser:mysqldbpassword@tcp(127.0.0.1:3306)/conversioncache" --db-type mysql
+
+# Retry, cross repo mount
+oras cp $sourceImage $registry/$repository-2:$tag
+./bin/convertor --repository $registry/$repository -u $username:$password --input-tag $tag --oci --overlaybd $tag-obd-cache-2 --db-str "$mysqldbuser:mysqldbpassword@tcp(127.0.0.1:3306)/conversioncache" --db-type mysql
diff --git a/cmd/convertor/resources/samples/mysql-db-setup.sh b/cmd/convertor/resources/samples/mysql-db-setup.sh
new file mode 100755
index 00000000..c3d00ab0
--- /dev/null
+++ b/cmd/convertor/resources/samples/mysql-db-setup.sh
@@ -0,0 +1,11 @@
+#!/bin/sh
+# Reference Script for getting started with mysql DB for userspace convertor
+mysqldbuser=$1
+mysqldbpassword=$2
+
+# Set up mysql
+apt update
+apt install mysql-server
+service mysql start
+cat ./mysql.conf | mysql
+echo "CREATE USER '$mysqldbuser'@'localhost' IDENTIFIED BY '$mysqldbpassword'; GRANT ALL PRIVILEGES ON conversioncache.* TO '$mysqldbuser'@'localhost'; FLUSH PRIVILEGES;" | mysql
\ No newline at end of file
diff --git a/cmd/convertor/resources/samples/mysql.conf b/cmd/convertor/resources/samples/mysql.conf
new file mode 100644
index 00000000..63a86648
--- /dev/null
+++ b/cmd/convertor/resources/samples/mysql.conf
@@ -0,0 +1,22 @@
+CREATE database conversioncache;
+USE conversioncache;
+CREATE TABLE `overlaybd_layers` (
+  `host` varchar(255) NOT NULL,
+  `repo` varchar(255) NOT NULL,
+  `chain_id` varchar(255) NOT NULL COMMENT 'chain-id of the normal image layer',
+  `data_digest` varchar(255) NOT NULL COMMENT 'digest of overlaybd layer',
+  `data_size` bigint(20) NOT NULL COMMENT 'size of overlaybd layer',
+  PRIMARY KEY (`host`,`repo`,`chain_id`),
+  KEY `index_registry_chainId` (`host`,`chain_id`) USING BTREE
+) DEFAULT CHARSET=utf8;
+
+CREATE TABLE `overlaybd_manifests` (
+  `host` varchar(255) NOT NULL,
+  `repo` varchar(255) NOT NULL,
+  `src_digest` varchar(255) NOT NULL COMMENT 'digest of the normal image manifest',
+  `out_digest` varchar(255) NOT NULL COMMENT 'digest of overlaybd manifest',
+  `data_size` bigint(20) NOT NULL COMMENT 'size of overlaybd manifest',
+  `mediatype` varchar(255) NOT NULL COMMENT 'mediatype of the converted image manifest',
+  PRIMARY KEY (`host`,`repo`,`src_digest`, `mediatype`),
+  KEY `index_registry_src_digest` (`host`,`src_digest`, `mediatype`) USING BTREE
+) DEFAULT CHARSET=utf8;
\ No newline at end of file
diff --git a/cmd/convertor/resources/samples/run-userspace-convertor-ubuntu.Dockerfile b/cmd/convertor/resources/samples/run-userspace-convertor-ubuntu.Dockerfile
new file mode 100644
index 00000000..4fb6f778
--- /dev/null
+++ b/cmd/convertor/resources/samples/run-userspace-convertor-ubuntu.Dockerfile
@@ -0,0 +1,63 @@
+#NOTE: This Dockerfile should be executed from the root of the repository
+# docker build . -f ./cmd/convertor/resources/samples/run-userspace-convertor-ubuntu.Dockerfile -t convertor
+
+FROM ubuntu:latest AS base
+# Required Build/Run Tools Dependencies for Overlaybd tools
+RUN apt-get update && \
+    apt-get install -y ca-certificates && \
+    update-ca-certificates
+
+RUN apt update && \
+    apt install -y libcurl4-openssl-dev libext2fs-dev libaio-dev mysql-server
+
+# --- OVERLAYBD TOOLS ---
+FROM base As overlaybd-build
+RUN apt update && \
+    apt install -y libgflags-dev libssl-dev libnl-3-dev libnl-genl-3-dev libzstd-dev && \
+    apt install -y zlib1g-dev binutils make git wget sudo tar gcc cmake build-essential g++ && \
+    apt install -y uuid-dev libjson-c-dev libkmod-dev libsystemd-dev autoconf automake libtool libpci-dev nasm && \
+    apt install -y pkg-config
+
+# Download and install Golang version 1.21
+RUN wget https://go.dev/dl/go1.20.12.linux-amd64.tar.gz && \
+    tar -C /usr/local -xzf go1.20.12.linux-amd64.tar.gz && \
+    rm go1.20.12.linux-amd64.tar.gz
+
+# Set environment variables
+ENV PATH="/usr/local/go/bin:${PATH}"
+ENV GOPATH="/go"
+
+RUN git clone https://github.com/containerd/overlaybd.git && \
+    cd overlaybd && \
+    git submodule update --init && \
+    mkdir build && \
+    cd build && \
+    cmake ..  -DCMAKE_BUILD_TYPE=Release -DBUILD_TESTING=0 -DENABLE_DSA=0 -DENABLE_ISAL=0 && \
+    make -j8 && \
+    make install
+
+# --- BUILD LOCAL CONVERTER ---
+FROM overlaybd-build AS convert-build
+WORKDIR /home/limiteduser/accelerated-container-image
+COPY . .
+WORKDIR /home/limiteduser/accelerated-container-image
+RUN make
+
+# --- FINAL ---
+FROM base
+WORKDIR /home/limiteduser/
+
+# Copy Conversion Tools
+COPY --from=overlaybd-build /opt/overlaybd/bin /opt/overlaybd/bin
+COPY --from=overlaybd-build /opt/overlaybd/lib /opt/overlaybd/lib
+COPY --from=overlaybd-build /opt/overlaybd/baselayers /opt/overlaybd/baselayers
+
+# This is necessary for overlaybd_apply to work
+COPY --from=overlaybd-build /etc/overlaybd/overlaybd.json /etc/overlaybd/overlaybd.json
+COPY --from=convert-build /home/limiteduser/accelerated-container-image/bin/convertor ./bin/convertor
+
+# Useful resources
+COPY cmd/convertor/resources/samples/mysql.conf ./mysql.conf
+COPY cmd/convertor/resources/samples/mysql-db-setup.sh ./mysql-db-setup.sh
+COPY cmd/convertor/resources/samples/mysql-db-manifest-cache-sample-workload.sh ./mysql-db-manifest-cache-sample-workload.sh
+CMD ["./bin/convertor"]
\ No newline at end of file
diff --git a/cmd/convertor/testingresources/consts.go b/cmd/convertor/testingresources/consts.go
index 50b24618..1f2a9035 100644
--- a/cmd/convertor/testingresources/consts.go
+++ b/cmd/convertor/testingresources/consts.go
@@ -47,6 +47,11 @@ const (
 	DockerV2_Manifest_Simple_Layer_0_Digest = "sha256:719385e32844401d57ecfd3eacab360bf551a1491c05b85806ed8f1b08d792f6"
 	DockerV2_Manifest_Simple_Layer_0_Size   = 2457
 
+	// DOCKER V2 (amd64-converted) - overlaybd
+	DockerV2_Manifest_Simple_Converted_Ref    = "sample.localstore.io/hello-world:amd64-converted"
+	DockerV2_Manifest_Simple_Converted_Digest = "sha256:42caa56a19e082b872d43f645bb392e25c9e78bce429755bd709fac598265f88"
+	DockerV2_Manifest_Simple_Converted_Size   = 641
+
 	// DOCKER MANIFEST LIST
 	Docker_Manifest_List_Ref    = "sample.localstore.io/hello-world:docker-list"
 	Docker_Manifest_List_Digest = "sha256:726023f73a8fc5103fa6776d48090539042cb822531c6b751b1f6dd18cb5705d"
diff --git a/cmd/convertor/testingresources/local_db.go b/cmd/convertor/testingresources/local_db.go
index e1cd9440..b442852a 100644
--- a/cmd/convertor/testingresources/local_db.go
+++ b/cmd/convertor/testingresources/local_db.go
@@ -18,14 +18,17 @@ package testingresources
 
 import (
 	"context"
+	"sync"
 
 	"github.com/containerd/accelerated-container-image/cmd/convertor/database"
 	"github.com/opencontainers/go-digest"
-	"github.com/pkg/errors"
 )
 
 type localdb struct {
-	records []*database.Entry
+	layerRecords    []*database.LayerEntry
+	manifestRecords []*database.ManifestEntry
+	layerLock       sync.Mutex // Protects layerRecords
+	manifestLock    sync.Mutex // Protects manifestRecords
 }
 
 // NewLocalDB returns a new local database for testing. This is a simple unoptimized in-memory database.
@@ -33,8 +36,23 @@ func NewLocalDB() database.ConversionDatabase {
 	return &localdb{}
 }
 
-func (l *localdb) GetEntryForRepo(ctx context.Context, host string, repository string, chainID string) *database.Entry {
-	for _, entry := range l.records {
+func (l *localdb) CreateLayerEntry(ctx context.Context, host string, repository string, convertedDigest digest.Digest, chainID string, size int64) error {
+	l.layerLock.Lock()
+	defer l.layerLock.Unlock()
+	l.layerRecords = append(l.layerRecords, &database.LayerEntry{
+		Host:            host,
+		Repository:      repository,
+		ChainID:         chainID,
+		ConvertedDigest: convertedDigest,
+		DataSize:        size,
+	})
+	return nil
+}
+
+func (l *localdb) GetLayerEntryForRepo(ctx context.Context, host string, repository string, chainID string) *database.LayerEntry {
+	l.layerLock.Lock()
+	defer l.layerLock.Unlock()
+	for _, entry := range l.layerRecords {
 		if entry.Host == host && entry.ChainID == chainID && entry.Repository == repository {
 			return entry
 		}
@@ -42,9 +60,11 @@ func (l *localdb) GetEntryForRepo(ctx context.Context, host string, repository s
 	return nil
 }
 
-func (l *localdb) GetCrossRepoEntries(ctx context.Context, host string, chainID string) []*database.Entry {
-	var entries []*database.Entry
-	for _, entry := range l.records {
+func (l *localdb) GetCrossRepoLayerEntries(ctx context.Context, host, chainID string) []*database.LayerEntry {
+	l.layerLock.Lock()
+	defer l.layerLock.Unlock()
+	var entries []*database.LayerEntry
+	for _, entry := range l.layerRecords {
 		if entry.Host == host && entry.ChainID == chainID {
 			entries = append(entries, entry)
 		}
@@ -52,33 +72,64 @@ func (l *localdb) GetCrossRepoEntries(ctx context.Context, host string, chainID
 	return entries
 }
 
-func (l *localdb) CreateEntry(ctx context.Context, host string, repository string, convertedDigest digest.Digest, chainID string, size int64) error {
-	l.records = append(l.records, &database.Entry{
+func (l *localdb) DeleteLayerEntry(ctx context.Context, host, repository, chainID string) error {
+	l.layerLock.Lock()
+	defer l.layerLock.Unlock()
+	// host - repo - chainID should be unique
+	for i, entry := range l.layerRecords {
+		if entry.Host == host && entry.ChainID == chainID && entry.Repository == repository {
+			l.layerRecords = append(l.layerRecords[:i], l.layerRecords[i+1:]...)
+			return nil
+		}
+	}
+	return nil // No error if entry not found
+}
+
+func (l *localdb) CreateManifestEntry(ctx context.Context, host, repository, mediaType string, original, convertedDigest digest.Digest, size int64) error {
+	l.manifestLock.Lock()
+	defer l.manifestLock.Unlock()
+	l.manifestRecords = append(l.manifestRecords, &database.ManifestEntry{
 		Host:            host,
 		Repository:      repository,
-		ChainID:         chainID,
+		OriginalDigest:  original,
 		ConvertedDigest: convertedDigest,
 		DataSize:        size,
+		MediaType:       mediaType,
 	})
 	return nil
 }
 
-func (l *localdb) DeleteEntry(ctx context.Context, host string, repository string, chainID string) error {
-	// Identify indices of items to be deleted.
-	var indicesToDelete []int
-	for i, entry := range l.records {
-		if entry.Host == host && entry.ChainID == chainID && entry.Repository == repository {
-			indicesToDelete = append(indicesToDelete, i)
+func (l *localdb) GetManifestEntryForRepo(ctx context.Context, host, repository, mediaType string, original digest.Digest) *database.ManifestEntry {
+	l.manifestLock.Lock()
+	defer l.manifestLock.Unlock()
+	for _, entry := range l.manifestRecords {
+		if entry.Host == host && entry.OriginalDigest == original && entry.Repository == repository && entry.MediaType == mediaType {
+			return entry
 		}
 	}
+	return nil
+}
 
-	if len(indicesToDelete) == 0 {
-		return errors.Errorf("failed to find entry for host %s, repository %s, chainID %s", host, repository, chainID)
+func (l *localdb) GetCrossRepoManifestEntries(ctx context.Context, host, mediaType string, original digest.Digest) []*database.ManifestEntry {
+	l.manifestLock.Lock()
+	defer l.manifestLock.Unlock()
+	var entries []*database.ManifestEntry
+	for _, entry := range l.manifestRecords {
+		if entry.Host == host && entry.OriginalDigest == original && entry.MediaType == mediaType {
+			entries = append(entries, entry)
+		}
 	}
+	return entries
+}
 
-	// Delete items at identified indices. (Reverse order to avoid index shifting.)
-	for i := len(indicesToDelete) - 1; i >= 0; i-- {
-		l.records = append(l.records[:indicesToDelete[i]], l.records[indicesToDelete[i]+1:]...)
+func (l *localdb) DeleteManifestEntry(ctx context.Context, host, repository, mediaType string, original digest.Digest) error {
+	l.manifestLock.Lock()
+	defer l.manifestLock.Unlock()
+	// Identify indices of items to be deleted.
+	for i, entry := range l.manifestRecords {
+		if entry.Host == host && entry.OriginalDigest == original && entry.Repository == repository && entry.MediaType == mediaType {
+			l.manifestRecords = append(l.manifestRecords[:i], l.manifestRecords[i+1:]...)
+		}
 	}
-	return nil
+	return nil // No error if entry not found
 }
diff --git a/cmd/convertor/testingresources/local_registry.go b/cmd/convertor/testingresources/local_registry.go
index d607fd92..ae4ff61e 100644
--- a/cmd/convertor/testingresources/local_registry.go
+++ b/cmd/convertor/testingresources/local_registry.go
@@ -42,7 +42,7 @@ type TestRegistry struct {
 }
 
 type RegistryOptions struct {
-	InmemoryOnly              bool   // Specifies if the registry should not load any resources from storage
+	InmemoryRegistryOnly      bool   // Specifies if the registry should not load any resources from storage
 	LocalRegistryPath         string // Specifies the path to the local registry
 	ManifestPushIgnoresLayers bool   // Specifies if the registry should require layers to be pushed before manifest
 }
@@ -54,7 +54,7 @@ func NewTestRegistry(ctx context.Context, opts RegistryOptions) (*TestRegistry,
 		internalRegistry: make(internalRegistry),
 		opts:             opts,
 	}
-	if !opts.InmemoryOnly {
+	if !opts.InmemoryRegistryOnly {
 		files, err := os.ReadDir(opts.LocalRegistryPath)
 		if err != nil {
 			return nil, err
@@ -103,8 +103,9 @@ func (r *TestRegistry) Push(ctx context.Context, repository string, tag string,
 		return repo.Push(ctx, descriptor, tag, content)
 	}
 
-	// If the repository does not exist we create a new one
+	// If the repository does not exist we create a new one inmemory
 	repo = NewRepoStore(ctx, repository, &r.opts)
+	repo.inmemoryRepoOnly = true
 	r.internalRegistry[repository] = repo
 	repo.Push(ctx, descriptor, tag, content)
 
@@ -137,9 +138,20 @@ func (r *TestRegistry) Exists(ctx context.Context, repository string, tag string
 		// we don't need to check if the digest exists in the repo stores.
 		return true, nil
 	default:
-		if tag != "" {
-			return false, errors.New("Tag specified for non manifest")
-		}
 		return repo.Exists(ctx, desc)
 	}
 }
+
+// Mount simulates a cross repo mount by copying blobs from srcRepository to targetRepository
+func (r *TestRegistry) Mount(ctx context.Context, srcRepository string, targetRepository string, desc v1.Descriptor) error {
+	rd, err := r.Fetch(ctx, srcRepository, desc)
+	if err != nil {
+		return err
+	}
+	defer rd.Close()
+	var body []byte
+	if body, err = io.ReadAll(rd); err != nil {
+		return err
+	}
+	return r.Push(ctx, targetRepository, "", desc, body)
+}
diff --git a/cmd/convertor/testingresources/local_remotes.go b/cmd/convertor/testingresources/local_remotes.go
index 73d94f7d..25ebccb1 100644
--- a/cmd/convertor/testingresources/local_remotes.go
+++ b/cmd/convertor/testingresources/local_remotes.go
@@ -31,6 +31,10 @@ import (
 	v1 "github.com/opencontainers/image-spec/specs-go/v1"
 )
 
+const (
+	labelDistributionSource = "containerd.io/distribution.source"
+)
+
 // RESOLVER
 type MockLocalResolver struct {
 	testReg *TestRegistry
@@ -39,7 +43,7 @@ type MockLocalResolver struct {
 func NewMockLocalResolver(ctx context.Context, localRegistryPath string) (*MockLocalResolver, error) {
 	reg, err := NewTestRegistry(ctx, RegistryOptions{
 		LocalRegistryPath:         localRegistryPath,
-		InmemoryOnly:              false,
+		InmemoryRegistryOnly:      false,
 		ManifestPushIgnoresLayers: false,
 	})
 	if err != nil {
@@ -78,7 +82,7 @@ func (r *MockLocalResolver) Fetcher(ctx context.Context, ref string) (remotes.Fe
 }
 
 func (r *MockLocalResolver) Pusher(ctx context.Context, ref string) (remotes.Pusher, error) {
-	_, repository, tag, err := ParseRef(ctx, ref)
+	host, repository, tag, err := ParseRef(ctx, ref)
 	if err != nil {
 		return nil, err
 	}
@@ -87,6 +91,7 @@ func (r *MockLocalResolver) Pusher(ctx context.Context, ref string) (remotes.Pus
 		testReg:    r.testReg,
 		repository: repository,
 		tag:        tag,
+		host:       host,
 		tracker:    docker.NewInMemoryTracker(),
 	}, nil
 }
@@ -106,6 +111,7 @@ type MockLocalPusher struct {
 	testReg    *TestRegistry
 	repository string
 	tag        string
+	host       string
 	tracker    docker.StatusTracker
 }
 
@@ -150,6 +156,15 @@ func (p MockLocalPusher) push(ctx context.Context, desc v1.Descriptor, ref strin
 		return nil, errdefs.ErrAlreadyExists
 	}
 
+	// Layer mounts
+	if mountRepo, ok := desc.Annotations[fmt.Sprintf("%s.%s", labelDistributionSource, p.host)]; ok {
+		err = p.testReg.Mount(ctx, mountRepo, p.repository, desc)
+		if err != nil {
+			return nil, err
+		}
+		return nil, errdefs.ErrAlreadyExists
+	}
+
 	respC := make(chan error, 1)
 
 	p.tracker.SetStatus(ref, docker.Status{
diff --git a/cmd/convertor/testingresources/local_repo.go b/cmd/convertor/testingresources/local_repo.go
index a1e1a45c..3244da43 100644
--- a/cmd/convertor/testingresources/local_repo.go
+++ b/cmd/convertor/testingresources/local_repo.go
@@ -34,10 +34,11 @@ import (
 
 // REPOSITORY
 type RepoStore struct {
-	path         string
-	fileStore    *oci.Store
-	inmemoryRepo *inmemoryRepo
-	opts         *RegistryOptions
+	path             string
+	inmemoryRepoOnly bool
+	fileStore        *oci.Store
+	inmemoryRepo     *inmemoryRepo
+	opts             *RegistryOptions
 }
 
 type inmemoryRepo struct {
@@ -68,8 +69,8 @@ func (r *RepoStore) LoadStore(ctx context.Context) error {
 		return nil
 	}
 
-	if r.opts.InmemoryOnly {
-		return errors.New("LoadStore should not be invoked if registry is memory only")
+	if r.opts.InmemoryRegistryOnly && !r.inmemoryRepoOnly {
+		return errors.New("LoadStore should not be invoked if registry or repo is memory only")
 	}
 
 	if r.path == "" {
@@ -100,7 +101,7 @@ func (r *RepoStore) Resolve(ctx context.Context, tag string) (v1.Descriptor, err
 		}
 	}
 
-	if !r.opts.InmemoryOnly {
+	if !r.opts.InmemoryRegistryOnly && !r.inmemoryRepoOnly {
 		if err := r.LoadStore(ctx); err != nil {
 			return v1.Descriptor{}, err
 		}
@@ -116,7 +117,7 @@ func (r *RepoStore) Fetch(ctx context.Context, descriptor v1.Descriptor) (io.Rea
 		return io.NopCloser(bytes.NewReader(blob)), nil
 	}
 
-	if !r.opts.InmemoryOnly {
+	if !r.opts.InmemoryRegistryOnly && !r.inmemoryRepoOnly {
 		if err := r.LoadStore(ctx); err != nil {
 			return nil, err
 		}
@@ -140,7 +141,7 @@ func (r *RepoStore) Exists(ctx context.Context, descriptor v1.Descriptor) (bool,
 		return true, nil
 	}
 
-	if !r.opts.InmemoryOnly {
+	if !r.opts.InmemoryRegistryOnly && !r.inmemoryRepoOnly {
 		if err := r.LoadStore(ctx); err != nil {
 			return false, err
 		}
diff --git a/cmd/convertor/testingresources/mocks/generate.sh b/cmd/convertor/testingresources/mocks/generate.sh
index 119d1c8b..d72e85ae 100755
--- a/cmd/convertor/testingresources/mocks/generate.sh
+++ b/cmd/convertor/testingresources/mocks/generate.sh
@@ -1,20 +1,33 @@
-# Usage: ./generate.sh <registry>
+# Usage: ./generate.sh $registry $username $password
 # Prerequisites: oras
 # Generates simple hello-world images in ./registry folder
 # This script serves as a way to regenerate the images in the registry folder if necessary
 # and to document the taken steps to generate the test registry. Add more if new images are needed.
+# do note that it might be necessary to adjust the consts.go file to make sure tests don't break if
+# the source hello-world images are updated.
 
+registry=$1
+username=$2
+password=$3
 srcTag="linux"
 srcRepo="hello-world"
 srcImage="docker.io/library/$srcRepo:$srcTag"
 srcRegistry="docker.io/library"
-registry=$1
 destFolder="./registry"
 
 echo "Begin image generation based on src image: $srcImage"
-
-# Docker
 oras cp --to-oci-layout $srcImage $destFolder/hello-world:docker-list
-# Tag Some submanifests
+
+# Tag some submanifests
 oras cp --to-oci-layout --platform linux/arm64 $srcRegistry/hello-world:linux $destFolder/hello-world/:arm64
-oras cp --to-oci-layout --platform linux/amd64 $srcRegistry/hello-world:linux $destFolder/hello-world/:amd64
\ No newline at end of file
+oras cp --to-oci-layout --platform linux/amd64 $srcRegistry/hello-world:linux $destFolder/hello-world/:amd64
+
+# Add sample converted manifest
+oras login $registry -u $username -p $password
+oras cp --from-oci-layout $destFolder/hello-world/:amd64 $registry/hello-world:amd64
+cwd=$(pwd)
+cd ../../../../
+make
+sudo bin/convertor --repository $registry/hello-world --input-tag amd64 --oci --overlaybd amd64-converted -u $username:$password
+cd $cwd
+oras cp --to-oci-layout $registry/hello-world:amd64-converted $destFolder/hello-world/:amd64-converted
\ No newline at end of file
diff --git a/cmd/convertor/testingresources/mocks/registry/hello-world/blobs/sha256/022231c3f1af70e0cc42f0c3781eabf174cff608210585bf93fc83e16d33672d b/cmd/convertor/testingresources/mocks/registry/hello-world/blobs/sha256/022231c3f1af70e0cc42f0c3781eabf174cff608210585bf93fc83e16d33672d
new file mode 100644
index 00000000..bdd6e808
Binary files /dev/null and b/cmd/convertor/testingresources/mocks/registry/hello-world/blobs/sha256/022231c3f1af70e0cc42f0c3781eabf174cff608210585bf93fc83e16d33672d differ
diff --git a/cmd/convertor/testingresources/mocks/registry/hello-world/blobs/sha256/42caa56a19e082b872d43f645bb392e25c9e78bce429755bd709fac598265f88 b/cmd/convertor/testingresources/mocks/registry/hello-world/blobs/sha256/42caa56a19e082b872d43f645bb392e25c9e78bce429755bd709fac598265f88
new file mode 100644
index 00000000..8d94c461
--- /dev/null
+++ b/cmd/convertor/testingresources/mocks/registry/hello-world/blobs/sha256/42caa56a19e082b872d43f645bb392e25c9e78bce429755bd709fac598265f88
@@ -0,0 +1 @@
+{"schemaVersion":2,"mediaType":"application/vnd.oci.image.manifest.v1+json","config":{"mediaType":"application/vnd.oci.image.config.v1+json","digest":"sha256:b67b1f1224de620e9a932ab7ff0e819d7dd9d236605dd8027e5ecf2df650e65e","size":584},"layers":[{"mediaType":"application/vnd.oci.image.layer.v1.tar","digest":"sha256:022231c3f1af70e0cc42f0c3781eabf174cff608210585bf93fc83e16d33672d","size":642560,"annotations":{"containerd.io/snapshot/overlaybd/blob-digest":"sha256:022231c3f1af70e0cc42f0c3781eabf174cff608210585bf93fc83e16d33672d","containerd.io/snapshot/overlaybd/blob-size":"642560","containerd.io/snapshot/overlaybd/version":"0.1.0"}}]}
\ No newline at end of file
diff --git a/cmd/convertor/testingresources/mocks/registry/hello-world/blobs/sha256/b67b1f1224de620e9a932ab7ff0e819d7dd9d236605dd8027e5ecf2df650e65e b/cmd/convertor/testingresources/mocks/registry/hello-world/blobs/sha256/b67b1f1224de620e9a932ab7ff0e819d7dd9d236605dd8027e5ecf2df650e65e
new file mode 100644
index 00000000..ff50eff4
--- /dev/null
+++ b/cmd/convertor/testingresources/mocks/registry/hello-world/blobs/sha256/b67b1f1224de620e9a932ab7ff0e819d7dd9d236605dd8027e5ecf2df650e65e
@@ -0,0 +1 @@
+{"created":"2023-05-04T17:37:03.872958712Z","architecture":"amd64","os":"linux","config":{"Env":["PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin"],"Cmd":["/hello"]},"rootfs":{"type":"layers","diff_ids":["sha256:022231c3f1af70e0cc42f0c3781eabf174cff608210585bf93fc83e16d33672d"]},"history":[{"created":"2023-05-04T17:37:03.801840823Z","created_by":"/bin/sh -c #(nop) COPY file:201f8f1849e89d53be9f6aa76937f5e209d745abfd15a8552fcf2ba45ab267f9 in / "},{"created":"2023-05-04T17:37:03.872958712Z","created_by":"/bin/sh -c #(nop)  CMD [\"/hello\"]","empty_layer":true}]}
\ No newline at end of file
diff --git a/cmd/convertor/testingresources/mocks/registry/hello-world/index.json b/cmd/convertor/testingresources/mocks/registry/hello-world/index.json
index 1be052e7..bdf6d3fc 100644
--- a/cmd/convertor/testingresources/mocks/registry/hello-world/index.json
+++ b/cmd/convertor/testingresources/mocks/registry/hello-world/index.json
@@ -1,6 +1,27 @@
 {
     "schemaVersion": 2,
     "manifests": [
+        {
+            "mediaType": "application/vnd.oci.image.manifest.v1+json",
+            "digest": "sha256:42caa56a19e082b872d43f645bb392e25c9e78bce429755bd709fac598265f88",
+            "size": 641,
+            "annotations": {
+                "org.opencontainers.image.ref.name": "amd64-converted"
+            }
+        },
+        {
+            "mediaType": "application/vnd.docker.distribution.manifest.v2+json",
+            "digest": "sha256:efebf0f7aee69450f99deafe11121afa720abed733943e50581a9dc7540689c8",
+            "size": 525,
+            "annotations": {
+                "org.opencontainers.image.ref.name": "arm64"
+            },
+            "platform": {
+                "architecture": "arm64",
+                "os": "linux",
+                "variant": "v8"
+            }
+        },
         {
             "mediaType": "application/vnd.docker.distribution.manifest.list.v2+json",
             "digest": "sha256:726023f73a8fc5103fa6776d48090539042cb822531c6b751b1f6dd18cb5705d",
@@ -23,23 +44,19 @@
         },
         {
             "mediaType": "application/vnd.docker.distribution.manifest.v2+json",
-            "digest": "sha256:efebf0f7aee69450f99deafe11121afa720abed733943e50581a9dc7540689c8",
+            "digest": "sha256:574efe68740d3bee2ef780036aee2e2da5cf7097ac06513f9f01f41e03365399",
             "size": 525,
-            "annotations": {
-                "org.opencontainers.image.ref.name": "arm64"
-            },
             "platform": {
-                "architecture": "arm64",
-                "os": "linux",
-                "variant": "v8"
+                "architecture": "s390x",
+                "os": "linux"
             }
         },
         {
             "mediaType": "application/vnd.docker.distribution.manifest.v2+json",
-            "digest": "sha256:06bca41ba617acf0b3644df05d0d9c2d2f82ccaab629c0e39792b24682970040",
+            "digest": "sha256:72ba79e34f1baa40cd4d9ecd684b8389d0a1e18cf6e6d5c44c19716d25f65e20",
             "size": 525,
             "platform": {
-                "architecture": "mips64le",
+                "architecture": "riscv64",
                 "os": "linux"
             }
         },
@@ -52,16 +69,6 @@
                 "os": "linux"
             }
         },
-        {
-            "mediaType": "application/vnd.docker.distribution.manifest.v2+json",
-            "digest": "sha256:084c3bdd1271adc754e2c5f6ba7046f1a2c099597dbd9643592fa8eb99981402",
-            "size": 525,
-            "platform": {
-                "architecture": "arm",
-                "os": "linux",
-                "variant": "v5"
-            }
-        },
         {
             "mediaType": "application/vnd.docker.distribution.manifest.v2+json",
             "digest": "sha256:004d23c66201b22fce069b7505756f17088de7889c83891e9bc69d749fa3690e",
@@ -73,30 +80,31 @@
         },
         {
             "mediaType": "application/vnd.docker.distribution.manifest.v2+json",
-            "digest": "sha256:72ba79e34f1baa40cd4d9ecd684b8389d0a1e18cf6e6d5c44c19716d25f65e20",
+            "digest": "sha256:a0a386314d69d1514d7aa63d12532b284bf37bba15ed7b4fc1a3f86605f86c63",
             "size": 525,
             "platform": {
-                "architecture": "riscv64",
-                "os": "linux"
+                "architecture": "arm",
+                "os": "linux",
+                "variant": "v7"
             }
         },
         {
             "mediaType": "application/vnd.docker.distribution.manifest.v2+json",
-            "digest": "sha256:574efe68740d3bee2ef780036aee2e2da5cf7097ac06513f9f01f41e03365399",
+            "digest": "sha256:084c3bdd1271adc754e2c5f6ba7046f1a2c099597dbd9643592fa8eb99981402",
             "size": 525,
             "platform": {
-                "architecture": "s390x",
-                "os": "linux"
+                "architecture": "arm",
+                "os": "linux",
+                "variant": "v5"
             }
         },
         {
             "mediaType": "application/vnd.docker.distribution.manifest.v2+json",
-            "digest": "sha256:a0a386314d69d1514d7aa63d12532b284bf37bba15ed7b4fc1a3f86605f86c63",
+            "digest": "sha256:06bca41ba617acf0b3644df05d0d9c2d2f82ccaab629c0e39792b24682970040",
             "size": 525,
             "platform": {
-                "architecture": "arm",
-                "os": "linux",
-                "variant": "v7"
+                "architecture": "mips64le",
+                "os": "linux"
             }
         }
     ]
diff --git a/docs/USERSPACE_CONVERTOR.md b/docs/USERSPACE_CONVERTOR.md
index 06cb4815..afdec708 100644
--- a/docs/USERSPACE_CONVERTOR.md
+++ b/docs/USERSPACE_CONVERTOR.md
@@ -65,7 +65,7 @@ $ bin/convertor -r docker.io/overlaybd/redis -u user:pass -i 6.2.6 -o 6.2.6_obd
 
 ```
 
-### Layer Deduplication
+### Layer/Manifest Deduplication
 
 To avoid converting the same layer for every image conversion, a database is required to store the correspondence between OCIv1 image layer and overlaybd layer.
 
@@ -85,6 +85,21 @@ CREATE TABLE `overlaybd_layers` (
 ) DEFAULT CHARSET=utf8;
 ```
 
+If you also want caching for manifests to avoid reconverting the same manifest twice, you can create the `overlaybd_manifests` table, the table schema is as follows:
+
+```sql
+CREATE TABLE `overlaybd_manifests` (
+  `host` varchar(255) NOT NULL,
+  `repo` varchar(255) NOT NULL,
+  `src_digest` varchar(255) NOT NULL COMMENT 'digest of the normal image manifest',
+  `out_digest` varchar(255) NOT NULL COMMENT 'digest of overlaybd manifest',
+  `data_size` bigint(20) NOT NULL COMMENT 'size of overlaybd manifest',
+  `mediatype` varchar(255) NOT NULL COMMENT 'mediatype of the converted image manifest',
+  PRIMARY KEY (`host`,`repo`,`src_digest`, `mediatype`),
+  KEY `index_registry_src_digest` (`host`,`src_digest`) USING BTREE
+) DEFAULT CHARSET=utf8;
+```
+
 with this database you can then provide the following flags:
 
 ```bash
@@ -96,6 +111,8 @@ Flags:
 $ bin/convertor -r docker.io/overlaybd/redis -u user:pass -i 6.2.6 -o 6.2.6_obd --db-str "dbuser:dbpass@tcp(127.0.0.1:3306)/dedup" --db-type mysql
 ```
 
+* Note that we have also provided some tools to create such a database and examples of usage as well as a dockerfile that could be used to setup a simple converter with caching capabilities, see [samples](../cmd/convertor/resources/samples).
+
 ## libext2fs
 
 Standalone userspace image-convertor is developed based on [libext2fs](https://github.com/tytso/e2fsprogs), and we have provided a [customized libext2fs](https://github.com/data-accelerator/e2fsprogs) to make the conversion faster. We used `standalone userspace image-convertor (with custom libext2fs)`, `standalone userspace image-convertor (with origin libext2fs)` and `embedded image-convertor` to convert some images and did a comparison for reference.