fix(gc): respect in-progress .sync sessions when removing repos

pkg/storage/gc/gc.go

@@ -20,6 +20,7 @@
 	zcommon "zotregistry.dev/zot/v2/pkg/common"
 	"zotregistry.dev/zot/v2/pkg/compat"
 	"zotregistry.dev/zot/v2/pkg/extensions/monitoring"
+	"zotregistry.dev/zot/v2/pkg/extensions/sync/constants"
 	zlog "zotregistry.dev/zot/v2/pkg/log"
 	mTypes "zotregistry.dev/zot/v2/pkg/meta/types"
 	"zotregistry.dev/zot/v2/pkg/retention"
@@ -195,15 +196,23 @@
 		return err
 	}
 
+	// gc stale sync staging sessions
+	sessionsDeleted, err := gc.removeStaleSyncSessions(repo, gc.opts.Delay)
+	if err != nil {
+		return err
+	}
+
 	if !gc.opts.ImageRetention.DryRun {
 		monitoring.IncGCDeleted(gc.metrics, "manifest", manifestsDeleted)
 		monitoring.IncGCDeleted(gc.metrics, "blob", blobsDeleted)
 		monitoring.IncGCDeleted(gc.metrics, "upload", uploadsDeleted)
+		monitoring.IncGCDeleted(gc.metrics, "syncSession", sessionsDeleted)
 	}
 
 	return nil
 }
 
+// removeStaleManifestEntries removes manifest/index descriptors whose blobs no longer exist in storage.
 func (gc GarbageCollect) removeStaleManifestEntries(repo string, index *ispec.Index) error {
 	if gc.opts.ImageRetention.DryRun {
 		return nil
@@ -884,6 +893,65 @@
 	return deleted, aggregatedErr
 }
 
+// removeStaleSyncSessions deletes sync staging sessions (<repo>/.sync/<uuid>)
+// which are past their gc delay. These are left behind when a sync is
+// interrupted; live sessions are protected by the delay.
+func (gc GarbageCollect) removeStaleSyncSessions(repo string, delay time.Duration) (int, error) {
+	gc.log.Debug().Str("module", "gc").Str("repository", repo).Msg("cleaning stale sync staging sessions")
+
+	repoSyncDir := path.Join(gc.imgStore.RootDir(), repo, constants.SyncBlobUploadDir)
+	if !gc.imgStore.DirExists(repoSyncDir) {
+		// The repository or .sync directory may have already been removed
+		return 0, nil
+	}
+
+	sessions, err := gc.imgStore.ListSyncSessions(repo)
+	if err != nil {
+		// A PathNotFoundError from the underlying driver means .sync is empty or absent
+		var pathNotFoundErr driver.PathNotFoundError
+		if errors.As(err, &pathNotFoundErr) {
+			return 0, nil
+		}
+
+		gc.log.Error().Err(err).Str("module", "gc").Str("repository", repo).Msg("failed to list sync sessions")
+
+		return 0, err
+	}
+
+	var aggregatedErr error
+
+	deleted := 0
+
+	for _, id := range sessions {
+		_, size, modtime, err := gc.imgStore.StatSyncSession(repo, id)
+		if err != nil {
+			gc.log.Error().Err(err).Str("module", "gc").Str("repository", repo).Str("session", id).
+				Msg("failed to stat sync session")
+
+			aggregatedErr = errors.Join(aggregatedErr, err)
+
+			continue
+		}
+
+		if modtime.Add(delay).After(time.Now()) {
+			// Do not delete sync sessions which have been updated recently
+			continue
+		}
+
+		err = gc.imgStore.DeleteSyncSession(repo, id)
+		if err != nil {
+			gc.log.Error().Err(err).Str("module", "gc").Str("repository", repo).Str("session", id).
+				Str("size", strconv.FormatInt(size, 10)).Str("modified", modtime.String()).Msg("failed to delete sync session")
+
+			aggregatedErr = errors.Join(aggregatedErr, err)
+		} else {
+			deleted++
+		}
+	}
+
+	return deleted, aggregatedErr
+}
+
 // removeUnreferencedBlobs gc all blobs which are not referenced by any manifest found in repo's index.json.
 func (gc GarbageCollect) removeUnreferencedBlobs(repo string, delay time.Duration, log zlog.Logger,
 ) (int, error) {
@@ -942,7 +1010,6 @@
 		}
 	}
 
-	// if we removed all blobs from repo
 	removeRepo := len(gcBlobs) > 0 && len(gcBlobs) == len(allBlobs)
 
 	reaped, err := gc.imgStore.CleanupRepo(repo, gcBlobs, removeRepo)

pkg/storage/gc/gc_internal_test.go

@@ -1760,3 +1760,178 @@ func TestCleanupRepoMissingBlob(t *testing.T) {
 		So(count, ShouldEqual, 1)
 	})
 }
