Validate conflicts of pack-refs housekeeping task

This commit adds some more validation of the housekeeping task:
- The housekeeping task should only run as its own. There shouldn't be
  another data modification in the same transaction.
- There shouldn't be any other concurrent housekeeping task.

2 files changed, 382 insertions, 17 deletions

diff --git a/internal/gitaly/storage/storagemgr/transaction_manager.go b/internal/gitaly/storage/storagemgr/transaction_manager.go
index f8f4d93b6..32ff747d0 100644
--- a/internal/gitaly/storage/storagemgr/transaction_manager.go
+++ b/internal/gitaly/storage/storagemgr/transaction_manager.go
@@ -57,9 +57,14 @@ var (
 	// errAlternateAlreadyLinked is returned when attempting to set an alternate on a repository that
 	// already has one.
 	errAlternateAlreadyLinked = errors.New("repository already has an alternate")
-	// errPackRefsConflictRefDeletion is returned when there is a committed ref deletion before pack-refs task is
-	// committed. The transaction should be aborted.
+	// errPackRefsConflictRefDeletion is returned when there is a committed ref deletion before pack-refs
+	// task is committed. The transaction should be aborted.
 	errPackRefsConflictRefDeletion = errors.New("detected a conflict with reference deletion when committing packed-refs")
+	// errHousekeepingConflictOtherUpdates is returned when the transaction includes housekeeping alongside
+	// with other updates.
+	errHousekeepingConflictOtherUpdates = errors.New("housekeeping in the same transaction with other updates")
+	// errHousekeepingConflictConcurrent is returned when there are another concurrent housekeeping task.
+	errHousekeepingConflictConcurrent = errors.New("conflict with another concurrent housekeeping task")
 
 	// Below errors are used to error out in cases when updates have been staged in a read-only transaction.
 	errReadOnlyReferenceUpdates    = errors.New("reference updates staged in a read-only transaction")
@@ -67,6 +72,7 @@ var (
 	errReadOnlyCustomHooksUpdate   = errors.New("custom hooks update staged in a read-only transaction")
 	errReadOnlyRepositoryDeletion  = errors.New("repository deletion staged in a read-only transaction")
 	errReadOnlyObjectsIncluded     = errors.New("objects staged in a read-only transaction")
+	errReadOnlyHousekeeping        = errors.New("housekeeping in a read-only transaction")
 )
 
 // InvalidReferenceFormatError is returned when a reference name was invalid.
@@ -433,11 +439,21 @@ func (txn *Transaction) Commit(ctx context.Context) (returnedErr error) {
 			return errReadOnlyRepositoryDeletion
 		case txn.includedObjects != nil:
 			return errReadOnlyObjectsIncluded
+		case txn.runHousekeeping != nil:
+			return errReadOnlyHousekeeping
 		default:
 			return nil
 		}
 	}
 
+	if txn.runHousekeeping != nil && (txn.referenceUpdates != nil ||
+		txn.defaultBranchUpdate != nil ||
+		txn.customHooksUpdate != nil ||
+		txn.deleteRepository ||
+		txn.includedObjects != nil) {
+		return errHousekeepingConflictOtherUpdates
+	}
+
 	return txn.commit(ctx, txn)
 }
 
