diff --git a/copy/copy.go b/copy/copy.go
index 328a6c51b..5232086ba 100644
--- a/copy/copy.go
+++ b/copy/copy.go
@@ -16,6 +16,7 @@ import (
 	"github.com/containers/image/v5/image"
 	internalblobinfocache "github.com/containers/image/v5/internal/blobinfocache"
 	"github.com/containers/image/v5/internal/pkg/platform"
+	internalTypes "github.com/containers/image/v5/internal/types"
 	"github.com/containers/image/v5/manifest"
 	"github.com/containers/image/v5/pkg/blobinfocache"
 	"github.com/containers/image/v5/pkg/compression"
@@ -920,7 +921,7 @@ func (ic *imageCopier) copyLayers(ctx context.Context) error {
 				logrus.Debugf("Skipping foreign layer %q copy to %s", cld.destInfo.Digest, ic.c.dest.Reference().Transport().Name())
 			}
 		} else {
-			cld.destInfo, cld.diffID, cld.err = ic.copyLayer(ctx, srcLayer, toEncrypt, pool)
+			cld.destInfo, cld.diffID, cld.err = ic.copyLayer(ctx, srcLayer, toEncrypt, pool, index)
 		}
 		data[index] = cld
 	}
@@ -1116,7 +1117,7 @@ func (c *copier) copyConfig(ctx context.Context, src types.Image) error {
 			progressPool, progressCleanup := c.newProgressPool(ctx)
 			defer progressCleanup()
 			bar := c.createProgressBar(progressPool, srcInfo, "config", "done")
-			destInfo, err := c.copyBlobFromStream(ctx, bytes.NewReader(configBlob), srcInfo, nil, false, true, false, bar)
+			destInfo, err := c.copyBlobFromStream(ctx, bytes.NewReader(configBlob), srcInfo, nil, false, true, false, bar, -1)
 			if err != nil {
 				return types.BlobInfo{}, err
 			}
@@ -1142,7 +1143,7 @@ type diffIDResult struct {
 
 // copyLayer copies a layer with srcInfo (with known Digest and Annotations and possibly known Size) in src to dest, perhaps (de/re/)compressing it,
 // and returns a complete blobInfo of the copied layer, and a value for LayerDiffIDs if diffIDIsNeeded
-func (ic *imageCopier) copyLayer(ctx context.Context, srcInfo types.BlobInfo, toEncrypt bool, pool *mpb.Progress) (types.BlobInfo, digest.Digest, error) {
+func (ic *imageCopier) copyLayer(ctx context.Context, srcInfo types.BlobInfo, toEncrypt bool, pool *mpb.Progress, layerIndex int) (types.BlobInfo, digest.Digest, error) {
 	// If the srcInfo doesn't contain compression information, try to compute it from the
 	// MediaType, which was either read from a manifest by way of LayerInfos() or constructed
 	// by LayerInfosForCopy(), if it was supplied at all.  If we succeed in copying the blob,
@@ -1175,7 +1176,26 @@ func (ic *imageCopier) copyLayer(ctx context.Context, srcInfo types.BlobInfo, to
 		// a failure when we eventually try to update the manifest with the digest and MIME type of the reused blob.
 		// Fixing that will probably require passing more information to TryReusingBlob() than the current version of
 		// the ImageDestination interface lets us pass in.
-		reused, blobInfo, err := ic.c.dest.TryReusingBlob(ctx, srcInfo, ic.c.blobInfoCache, ic.canSubstituteBlobs)
+		var (
+			blobInfo types.BlobInfo
+			reused   bool
+			err      error
+		)
+		// Note: the storage destination optimizes the committing of
+		// layers which requires passing the index of the layer.
+		// Hence, we need to special case and cast.
+		dest, ok := ic.c.dest.(internalTypes.ImageDestinationWithOptions)
+		if ok {
+			options := internalTypes.TryReusingBlobOptions{
+				Cache:         ic.c.blobInfoCache,
+				CanSubstitute: ic.canSubstituteBlobs,
+				LayerIndex:    &layerIndex,
+			}
+			reused, blobInfo, err = dest.TryReusingBlobWithOptions(ctx, srcInfo, options)
+		} else {
+			reused, blobInfo, err = ic.c.dest.TryReusingBlob(ctx, srcInfo, ic.c.blobInfoCache, ic.canSubstituteBlobs)
+		}
+
 		if err != nil {
 			return types.BlobInfo{}, "", errors.Wrapf(err, "Error trying to reuse blob %s at destination", srcInfo.Digest)
 		}
@@ -1217,7 +1237,7 @@ func (ic *imageCopier) copyLayer(ctx context.Context, srcInfo types.BlobInfo, to
 
 	bar := ic.c.createProgressBar(pool, srcInfo, "blob", "done")
 
-	blobInfo, diffIDChan, err := ic.copyLayerFromStream(ctx, srcStream, types.BlobInfo{Digest: srcInfo.Digest, Size: srcBlobSize, MediaType: srcInfo.MediaType, Annotations: srcInfo.Annotations}, diffIDIsNeeded, toEncrypt, bar)
+	blobInfo, diffIDChan, err := ic.copyLayerFromStream(ctx, srcStream, types.BlobInfo{Digest: srcInfo.Digest, Size: srcBlobSize, MediaType: srcInfo.MediaType, Annotations: srcInfo.Annotations}, diffIDIsNeeded, toEncrypt, bar, layerIndex)
 	if err != nil {
 		return types.BlobInfo{}, "", err
 	}
@@ -1248,7 +1268,7 @@ func (ic *imageCopier) copyLayer(ctx context.Context, srcInfo types.BlobInfo, to
 // perhaps (de/re/)compressing the stream,
 // and returns a complete blobInfo of the copied blob and perhaps a <-chan diffIDResult if diffIDIsNeeded, to be read by the caller.
 func (ic *imageCopier) copyLayerFromStream(ctx context.Context, srcStream io.Reader, srcInfo types.BlobInfo,
-	diffIDIsNeeded bool, toEncrypt bool, bar *mpb.Bar) (types.BlobInfo, <-chan diffIDResult, error) {
+	diffIDIsNeeded bool, toEncrypt bool, bar *mpb.Bar, layerIndex int) (types.BlobInfo, <-chan diffIDResult, error) {
 	var getDiffIDRecorder func(compression.DecompressorFunc) io.Writer // = nil
 	var diffIDChan chan diffIDResult
 
@@ -1273,7 +1293,7 @@ func (ic *imageCopier) copyLayerFromStream(ctx context.Context, srcStream io.Rea
 		}
 	}
 
-	blobInfo, err := ic.c.copyBlobFromStream(ctx, srcStream, srcInfo, getDiffIDRecorder, ic.canModifyManifest, false, toEncrypt, bar) // Sets err to nil on success
+	blobInfo, err := ic.c.copyBlobFromStream(ctx, srcStream, srcInfo, getDiffIDRecorder, ic.canModifyManifest, false, toEncrypt, bar, layerIndex) // Sets err to nil on success
 	return blobInfo, diffIDChan, err
 	// We need the defer … pipeWriter.CloseWithError() to happen HERE so that the caller can block on reading from diffIDChan
 }
@@ -1325,7 +1345,7 @@ func (r errorAnnotationReader) Read(b []byte) (n int, err error) {
 // and returns a complete blobInfo of the copied blob.
 func (c *copier) copyBlobFromStream(ctx context.Context, srcStream io.Reader, srcInfo types.BlobInfo,
 	getOriginalLayerCopyWriter func(decompressor compression.DecompressorFunc) io.Writer,
-	canModifyBlob bool, isConfig bool, toEncrypt bool, bar *mpb.Bar) (types.BlobInfo, error) {
+	canModifyBlob bool, isConfig bool, toEncrypt bool, bar *mpb.Bar, layerIndex int) (types.BlobInfo, error) {
 	if isConfig { // This is guaranteed by the caller, but set it here to be explicit.
 		canModifyBlob = false
 	}
@@ -1521,7 +1541,23 @@ func (c *copier) copyBlobFromStream(ctx context.Context, srcStream io.Reader, sr
 	}
 
 	// === Finally, send the layer stream to dest.
-	uploadedInfo, err := c.dest.PutBlob(ctx, &errorAnnotationReader{destStream}, inputInfo, c.blobInfoCache, isConfig)
+	var uploadedInfo types.BlobInfo
+	// Note: the storage destination optimizes the committing of layers
+	// which requires passing the index of the layer.  Hence, we need to
+	// special case and cast.
+	dest, ok := c.dest.(internalTypes.ImageDestinationWithOptions)
+	if ok {
+		options := internalTypes.PutBlobOptions{
+			Cache:    c.blobInfoCache,
+			IsConfig: isConfig,
+		}
+		if !isConfig {
+			options.LayerIndex = &layerIndex
+		}
+		uploadedInfo, err = dest.PutBlobWithOptions(ctx, &errorAnnotationReader{destStream}, inputInfo, options)
+	} else {
+		uploadedInfo, err = c.dest.PutBlob(ctx, &errorAnnotationReader{destStream}, inputInfo, c.blobInfoCache, isConfig)
+	}
 	if err != nil {
 		return types.BlobInfo{}, errors.Wrap(err, "Error writing blob")
 	}
diff --git a/internal/types/types.go b/internal/types/types.go
new file mode 100644
index 000000000..9adf0d536
--- /dev/null
+++ b/internal/types/types.go
@@ -0,0 +1,53 @@
+package types
+
+import (
+	"context"
+	"io"
+
+	publicTypes "github.com/containers/image/v5/types"
+)
+
+// ImageDestinationWithOptions is an internal extension to the ImageDestination
+// interface.
+type ImageDestinationWithOptions interface {
+	publicTypes.ImageDestination
+
+	// PutBlobWithOptions is a wrapper around PutBlob.  If
+	// options.LayerIndex is set, the blob will be committed directly.
+	// Either by the calling goroutine or by another goroutine already
+	// committing layers.
+	//
+	// Please note that TryReusingBlobWithOptions and PutBlobWithOptions
+	// *must* be used the together.  Mixing the two with non "WithOptions"
+	// functions is not supported.
+	PutBlobWithOptions(ctx context.Context, stream io.Reader, blobinfo publicTypes.BlobInfo, options PutBlobOptions) (publicTypes.BlobInfo, error)
+
+	// TryReusingBlobWithOptions is a wrapper around TryReusingBlob.  If
+	// options.LayerIndex is set, the reused blob will be recoreded as
+	// already pulled.
+	//
+	// Please note that TryReusingBlobWithOptions and PutBlobWithOptions
+	// *must* be used the together.  Mixing the two with non "WithOptions"
+	// functions is not supported.
+	TryReusingBlobWithOptions(ctx context.Context, blobinfo publicTypes.BlobInfo, options TryReusingBlobOptions) (bool, publicTypes.BlobInfo, error)
+}
+
+// PutBlobOptions are used in PutBlobWithOptions.
+type PutBlobOptions struct {
+	// Cache to look up blob infos.
+	Cache publicTypes.BlobInfoCache
+	// Denotes whether the blob is a config or not.
+	IsConfig bool
+	// The corresponding index in the layer slice.
+	LayerIndex *int
+}
+
+// TryReusingBlobOptions are used in TryReusingBlobWithOptions.
+type TryReusingBlobOptions struct {
+	// Cache to look up blob infos.
+	Cache publicTypes.BlobInfoCache
+	// Use an equivalent of the desired blob.
+	CanSubstitute bool
+	// The corresponding index in the layer slice.
+	LayerIndex *int
+}
diff --git a/storage/storage_image.go b/storage/storage_image.go
index 3fa1bf618..2ff9b60a6 100644
--- a/storage/storage_image.go
+++ b/storage/storage_image.go
@@ -18,6 +18,7 @@ import (
 	"github.com/containers/image/v5/docker/reference"
 	"github.com/containers/image/v5/image"
 	"github.com/containers/image/v5/internal/tmpdir"
+	internalTypes "github.com/containers/image/v5/internal/types"
 	"github.com/containers/image/v5/manifest"
 	"github.com/containers/image/v5/pkg/blobinfocache/none"
 	"github.com/containers/image/v5/types"
@@ -60,12 +61,16 @@ type storageImageDestination struct {
 	manifest        []byte                          // Manifest contents, temporary
 	signatures      []byte                          // Signature contents, temporary
 	signatureses    map[digest.Digest][]byte        // Instance signature contents, temporary
-	putBlobMutex    sync.Mutex                      // Mutex to sync state for parallel PutBlob executions
+	lock            sync.Mutex                      // Mutex to sync state for parallel executions of PutBlob and TryReusing blob
 	blobDiffIDs     map[digest.Digest]digest.Digest // Mapping from layer blobsums to their corresponding DiffIDs
 	fileSizes       map[digest.Digest]int64         // Mapping from layer blobsums to their sizes
 	filenames       map[digest.Digest]string        // Mapping from layer blobsums to names of files we used to hold them
 	SignatureSizes  []int                           `json:"signature-sizes,omitempty"`  // List of sizes of each signature slice
 	SignaturesSizes map[digest.Digest][]int         `json:"signatures-sizes,omitempty"` // Sizes of each manifest's signature slice
+	// fields to support early commit of layers below
+	currentIndex      int                     // The index of the layer to be committed (i.e., lower indices have already been committed)
+	indexToPulledBlob map[int]*types.BlobInfo // Mapping from layer (by index) to pulled down blob
+	indexToStorageID  map[int]*string         // Mapping from layer (by index) to the associated ID in the storage
 }
 
 type storageImageCloser struct {
@@ -376,14 +381,16 @@ func newImageDestination(sys *types.SystemContext, imageRef storageReference) (*
 		return nil, errors.Wrapf(err, "error creating a temporary directory")
 	}
 	image := &storageImageDestination{
-		imageRef:        imageRef,
-		directory:       directory,
-		signatureses:    make(map[digest.Digest][]byte),
-		blobDiffIDs:     make(map[digest.Digest]digest.Digest),
-		fileSizes:       make(map[digest.Digest]int64),
-		filenames:       make(map[digest.Digest]string),
-		SignatureSizes:  []int{},
-		SignaturesSizes: make(map[digest.Digest][]int),
+		imageRef:          imageRef,
+		directory:         directory,
+		signatureses:      make(map[digest.Digest][]byte),
+		blobDiffIDs:       make(map[digest.Digest]digest.Digest),
+		fileSizes:         make(map[digest.Digest]int64),
+		filenames:         make(map[digest.Digest]string),
+		SignatureSizes:    []int{},
+		SignaturesSizes:   make(map[digest.Digest][]int),
+		indexToStorageID:  make(map[int]*string),
+		indexToPulledBlob: make(map[int]*types.BlobInfo),
 	}
 	return image, nil
 }
@@ -410,6 +417,26 @@ func (s *storageImageDestination) computeNextBlobCacheFile() string {
 	return filepath.Join(s.directory, fmt.Sprintf("%d", atomic.AddInt32(&s.nextTempFileID, 1)))
 }
 
+// PutBlobWithOptions is a wrapper around PutBlob.  If options.LayerIndex is
+// set, the blob will be committed directly.  Either by the calling goroutine
+// or by another goroutine already committing layers.
+//
+// Please not that TryReusingBlobWithOptions and PutBlobWithOptions *must* be
+// used the together.  Mixing the two with non "WithOptions" functions is not
+// supported.
+func (s *storageImageDestination) PutBlobWithOptions(ctx context.Context, stream io.Reader, blobinfo types.BlobInfo, options internalTypes.PutBlobOptions) (types.BlobInfo, error) {
+	info, err := s.PutBlob(ctx, stream, blobinfo, options.Cache, options.IsConfig)
+	if err != nil {
+		return info, err
+	}
+
+	if options.IsConfig || options.LayerIndex == nil {
+		return info, nil
+	}
+
+	return info, s.queueOrCommit(ctx, info, *options.LayerIndex)
+}
+
 // HasThreadSafePutBlob indicates whether PutBlob can be executed concurrently.
 func (s *storageImageDestination) HasThreadSafePutBlob() bool {
 	return true
@@ -465,11 +492,11 @@ func (s *storageImageDestination) PutBlob(ctx context.Context, stream io.Reader,
 		return errorBlobInfo, errors.WithStack(ErrBlobSizeMismatch)
 	}
 	// Record information about the blob.
-	s.putBlobMutex.Lock()
+	s.lock.Lock()
 	s.blobDiffIDs[hasher.Digest()] = diffID.Digest()
 	s.fileSizes[hasher.Digest()] = counter.Count
 	s.filenames[hasher.Digest()] = filename
-	s.putBlobMutex.Unlock()
+	s.lock.Unlock()
 	blobDigest := blobinfo.Digest
 	if blobDigest.Validate() != nil {
 		blobDigest = hasher.Digest()
@@ -487,6 +514,22 @@ func (s *storageImageDestination) PutBlob(ctx context.Context, stream io.Reader,
 	}, nil
 }
 
+// TryReusingBlobWithOptions is a wrapper around TryReusingBlob.  If
+// options.LayerIndex is set, the reused blob will be recoreded as already
+// pulled.
+//
+// Please not that TryReusingBlobWithOptions and PutBlobWithOptions *must* be
+// used the together.  Mixing the two with the non "WithOptions" functions
+// is not supported.
+func (s *storageImageDestination) TryReusingBlobWithOptions(ctx context.Context, blobinfo types.BlobInfo, options internalTypes.TryReusingBlobOptions) (bool, types.BlobInfo, error) {
+	reused, info, err := s.TryReusingBlob(ctx, blobinfo, options.Cache, options.CanSubstitute)
+	if err != nil || !reused || options.LayerIndex == nil {
+		return reused, info, err
+	}
+
+	return reused, info, s.queueOrCommit(ctx, info, *options.LayerIndex)
+}
+
 // TryReusingBlob checks whether the transport already contains, or can efficiently reuse, a blob, and if so, applies it to the current destination
 // (e.g. if the blob is a filesystem layer, this signifies that the changes it describes need to be applied again when composing a filesystem tree).
 // info.Digest must not be empty.
@@ -498,8 +541,8 @@ func (s *storageImageDestination) PutBlob(ctx context.Context, stream io.Reader,
 // May use and/or update cache.
 func (s *storageImageDestination) TryReusingBlob(ctx context.Context, blobinfo types.BlobInfo, cache types.BlobInfoCache, canSubstitute bool) (bool, types.BlobInfo, error) {
 	// lock the entire method as it executes fairly quickly
-	s.putBlobMutex.Lock()
-	defer s.putBlobMutex.Unlock()
+	s.lock.Lock()
+	defer s.lock.Unlock()
 	if blobinfo.Digest == "" {
 		return false, types.BlobInfo{}, errors.Errorf(`Can not check for a blob with unknown digest`)
 	}
@@ -636,6 +679,189 @@ func (s *storageImageDestination) getConfigBlob(info types.BlobInfo) ([]byte, er
 	return nil, errors.New("blob not found")
 }
 
+// queueOrCommit queues in the specified blob to be committed to the storage.
+// If no other goroutine is already committing layers, the layer and all
+// subsequent layers (if already queued) will be committed to the storage.
+func (s *storageImageDestination) queueOrCommit(ctx context.Context, blob types.BlobInfo, index int) error {
+	// NOTE: whenever the code below is touched, make sure that all code
+	// paths unlock the lock and to unlock it exactly once.
+	//
+	// Conceptually, the code is divided in two stages:
+	//
+	// 1) Queue in work by marking the layer as ready to be committed.
+	//    If at least one previous/parent layer with a lower index has
+	//    not yet been committed, return early.
+	//
+	// 2) Process the queued-in work by committing the "ready" layers
+	//    in sequence.  Make sure that more items can be queued-in
+	//    during the comparatively I/O expensive task of committing a
+	//    layer.
+	//
+	// The conceptual benefit of this design is that caller can continue
+	// pulling layers after an early return.  At any given time, only one
+	// caller is the "worker" routine comitting layers.  All other routines
+	// can continue pulling and queuing in layers.
+	s.lock.Lock()
+	s.indexToPulledBlob[index] = &blob
+
+	// We're still waiting for at least one previous/parent layer to be
+	// committed, so there's nothing to do.
+	if index != s.currentIndex {
+		s.lock.Unlock()
+		return nil
+	}
+
+	for info := s.indexToPulledBlob[index]; info != nil; info = s.indexToPulledBlob[index] {
+		s.lock.Unlock()
+		layerInfo := manifest.LayerInfo{
+			BlobInfo:   *info,
+			EmptyLayer: info.Digest == image.GzippedEmptyLayerDigest,
+		}
+		// Note: commitLayer locks on-demand.
+		if err := s.commitLayer(ctx, layerInfo, index); err != nil {
+			return err
+		}
+		s.lock.Lock()
+		index++
+	}
+
+	// Set the index at the very end to make sure that only one routine
+	// enters stage 2).
+	s.currentIndex = index
+	s.lock.Unlock()
+	return nil
+}
+
+// commitLayer commits the specified blob with the given index to the storage.
+// Note that the previous layer is expected to already be committed.  Callers
+// must take care of sequencing and locking.
+func (s *storageImageDestination) commitLayer(ctx context.Context, blob manifest.LayerInfo, index int) error {
+	// Already commited?  Return early.
+	if _, alreadyCommitted := s.indexToStorageID[index]; alreadyCommitted {
+		return nil
+	}
+
+	// Start with an empty string or the previous layer ID.  Note that
+	// `s.indexToStorageID` can only be accessed by *one* goroutine at any
+	// given time. Hence, we don't need to lock accesses.
+	var lastLayer string
+	if prev := s.indexToStorageID[index-1]; prev != nil {
+		lastLayer = *prev
+	}
+
+	// Carry over the previous ID for empty non-base layers.
+	if blob.EmptyLayer && index > 0 {
+		s.indexToStorageID[index] = &lastLayer
+		return nil
+	}
+
+	// Check if there's already a layer with the ID that we'd give to the result of applying
+	// this layer blob to its parent, if it has one, or the blob's hex value otherwise.
+	s.lock.Lock()
+	diffID, haveDiffID := s.blobDiffIDs[blob.Digest]
+	s.lock.Unlock()
+	if !haveDiffID {
+		// Check if it's elsewhere and the caller just forgot to pass it to us in a PutBlob(),
+		// or to even check if we had it.
+		// Use none.NoCache to avoid a repeated DiffID lookup in the BlobInfoCache; a caller
+		// that relies on using a blob digest that has never been seen by the store had better call
+		// TryReusingBlob; not calling PutBlob already violates the documented API, so there’s only
+		// so far we are going to accommodate that (if we should be doing that at all).
+		logrus.Debugf("looking for diffID for blob %+v", blob.Digest)
+		// NOTE: use `TryReusingBlob` to prevent recursion.
+		has, _, err := s.TryReusingBlob(ctx, blob.BlobInfo, none.NoCache, false)
+		if err != nil {
+			return errors.Wrapf(err, "error checking for a layer based on blob %q", blob.Digest.String())
+		}
+		if !has {
+			return errors.Errorf("error determining uncompressed digest for blob %q", blob.Digest.String())
+		}
+		diffID, haveDiffID = s.blobDiffIDs[blob.Digest]
+		if !haveDiffID {
+			return errors.Errorf("we have blob %q, but don't know its uncompressed digest", blob.Digest.String())
+		}
+	}
+	id := diffID.Hex()
+	if lastLayer != "" {
+		id = digest.Canonical.FromBytes([]byte(lastLayer + "+" + diffID.Hex())).Hex()
+	}
+	if layer, err2 := s.imageRef.transport.store.Layer(id); layer != nil && err2 == nil {
+		// There's already a layer that should have the right contents, just reuse it.
+		lastLayer = layer.ID
+		s.indexToStorageID[index] = &lastLayer
+		return nil
+	}
+	// Check if we previously cached a file with that blob's contents.  If we didn't,
+	// then we need to read the desired contents from a layer.
+	s.lock.Lock()
+	filename, ok := s.filenames[blob.Digest]
+	s.lock.Unlock()
+	if !ok {
+		// Try to find the layer with contents matching that blobsum.
+		layer := ""
+		layers, err2 := s.imageRef.transport.store.LayersByUncompressedDigest(diffID)
+		if err2 == nil && len(layers) > 0 {
+			layer = layers[0].ID
+		} else {
+			layers, err2 = s.imageRef.transport.store.LayersByCompressedDigest(blob.Digest)
+			if err2 == nil && len(layers) > 0 {
+				layer = layers[0].ID
+			}
+		}
+		if layer == "" {
+			return errors.Wrapf(err2, "error locating layer for blob %q", blob.Digest)
+		}
+		// Read the layer's contents.
+		noCompression := archive.Uncompressed
+		diffOptions := &storage.DiffOptions{
+			Compression: &noCompression,
+		}
+		diff, err2 := s.imageRef.transport.store.Diff("", layer, diffOptions)
+		if err2 != nil {
+			return errors.Wrapf(err2, "error reading layer %q for blob %q", layer, blob.Digest)
+		}
+		// Copy the layer diff to a file.  Diff() takes a lock that it holds
+		// until the ReadCloser that it returns is closed, and PutLayer() wants
+		// the same lock, so the diff can't just be directly streamed from one
+		// to the other.
+		filename = s.computeNextBlobCacheFile()
+		file, err := os.OpenFile(filename, os.O_CREATE|os.O_TRUNC|os.O_WRONLY|os.O_EXCL, 0600)
+		if err != nil {
+			diff.Close()
+			return errors.Wrapf(err, "error creating temporary file %q", filename)
+		}
+		// Copy the data to the file.
+		// TODO: This can take quite some time, and should ideally be cancellable using
+		// ctx.Done().
+		_, err = io.Copy(file, diff)
+		diff.Close()
+		file.Close()
+		if err != nil {
+			return errors.Wrapf(err, "error storing blob to file %q", filename)
+		}
+		// Make sure that we can find this file later, should we need the layer's
+		// contents again.
+		s.lock.Lock()
+		s.filenames[blob.Digest] = filename
+		s.lock.Unlock()
+	}
+	// Read the cached blob and use it as a diff.
+	file, err := os.Open(filename)
+	if err != nil {
+		return errors.Wrapf(err, "error opening file %q", filename)
+	}
+	defer file.Close()
+	// Build the new layer using the diff, regardless of where it came from.
+	// TODO: This can take quite some time, and should ideally be cancellable using ctx.Done().
+	layer, _, err := s.imageRef.transport.store.PutLayer(id, lastLayer, nil, "", false, nil, file)
+	if err != nil && errors.Cause(err) != storage.ErrDuplicateID {
+		return errors.Wrapf(err, "error adding layer with blob %q", blob.Digest)
+	}
+
+	s.indexToStorageID[index] = &layer.ID
+	return nil
+}
+
 func (s *storageImageDestination) Commit(ctx context.Context, unparsedToplevel types.UnparsedImage) error {
 	if len(s.manifest) == 0 {
 		return errors.New("Internal error: storageImageDestination.Commit() called without PutManifest()")
@@ -673,108 +899,19 @@ func (s *storageImageDestination) Commit(ctx context.Context, unparsedToplevel t
 		return errors.Wrapf(err, "error parsing manifest")
 	}
 	layerBlobs := man.LayerInfos()
-	// Extract or find the layers.
-	lastLayer := ""
-	for _, blob := range layerBlobs {
-		if blob.EmptyLayer {
-			continue
-		}
-
-		// Check if there's already a layer with the ID that we'd give to the result of applying
-		// this layer blob to its parent, if it has one, or the blob's hex value otherwise.
-		diffID, haveDiffID := s.blobDiffIDs[blob.Digest]
-		if !haveDiffID {
-			// Check if it's elsewhere and the caller just forgot to pass it to us in a PutBlob(),
-			// or to even check if we had it.
-			// Use none.NoCache to avoid a repeated DiffID lookup in the BlobInfoCache; a caller
-			// that relies on using a blob digest that has never been seen by the store had better call
-			// TryReusingBlob; not calling PutBlob already violates the documented API, so there’s only
-			// so far we are going to accommodate that (if we should be doing that at all).
-			logrus.Debugf("looking for diffID for blob %+v", blob.Digest)
-			has, _, err := s.TryReusingBlob(ctx, blob.BlobInfo, none.NoCache, false)
-			if err != nil {
-				return errors.Wrapf(err, "error checking for a layer based on blob %q", blob.Digest.String())
-			}
-			if !has {
-				return errors.Errorf("error determining uncompressed digest for blob %q", blob.Digest.String())
-			}
-			diffID, haveDiffID = s.blobDiffIDs[blob.Digest]
-			if !haveDiffID {
-				return errors.Errorf("we have blob %q, but don't know its uncompressed digest", blob.Digest.String())
-			}
-		}
-		id := diffID.Hex()
-		if lastLayer != "" {
-			id = digest.Canonical.FromBytes([]byte(lastLayer + "+" + diffID.Hex())).Hex()
-		}
-		if layer, err2 := s.imageRef.transport.store.Layer(id); layer != nil && err2 == nil {
-			// There's already a layer that should have the right contents, just reuse it.
-			lastLayer = layer.ID
-			continue
-		}
-		// Check if we previously cached a file with that blob's contents.  If we didn't,
-		// then we need to read the desired contents from a layer.
-		filename, ok := s.filenames[blob.Digest]
-		if !ok {
-			// Try to find the layer with contents matching that blobsum.
-			layer := ""
-			layers, err2 := s.imageRef.transport.store.LayersByUncompressedDigest(diffID)
-			if err2 == nil && len(layers) > 0 {
-				layer = layers[0].ID
-			} else {
-				layers, err2 = s.imageRef.transport.store.LayersByCompressedDigest(blob.Digest)
-				if err2 == nil && len(layers) > 0 {
-					layer = layers[0].ID
-				}
-			}
-			if layer == "" {
-				return errors.Wrapf(err2, "error locating layer for blob %q", blob.Digest)
-			}
-			// Read the layer's contents.
-			noCompression := archive.Uncompressed
-			diffOptions := &storage.DiffOptions{
-				Compression: &noCompression,
-			}
-			diff, err2 := s.imageRef.transport.store.Diff("", layer, diffOptions)
-			if err2 != nil {
-				return errors.Wrapf(err2, "error reading layer %q for blob %q", layer, blob.Digest)
-			}
-			// Copy the layer diff to a file.  Diff() takes a lock that it holds
-			// until the ReadCloser that it returns is closed, and PutLayer() wants
-			// the same lock, so the diff can't just be directly streamed from one
-			// to the other.
-			filename = s.computeNextBlobCacheFile()
-			file, err := os.OpenFile(filename, os.O_CREATE|os.O_TRUNC|os.O_WRONLY|os.O_EXCL, 0600)
-			if err != nil {
-				diff.Close()
-				return errors.Wrapf(err, "error creating temporary file %q", filename)
-			}
-			// Copy the data to the file.
-			// TODO: This can take quite some time, and should ideally be cancellable using
-			// ctx.Done().
-			_, err = io.Copy(file, diff)
-			diff.Close()
-			file.Close()
-			if err != nil {
-				return errors.Wrapf(err, "error storing blob to file %q", filename)
-			}
-			// Make sure that we can find this file later, should we need the layer's
-			// contents again.
-			s.filenames[blob.Digest] = filename
-		}
-		// Read the cached blob and use it as a diff.
-		file, err := os.Open(filename)
-		if err != nil {
-			return errors.Wrapf(err, "error opening file %q", filename)
+	// Extract, commit, or find the layers.
+	for i, blob := range layerBlobs {
+		if err := s.commitLayer(ctx, blob, i); err != nil {
+			return err
 		}
-		defer file.Close()
-		// Build the new layer using the diff, regardless of where it came from.
-		// TODO: This can take quite some time, and should ideally be cancellable using ctx.Done().
-		layer, _, err := s.imageRef.transport.store.PutLayer(id, lastLayer, nil, "", false, nil, file)
-		if err != nil && errors.Cause(err) != storage.ErrDuplicateID {
-			return errors.Wrapf(err, "error adding layer with blob %q", blob.Digest)
+	}
+	var lastLayer string
+	if len(layerBlobs) > 0 { // Can happen when using caches
+		prev := s.indexToStorageID[len(layerBlobs)-1]
+		if prev == nil {
+			return errors.Errorf("Internal error: StorageImageDestination.Commit(): previous layer %d hasn't been commited (lastLayer == nil)", len(layerBlobs)-1)
 		}
-		lastLayer = layer.ID
+		lastLayer = *prev
 	}
 
 	// If one of those blobs was a configuration blob, then we can try to dig out the date when the image