+
+func TestRemoveStaleSyncSessions(t *testing.T) {
+	Convey("removeStaleSyncSessions reaps stale sessions and keeps recent ones", t, func() {
+		dir := t.TempDir()
+
+		audit := zlog.NewAuditLogger("debug", "")
+		log := zlog.NewTestLogger()
+
+		metrics := monitoring.NewMetricsServer(false, log)
+		defer metrics.Stop()
+
+		cacheDriver, _ := storage.Create("boltdb", cache.BoltDBDriverParameters{
+			RootDir:     dir,
+			Name:        "cache",
+			UseRelPaths: true,
+		}, log)
+		imgStore := local.NewImageStore(dir, true, true, log, metrics, nil, cacheDriver, nil, nil)
+
+		falseVal := false
+		gcOptions := Options{
+			Delay: 1 * time.Hour,
+			ImageRetention: config.ImageRetention{
+				Delay:    storageConstants.DefaultGCDelay,
+				Policies: []config.RetentionPolicy{{Repositories: []string{"**"}, DeleteUntagged: &falseVal}},
+			},
+		}
+
+		imgStore.InitRepo(context.Background(), repoName)
+
+		syncDir := path.Join(dir, repoName, ".sync")
+		err := os.MkdirAll(syncDir, 0o755)
+		So(err, ShouldBeNil)
+
+		older := time.Now().Add(-2 * time.Hour)
+
+		staleDir := path.Join(syncDir, "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee")
+		err = os.MkdirAll(staleDir, 0o755)
+		So(err, ShouldBeNil)
+		err = os.Chtimes(staleDir, older, older)
+		So(err, ShouldBeNil)
+
+		// fresh session (recent)
+		freshDir := path.Join(syncDir, "bbbbbbbb-cccc-dddd-eeee-ffffffffffffff")
+		err = os.MkdirAll(freshDir, 0o755)
+		So(err, ShouldBeNil)
+
+		gc := NewGarbageCollect(imgStore, mocks.MetaDBMock{}, gcOptions, audit, log, metrics)
+
+		deleted, err := gc.removeStaleSyncSessions(repoName, gc.opts.Delay)
+		So(err, ShouldBeNil)
+		So(deleted, ShouldEqual, 1)
+
+		// stale session should be gone
+		_, err = os.Stat(staleDir)
+		So(os.IsNotExist(err), ShouldBeTrue)
+		// fresh session should remain
+		_, err = os.Stat(freshDir)
+		So(err, ShouldBeNil)
+	})
+}
+
+func TestRemoveStaleSyncSessionsNoSessions(t *testing.T) {
+	Convey("SyncSessions returns 0 when no .sync directory exists", t, func() {
+		dir := t.TempDir()
+
+		audit := zlog.NewAuditLogger("debug", "")
+		log := zlog.NewTestLogger()
+
+		metrics := monitoring.NewMetricsServer(false, log)
+		defer metrics.Stop()
+
+		cacheDriver, _ := storage.Create("boltdb", cache.BoltDBDriverParameters{
+			RootDir:     dir,
+			Name:        "cache",
+			UseRelPaths: true,
+		}, log)
+		imgStore := local.NewImageStore(dir, true, true, log, metrics, nil, cacheDriver, nil, nil)
+
+		falseVal := false
+		gcOptions := Options{
+			Delay: 1 * time.Hour,
+			ImageRetention: config.ImageRetention{
+				Delay:    storageConstants.DefaultGCDelay,
+				Policies: []config.RetentionPolicy{{Repositories: []string{"**"}, DeleteUntagged: &falseVal}},
+			},
+		}
+
+		imgStore.InitRepo(context.Background(), repoName)
+
+		gc := NewGarbageCollect(imgStore, mocks.MetaDBMock{}, gcOptions, audit, log, metrics)
+
+		deleted, err := gc.removeStaleSyncSessions(repoName, gc.opts.Delay)
+		So(err, ShouldBeNil)
+		So(deleted, ShouldEqual, 0)
+	})
+}
+
+func TestGCReaperSyncSessionsWithRepoRemoval(t *testing.T) {
+	Convey("stale session blocks repo removal and is reaped; empty repo left in place", t, func() {
+		dir := t.TempDir()
+
+		audit := zlog.NewAuditLogger("debug", "")
+		log := zlog.NewTestLogger()
+
+		metrics := monitoring.NewMetricsServer(false, log)
+		defer metrics.Stop()
+
+		cacheDriver, _ := storage.Create("boltdb", cache.BoltDBDriverParameters{
+			RootDir:     dir,
+			Name:        "cache",
+			UseRelPaths: true,
+		}, log)
+
+		falseVal := false
+		gcOptions := Options{
+			Delay: 1 * time.Hour,
+			ImageRetention: config.ImageRetention{
+				Delay:    storageConstants.DefaultGCDelay,
+				Policies: []config.RetentionPolicy{{Repositories: []string{"**"}, DeleteUntagged: &falseVal}},
+			},
+		}
+
+		imgStore := local.NewImageStore(dir, true, true, log, metrics, nil, cacheDriver, nil, nil)
+		imgStore.InitRepo(context.Background(), repoName)
+
+		// Upload a blob then remove it from the index so it becomes unreferenced
+		content := []byte("disappearing blob")
+		digest := godigest.FromBytes(content)
+		_, _, err := imgStore.FullBlobUpload(context.Background(), repoName, bytes.NewReader(content), digest)
+		So(err, ShouldBeNil)
+
+		// Write an empty index so this blob is unreferenced
+		idx := ispec.Index{Manifests: []ispec.Descriptor{}}
+		err = imgStore.PutIndexContent(repoName, idx)
+		So(err, ShouldBeNil)
+
+		// Create a stale sync session
+		syncDir := path.Join(dir, repoName, ".sync")
+		err = os.MkdirAll(syncDir, 0o755)
+		So(err, ShouldBeNil)
+		staleDir := path.Join(syncDir, "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee")
+		err = os.MkdirAll(staleDir, 0o755)
+		So(err, ShouldBeNil)
+		older := time.Now().Add(-2 * time.Hour)
+		err = os.Chtimes(staleDir, older, older)
+		So(err, ShouldBeNil)
+
+		// Age the blob so it is eligible for garbage collection (older than the 1h delay).
+		blobPath := path.Join(dir, repoName, "blobs", digest.Algorithm().String(), digest.Encoded())
+		err = os.Chtimes(blobPath, older, older)
+		So(err, ShouldBeNil)
+
+		gc := NewGarbageCollect(imgStore, mocks.MetaDBMock{}, gcOptions, audit, log, metrics)
+		ctx := context.Background()
+
+		// A single GC pass: removeUnreferencedBlobs reaps the unreferenced blob, the
+		// in-progress sync session blocks CleanupRepo from removing the repo, and
+		// removeStaleSyncSessions then reaps the now-stale session.
+		err = gc.CleanRepo(ctx, repoName)
+		So(err, ShouldBeNil)
+
+		// Stale session should be gone after pass 1
+		_, err = os.Stat(staleDir)
+		So(os.IsNotExist(err), ShouldBeTrue)
+
+		// Blob should also be gone (unreferenced and past delay).
+		_, err = os.Stat(blobPath)
+		So(os.IsNotExist(err), ShouldBeTrue)
+
+		// The repo directory still exists - GC does not remove empty repos.
+		repoDir := path.Join(dir, repoName)
+		_, err = os.Stat(repoDir)
+		So(err, ShouldBeNil)
+	})
+}