@@ -1347,13 +1363,11 @@ func (mgr *TransactionManager) processTransaction() (returnedErr error) {
 
 		if transaction.runHousekeeping != nil {
 			shouldStoreWALFiles = true
-			logEntry.Housekeeping = &gitalypb.LogEntry_Housekeeping{}
-
-			packRefsEntry, err := mgr.verifyPackRefs(mgr.ctx, transaction)
+			housekeepingEntry, err := mgr.verifyHousekeeping(mgr.ctx, transaction)
 			if err != nil {
 				return fmt.Errorf("verifying pack refs: %w", err)
 			}
-			logEntry.Housekeeping.PackRefs = packRefsEntry
+			logEntry.Housekeeping = housekeepingEntry
 		}
 
 		if shouldStoreWALFiles {
@@ -1782,6 +1796,38 @@ func (mgr *TransactionManager) verifyDefaultBranchUpdate(ctx context.Context, tr
 	return nil
 }
 
+// verifyHousekeeping verifies if all included housekeeping tasks can be performed. Although it's feasible for multiple
+// housekeeping tasks running at the same time, it's not guaranteed they are conflict-free. So, we need to ensure there
+// is no other concurrent housekeeping task. Each sub-task also needs specific verification.
+func (mgr *TransactionManager) verifyHousekeeping(ctx context.Context, transaction *Transaction) (*gitalypb.LogEntry_Housekeeping, error) {
+	mgr.mutex.Lock()
+	defer mgr.mutex.Unlock()
+
+	// Check for any concurrent housekeeping between this transaction's snapshot LSN and the latest appended LSN.
+	elm := mgr.committedEntries.Front()
+	for elm != nil {
+		entry := elm.Value.(*committedEntry)
+		if entry.lsn > transaction.snapshotLSN && entry.entry.RelativePath == transaction.relativePath {
+			if entry.entry.GetHousekeeping() != nil {
+				return nil, errHousekeepingConflictConcurrent
+			}
+			if entry.entry.GetRepositoryDeletion() != nil {
+				return nil, errHousekeepingConflictOtherUpdates
+			}
+		}
+		elm = elm.Next()
+	}
+
+	packRefsEntry, err := mgr.verifyPackRefs(mgr.ctx, transaction)
+	if err != nil {
+		return nil, fmt.Errorf("verifying pack refs: %w", err)
+	}
+
+	return &gitalypb.LogEntry_Housekeeping{
+		PackRefs: packRefsEntry,
+	}, nil
+}
+
 // verifyPackRefs verifies if the pack-refs housekeeping task can be logged. Ideally, we can just apply the packed-refs
 // file and prune the loose references. Unfortunately, there could be a ref modification between the time the pack-refs
 // command runs and the time this transaction is logged. Thus, we need to verify if the transaction conflicts with the
@@ -2366,11 +2412,6 @@ func (mgr *TransactionManager) applyHousekeeping(ctx context.Context, lsn LSN, l
 		// command does dir removal for us, but in staginge repository during preparation stage. In the actual
 		// repository,  we need to do it ourselves.
 		rootRefDir := filepath.Join(repositoryPath, "refs")
-		protectedDirs := map[string]struct{}{
-			rootRefDir:                         {},
-			filepath.Join(rootRefDir, "heads"): {},
-			filepath.Join(rootRefDir, "tags"):  {},
-		}
 		for dir := range modifiedDirs {
 			for dir != rootRefDir {
 				if isEmpty, err := isDirEmpty(dir); err != nil {
@@ -2387,12 +2428,8 @@ func (mgr *TransactionManager) applyHousekeeping(ctx context.Context, lsn LSN, l
 					break
 				}
 
-				// Do not remove refs/heads and refs/tags because it's likely command will re-create it
-				// right afterward.
-				if _, exist := protectedDirs[dir]; !exist {
-					if err := os.Remove(dir); err != nil {
-						return fmt.Errorf("removing empty ref dir: %w", err)
-					}
+				if err := os.Remove(dir); err != nil {
+					return fmt.Errorf("removing empty ref dir: %w", err)
 				}
 				dir = filepath.Dir(dir)
 			}
diff --git a/internal/gitaly/storage/storagemgr/transaction_manager_housekeeping_test.go b/internal/gitaly/storage/storagemgr/transaction_manager_housekeeping_test.go
index 440a00e75..6a925ce19 100644
--- a/internal/gitaly/storage/storagemgr/transaction_manager_housekeeping_test.go
+++ b/internal/gitaly/storage/storagemgr/transaction_manager_housekeeping_test.go
@@ -48,6 +48,14 @@ func generateHousekeepingTests(t *testing.T, ctx context.Context, testPartitionI
 		}
 	}
 
+	defaultRefs := []git.Reference{
+		{Name: "refs/heads/branch-1", Target: setup.Commits.Second.OID.String()},
+		{Name: "refs/heads/branch-2", Target: setup.Commits.Third.OID.String()},
+		{Name: "refs/heads/main", Target: setup.Commits.First.OID.String()},
+		{Name: "refs/tags/v1.0.0", Target: lightweightTag.String()},
+		{Name: "refs/tags/v2.0.0", Target: annotatedTag.OID.String()},
+	}
+
 	return []transactionTestCase{
 		{
 			desc:        "run pack-refs on a repository without packed-refs",
@@ -805,5 +813,325 @@ func generateHousekeepingTests(t *testing.T, ctx context.Context, testPartitionI
 				},
 			},
 		},
+		{
+			desc:        "housekeeping fails in read-only transaction",
+			customSetup: customSetup,
+			steps: steps{
+				StartManager{},
+				Begin{
+					RelativePath: setup.RelativePath,
+					ReadOnly:     true,
+				},
+				RunPackRefs{},
+				Commit{
+					ExpectedError: errReadOnlyHousekeeping,
+				},
+			},
+			expectedState: StateAssertion{
+				Repositories: RepositoryStates{
+					relativePath: {
+						DefaultBranch: "refs/heads/main",
+						References:    defaultRefs,
+					},
+				},
+			},
+		},
+		{
+			desc:        "housekeeping fails when there are other updates in transaction",
+			customSetup: customSetup,
+			steps: steps{
+				StartManager{},
+				Begin{
+					RelativePath: setup.RelativePath,
+				},
+				RunPackRefs{},
+				Commit{
+					ReferenceUpdates: ReferenceUpdates{
+						"refs/heads/main": {OldOID: setup.Commits.First.OID, NewOID: setup.Commits.Second.OID},
+					},
+					ExpectedError: errHousekeepingConflictOtherUpdates,
+				},
+			},
+			expectedState: StateAssertion{
+				Repositories: RepositoryStates{
+					relativePath: {
+						DefaultBranch: "refs/heads/main",
+						References:    defaultRefs,
+					},
+				},
+			},
+		},
+		{
+			desc:        "housekeeping transaction runs concurrently with another housekeeping transaction",
+			customSetup: customSetup,
+			steps: steps{
+				StartManager{},
+				Begin{
+					TransactionID: 1,
+					RelativePath:  setup.RelativePath,
+				},
+				RunPackRefs{
+					TransactionID: 1,
+				},
+				Begin{
+					TransactionID: 2,
+					RelativePath:  setup.RelativePath,
+				},
+				RunPackRefs{
+					TransactionID: 2,
+				},
+				Commit{
+					TransactionID: 1,
+				},
+				Commit{
+					TransactionID: 2,
+					ExpectedError: errHousekeepingConflictConcurrent,
+				},
+			},
+			expectedState: StateAssertion{
+				Database: DatabaseState{
+					string(keyAppliedLSN(setup.PartitionID)): LSN(1).toProto(),
+				},
+				Directory: directoryStateWithPackedRefs(1),
+				Repositories: RepositoryStates{
+					relativePath: {
+						DefaultBranch: "refs/heads/main",
+						References:    defaultRefs,
+						PackedRefs: &PackedRefsState{
+							PackedRefsContent: []string{
+								"# pack-refs with: peeled fully-peeled sorted ",
+								fmt.Sprintf("%s refs/heads/branch-1", setup.Commits.Second.OID.String()),
+								fmt.Sprintf("%s refs/heads/branch-2", setup.Commits.Third.OID.String()),
+								fmt.Sprintf("%s refs/heads/main", setup.Commits.First.OID.String()),
+								fmt.Sprintf("%s refs/tags/v1.0.0", lightweightTag.String()),
+								fmt.Sprintf("%s refs/tags/v2.0.0", annotatedTag.OID.String()),
+								fmt.Sprintf("^%s", setup.Commits.Diverging.OID.String()),
+							},
+							LooseReferences: map[git.ReferenceName]git.ObjectID{},
+						},
+					},
+				},
+			},
+		},
+		{
+			desc: "housekeeping transaction runs after another housekeeping transaction in other repository of a pool",
+			steps: steps{
+				RemoveRepository{},
+				StartManager{},
+				Begin{
+					TransactionID: 1,
+					RelativePath:  "pool",
+				},
+				CreateRepository{
+					TransactionID: 1,
+					References: map[git.ReferenceName]git.ObjectID{
+						"refs/heads/main": setup.Commits.First.OID,
+					},
+					Packs: [][]byte{setup.Commits.First.Pack},
+				},
+				Commit{
+					TransactionID: 1,
+				},
+				Begin{
+					TransactionID:       2,
+					RelativePath:        "member",
+					ExpectedSnapshotLSN: 1,
+				},
+				CreateRepository{
+					TransactionID: 2,
+					Alternate:     "../../pool/objects",
+				},
+				Commit{
+					TransactionID: 2,
+				},
+				Begin{
+					TransactionID:       3,
+					RelativePath:        "member",
+					ExpectedSnapshotLSN: 2,
+				},
+				Begin{
+					TransactionID:       4,
+					RelativePath:        "pool",
+					ExpectedSnapshotLSN: 2,
+				},
+				RunPackRefs{
+					TransactionID: 3,
+				},
+				RunPackRefs{
+					TransactionID: 4,
+				},
+				Commit{
+					TransactionID: 3,
+				},
+				Commit{
+					TransactionID: 4,
+				},
+			},
+			expectedState: StateAssertion{
+				Database: DatabaseState{
+					string(keyAppliedLSN(setup.PartitionID)): LSN(4).toProto(),
+				},
+				Repositories: RepositoryStates{
+					"pool": {
+						Objects: []git.ObjectID{
+							setup.ObjectHash.EmptyTreeOID,
+							setup.Commits.First.OID,
+						},
+						DefaultBranch: "refs/heads/main",
+						References: []git.Reference{
+							{Name: "refs/heads/main", Target: setup.Commits.First.OID.String()},
+						},
+						PackedRefs: &PackedRefsState{
+							PackedRefsContent: []string{
+								"# pack-refs with: peeled fully-peeled sorted ",
+								fmt.Sprintf("%s refs/heads/main", setup.Commits.First.OID.String()),
+							},
+							LooseReferences: map[git.ReferenceName]git.ObjectID{},
+						},
+					},
+					"member": {
+						Objects: []git.ObjectID{
+							setup.ObjectHash.EmptyTreeOID,
+							setup.Commits.First.OID,
+						},
+						Alternate: "../../pool/objects",
+					},
+				},
+				Directory: testhelper.DirectoryState{
+					"/":                  {Mode: fs.ModeDir | perm.PrivateDir},
+					"/wal":               {Mode: fs.ModeDir | perm.PrivateDir},
+					"/wal/1":             {Mode: fs.ModeDir | perm.PrivateDir},
+					"/wal/1/objects.idx": indexFileDirectoryEntry(setup.Config),
+					"/wal/1/objects.pack": packFileDirectoryEntry(
+						setup.Config,
+						[]git.ObjectID{
+							setup.ObjectHash.EmptyTreeOID,
+							setup.Commits.First.OID,
+						},
+					),
+					"/wal/1/objects.rev": reverseIndexFileDirectoryEntry(setup.Config),
+					"/wal/3":             {Mode: fs.ModeDir | perm.PrivateDir},
+					"/wal/3/packed-refs": packRefsDirectoryEntry(setup.Config),
+					"/wal/4":             {Mode: fs.ModeDir | perm.PrivateDir},
+					"/wal/4/packed-refs": packRefsDirectoryEntry(setup.Config),
+				},
+			},
+		},
+		{
+			desc:        "housekeeping transaction runs after another housekeeping transaction",
+			customSetup: customSetup,
+			steps: steps{
+				StartManager{},
+				Begin{
+					TransactionID: 1,
+					RelativePath:  setup.RelativePath,
+				},
+				RunPackRefs{
+					TransactionID: 1,
+				},
+				Commit{
+					TransactionID: 1,
+				},
+				Begin{
+					TransactionID:       2,
+					RelativePath:        setup.RelativePath,
+					ExpectedSnapshotLSN: 1,
+				},
+				RunPackRefs{
+					TransactionID: 2,
+				},
+				Commit{
+					TransactionID: 2,
+				},
+			},
+			expectedState: StateAssertion{
+				Database: DatabaseState{
+					string(keyAppliedLSN(setup.PartitionID)): LSN(2).toProto(),
+				},
+				Directory: testhelper.DirectoryState{
+					"/":                  {Mode: fs.ModeDir | perm.PrivateDir},
+					"/wal":               {Mode: fs.ModeDir | perm.PrivateDir},
+					"/wal/1":             {Mode: fs.ModeDir | perm.PrivateDir},
+					"/wal/1/packed-refs": packRefsDirectoryEntry(setup.Config),
+					"/wal/2":             {Mode: fs.ModeDir | perm.PrivateDir},
+					"/wal/2/packed-refs": packRefsDirectoryEntry(setup.Config),
+				},
+				Repositories: RepositoryStates{
+					relativePath: {
+						DefaultBranch: "refs/heads/main",
+						References:    defaultRefs,
+						PackedRefs: &PackedRefsState{
+							PackedRefsContent: []string{
+								"# pack-refs with: peeled fully-peeled sorted ",
+								fmt.Sprintf("%s refs/heads/branch-1", setup.Commits.Second.OID.String()),
+								fmt.Sprintf("%s refs/heads/branch-2", setup.Commits.Third.OID.String()),
+								fmt.Sprintf("%s refs/heads/main", setup.Commits.First.OID.String()),
+								fmt.Sprintf("%s refs/tags/v1.0.0", lightweightTag.String()),
+								fmt.Sprintf("%s refs/tags/v2.0.0", annotatedTag.OID.String()),
+								fmt.Sprintf("^%s", setup.Commits.Diverging.OID.String()),
+							},
+							LooseReferences: map[git.ReferenceName]git.ObjectID{},
+						},
+					},
+				},
+			},
+		},
+		{
+			desc:        "housekeeping transaction runs concurrently with a repository deletion",
+			customSetup: customSetup,
+			steps: steps{
+				StartManager{},
+				Begin{
+					TransactionID: 1,
+					RelativePath:  setup.RelativePath,
+				},
+				RunPackRefs{
+					TransactionID: 1,
+				},
+				Begin{
+					TransactionID: 2,
+					RelativePath:  setup.RelativePath,
+				},
+				Commit{
+					TransactionID:    2,
+					DeleteRepository: true,
+				},
+				Begin{
+					TransactionID:       3,
+					RelativePath:        setup.RelativePath,
+					ExpectedSnapshotLSN: 1,
+				},
+				CreateRepository{
+					TransactionID: 3,
+				},
+				Commit{
+					TransactionID: 3,
+				},
+				Commit{
+					TransactionID: 1,
+					ExpectedError: errHousekeepingConflictOtherUpdates,
+				},
+			},
+			expectedState: StateAssertion{
+				Database: DatabaseState{
+					string(keyAppliedLSN(setup.PartitionID)): LSN(2).toProto(),
+				},
+				Directory: testhelper.DirectoryState{
+					"/":    {Mode: fs.ModeDir | perm.PrivateDir},
+					"/wal": {Mode: fs.ModeDir | perm.PrivateDir},
+				},
+				Repositories: RepositoryStates{
+					relativePath: {
+						DefaultBranch: "refs/heads/main",
+						References:    nil,
+						PackedRefs: &PackedRefsState{
+							PackedRefsContent: []string{""},
+							LooseReferences:   map[git.ReferenceName]git.ObjectID{},
+						},
+						Objects: []git.ObjectID{},
+					},
+				},
+			},
+		},
 	}
 }