issue-1559: [Disk Manager] Add filesystem snapshot barrier mechanism

cloud/disk_manager/internal/pkg/dataplane/filesystem/snapshot/create_snapshot_from_filesystem_task.go

@@ -176,13 +176,12 @@ func (t *createSnapshotFromFilesystemTask) Run(
 
 func (t *createSnapshotFromFilesystemTask) Cancel(
 	ctx context.Context,
-	execCtx tasks.ExecutionContext,
+	_ tasks.ExecutionContext,
 ) error {
 
 	_, err := t.storage.DeletingFilesystemSnapshot(
 		ctx,
 		t.request.GetSnapshotId(),
-		execCtx.GetTaskID(),
 	)
 	return err
 }

cloud/disk_manager/internal/pkg/dataplane/filesystem/snapshot/delete_filesystem_snapshot_task.go

@@ -36,23 +36,21 @@ func (t *deleteFilesystemSnapshotTask) Load(request, state []byte) error {
 
 func (t *deleteFilesystemSnapshotTask) deleteFilesystemSnapshot(
 	ctx context.Context,
-	execCtx tasks.ExecutionContext,
 ) error {
 
 	_, err := t.storage.DeletingFilesystemSnapshot(
 		ctx,
 		t.request.SnapshotId,
-		execCtx.GetTaskID(),
 	)
 	return err
 }
 
 func (t *deleteFilesystemSnapshotTask) Run(
 	ctx context.Context,
-	execCtx tasks.ExecutionContext,
+	_ tasks.ExecutionContext,
 ) error {
 
-	err := t.deleteFilesystemSnapshot(ctx, execCtx)
+	err := t.deleteFilesystemSnapshot(ctx)
 	if err != nil {
 		return errors.NewRetriableErrorWithIgnoreRetryLimit(err)
 	}
@@ -62,10 +60,10 @@ func (t *deleteFilesystemSnapshotTask) Run(
 
 func (t *deleteFilesystemSnapshotTask) Cancel(
 	ctx context.Context,
-	execCtx tasks.ExecutionContext,
+	_ tasks.ExecutionContext,
 ) error {
 
-	return t.deleteFilesystemSnapshot(ctx, execCtx)
+	return t.deleteFilesystemSnapshot(ctx)
 }
 
 func (t *deleteFilesystemSnapshotTask) GetMetadata(

cloud/disk_manager/internal/pkg/dataplane/filesystem/snapshot/storage/mocks/storage_mock.go

@@ -40,10 +40,9 @@ func (s *StorageMock) FilesystemSnapshotCreated(
 func (s *StorageMock) DeletingFilesystemSnapshot(
 	ctx context.Context,
 	snapshotID string,
-	taskID string,
 ) (*storage.FilesystemSnapshotMeta, error) {
 
-	args := s.Called(ctx, snapshotID, taskID)
+	args := s.Called(ctx, snapshotID)
 	return args.Get(0).(*storage.FilesystemSnapshotMeta), args.Error(1)
 }
 
@@ -99,23 +98,23 @@ func (s *StorageMock) GetTotalFilesystemSnapshotStorageSize(ctx context.Context)
 	return args.Get(0).(uint64), args.Error(1)
 }
 
-func (s *StorageMock) LockFilesystemSnapshot(
+func (s *StorageMock) AcquireFilesystemSnapshotBarrier(
 	ctx context.Context,
 	snapshotID string,
-	lockTaskID string,
-) (locked bool, err error) {
+	taskID string,
+) error {
 
-	args := s.Called(ctx, snapshotID, lockTaskID)
-	return args.Get(0).(bool), args.Error(1)
+	args := s.Called(ctx, snapshotID, taskID)
+	return args.Error(0)
 }
 
-func (s *StorageMock) UnlockFilesystemSnapshot(
+func (s *StorageMock) ReleaseFilesystemSnapshotBarrier(
 	ctx context.Context,
 	snapshotID string,
-	lockTaskID string,
+	taskID string,
 ) error {
 
-	args := s.Called(ctx, snapshotID, lockTaskID)
+	args := s.Called(ctx, snapshotID, taskID)
 	return args.Error(0)
 }
 

cloud/disk_manager/internal/pkg/dataplane/filesystem/snapshot/storage/schema/schema.go

@@ -32,6 +32,22 @@ func Create(
 	}
 	logging.Info(ctx, "Created filesystem_snapshots table")
 
+	err = db.CreateOrAlterTable(
+		ctx,
+		storageFolder,
+		"filesystem_snapshot_lock_holders",
+		persistence.NewCreateTableDescription(
+			persistence.WithColumn("snapshot_id", persistence.Optional(persistence.TypeUTF8)),
+			persistence.WithColumn("lock_task_id", persistence.Optional(persistence.TypeUTF8)),
+			persistence.WithPrimaryKeyColumn("snapshot_id", "lock_task_id"),
+		),
+		dropUnusedColumns,
+	)
+	if err != nil {
+		return err
+	}
+	logging.Info(ctx, "Created filesystem_snapshot_lock_holders table")
+
 	err = db.CreateOrAlterTable(
 		ctx,
 		storageFolder,
@@ -176,7 +192,13 @@ func Drop(
 
 	logging.Info(ctx, "Dropping schema for dataplane filesystem snapshot storage in %v", db.AbsolutePath(storageFolder))
 
-	err := db.DropTable(ctx, storageFolder, "filesystem_snapshots")
+	err := db.DropTable(ctx, storageFolder, "filesystem_snapshot_lock_holders")
+	if err != nil {
+		return err
+	}
+	logging.Info(ctx, "Dropped filesystem_snapshot_lock_holders table")
+
+	err = db.DropTable(ctx, storageFolder, "filesystem_snapshots")
 	if err != nil {
 		return err
 	}

cloud/disk_manager/internal/pkg/dataplane/filesystem/snapshot/storage/storage.go

@@ -43,7 +43,6 @@ type Storage interface {
 	DeletingFilesystemSnapshot(
 		ctx context.Context,
 		snapshotID string,
-		taskID string,
 	) (*FilesystemSnapshotMeta, error)
 
 	GetFilesystemSnapshotsToDelete(
@@ -77,16 +76,21 @@ type Storage interface {
 		ctx context.Context,
 	) (storageSize uint64, err error)
 
-	LockFilesystemSnapshot(
+	// AcquireFilesystemSnapshotBarrier joins taskID to the snapshot's shared
+	// deletion barrier. Repeated calls with the same taskID are idempotent.
+	AcquireFilesystemSnapshotBarrier(
 		ctx context.Context,
 		snapshotID string,
-		lockTaskID string,
-	) (locked bool, err error)
+		taskID string,
+	) error
 
-	UnlockFilesystemSnapshot(
+	// ReleaseFilesystemSnapshotBarrier removes taskID from the shared deletion
+	// barrier. It is idempotent, including for deleting or missing snapshots;
+	// deletion is unblocked after the last holder exits.
+	ReleaseFilesystemSnapshotBarrier(
 		ctx context.Context,
 		snapshotID string,
-		lockTaskID string,
+		taskID string,
 	) error
 
 	GetFilesystemSnapshotMeta(