pkg/storage/imagestore/imagestore.go

@@ -936,6 +936,45 @@
 	return blobUploads, err
 }
 
+// syncSessionPath returns the full path for a sync staging session directory.
+func (is *ImageStore) syncSessionPath(repo, id string) string {
+	return path.Join(is.rootDir, repo, syncConstants.SyncBlobUploadDir, id)
+}
+
+// ListSyncSessions lists sync staging session IDs (<uuid>) under <repo>/.sync.
+func (is *ImageStore) ListSyncSessions(repo string) ([]string, error) {
+	sessionPaths, err := is.storeDriver.List(path.Join(is.rootDir, repo, syncConstants.SyncBlobUploadDir))
+	if err != nil {
+		return nil, err
+	}
+	sessions := make([]string, 0, len(sessionPaths))
+	for _, sessionPath := range sessionPaths {
+		sessions = append(sessions, path.Base(sessionPath))
+	}
+	return sessions, nil
+}
+
+// StatSyncSession returns the existence, size, and modtime of a sync session directory.
+func (is *ImageStore) StatSyncSession(repo, id string) (bool, int64, time.Time, error) {
+	sessionPath := is.syncSessionPath(repo, id)
+	binfo, err := is.storeDriver.Stat(sessionPath)
+	if err != nil {
+		is.log.Debug().Err(err).Str("session", id).Msg("failed to stat sync session")
+		return false, -1, time.Time{}, err
+	}
+	return true, binfo.Size(), binfo.ModTime(), nil
+}
+
+// DeleteSyncSession removes a sync staging session directory.
+func (is *ImageStore) DeleteSyncSession(repo, id string) error {
+	sessionPath := is.syncSessionPath(repo, id)
+	if err := is.storeDriver.Delete(sessionPath); err != nil {
+		is.log.Error().Err(err).Str("session", id).Msg("failed to delete sync session")
+		return err
+	}
+	return nil
+}
+
 // StatBlobUpload verifies if a blob upload is present inside a repository. The caller function MUST lock from outside.
 func (is *ImageStore) StatBlobUpload(repo, uuid string) (bool, int64, time.Time, error) {
 	blobUploadPath := is.BlobUploadPath(repo, uuid)
@@ -1974,9 +2013,11 @@
 	}
 
 	blobUploads, _ := is.ListBlobUploads(repo)
+	syncSessions, _ := is.ListSyncSessions(repo)
 
-	// if removeRepo flag is true and we cleanup all blobs and there are no blobs currently being uploaded.
-	if removeRepo && count == len(blobs) && count > 0 && len(blobUploads) == 0 {
+	// if removeRepo flag is true and there are no uploads or sync sessions currently in-progress.
+	if removeRepo && count == len(blobs) && count > 0 &&
+		len(blobUploads) == 0 && len(syncSessions) == 0 {
 		is.log.Info().Str("repository", repo).Msg("removed all blobs, removing repo")
 
 		if err := is.storeDriver.Delete(path.Join(is.rootDir, repo)); err != nil {