local: handle canceled uploads

cache/contenthash/checksum.go

@@ -60,6 +60,10 @@ func SetCacheContext(ctx context.Context, md *metadata.StorageItem, cc CacheCont
 	return getDefaultManager().SetCacheContext(ctx, md, cc)
 }
 
+func ClearCacheContext(md *metadata.StorageItem) {
+	getDefaultManager().clearCacheContext(md.ID())
+}
+
 type CacheContext interface {
 	Checksum(ctx context.Context, ref cache.Mountable, p string, followLinks bool) (digest.Digest, error)
 	ChecksumWildcard(ctx context.Context, ref cache.Mountable, p string, followLinks bool) (digest.Digest, error)
@@ -142,6 +146,12 @@ func (cm *cacheManager) SetCacheContext(ctx context.Context, md *metadata.Storag
 	return nil
 }
 
+func (cm *cacheManager) clearCacheContext(id string) {
+	cm.lruMu.Lock()
+	cm.lru.Remove(id)
+	cm.lruMu.Unlock()
+}
+
 type cacheContext struct {
 	mu    sync.RWMutex
 	md    *metadata.StorageItem

cache/manager.go

@@ -737,6 +737,10 @@ func CachePolicyRetain(m withMetadata) error {
 	return queueCachePolicy(m.Metadata(), cachePolicyRetain)
 }
 
+func CachePolicyDefault(m withMetadata) error {
+	return queueCachePolicy(m.Metadata(), cachePolicyDefault)
+}
+
 func WithDescription(descr string) RefOption {
 	return func(m withMetadata) error {
 		return queueDescription(m.Metadata(), descr)

source/local/local.go

@@ -133,6 +133,12 @@ func (ls *localSourceHandler) Snapshot(ctx context.Context) (out cache.Immutable
 
 	defer func() {
 		if retErr != nil && mutable != nil {
+			// on error remove the record as checksum update is in undefined state
+			cache.CachePolicyDefault(mutable)
+			if err := mutable.Metadata().Commit(); err != nil {
+				logrus.Errorf("failed to reset mutable cachepolicy: %v", err)
+			}
+			contenthash.ClearCacheContext(mutable.Metadata())
 			go mutable.Release(context.TODO())
 		}
 	}()