Include repository ID in jobs scheduled by the reconciler

Reconciler is the only other component that currently schedules
replication jobs along the request finalizers. As such, we need to
update it to produce replicaton jobs similar to those producer by
the request finalizers. Namely, this commit includes the repository's
ID in the replication jobs produced by the reconciler.

2 files changed, 27 insertions, 3 deletions

diff --git a/internal/praefect/reconciler/reconciler.go b/internal/praefect/reconciler/reconciler.go
index 308a7aff1..a37fd5a05 100644
--- a/internal/praefect/reconciler/reconciler.go
+++ b/internal/praefect/reconciler/reconciler.go
@@ -84,6 +84,7 @@ func (r *Reconciler) Run(ctx context.Context, ticker helper.Ticker) error {
 
 // job is an internal type for formatting log messages
 type job struct {
+	RepositoryID   int64   `json:"repository_id"`
 	Change         string  `json:"change"`
 	CorrelationID  string  `json:"correlation_id"`
 	VirtualStorage string  `json:"virtual_storage"`
@@ -148,10 +149,12 @@ healthy_storages AS (
 
 delete_jobs AS (
 	SELECT DISTINCT ON (virtual_storage, relative_path)
+		repositories.repository_id,
 		virtual_storage,
 		relative_path,
 		storage
 	FROM storage_repositories
+	JOIN repositories USING (virtual_storage, relative_path)
 	JOIN healthy_storages USING (virtual_storage, storage)
 	WHERE (
 		-- Only unassigned repositories should be targeted for deletion. If no assignment exists,
@@ -161,7 +164,7 @@ delete_jobs AS (
 		WHERE virtual_storage = storage_repositories.virtual_storage
 		AND   relative_path   = storage_repositories.relative_path
 	)
-	AND generation <= (
+	AND storage_repositories.generation <= (
 		-- Check whether the replica's generation is equal or lower than the generation of every assigned
 		-- replica of the repository. If so, then it is eligible for deletion.
 		SELECT MIN(COALESCE(generation, -1))
@@ -196,6 +199,7 @@ delete_jobs AS (
 	-- repository remains on storage unused (because it doesn't exist in the 'repositories' table anymore).
 	SELECT * FROM (
 		SELECT DISTINCT ON (virtual_storage, relative_path)
+			0 AS repository_id,
 			virtual_storage,
 			relative_path,
 			storage
@@ -219,12 +223,13 @@ delete_jobs AS (
 
 update_jobs AS (
 	SELECT DISTINCT ON (virtual_storage, relative_path, target_node_storage)
+		repository_id,
 		virtual_storage,
 		relative_path,
 		source_node_storage,
 		target_node_storage
 	FROM (
-		SELECT virtual_storage, relative_path, storage AS target_node_storage
+		SELECT repositories.repository_id, virtual_storage, relative_path, storage AS target_node_storage
 		FROM repositories
 		JOIN healthy_storages USING (virtual_storage)
 		LEFT JOIN storage_repositories USING (virtual_storage, relative_path, storage)
@@ -273,6 +278,7 @@ reconciliation_jobs AS (
 		jsonb_build_object('correlation_id', encode(random()::text::bytea, 'base64'))
 	FROM (
 		SELECT
+			COALESCE(repository_id, 0) AS repository_id,
 			virtual_storage,
 			relative_path,
 			source_node_storage,
@@ -281,6 +287,7 @@ reconciliation_jobs AS (
 		FROM update_jobs
 			UNION ALL
 		SELECT
+			COALESCE(repository_id, 0) AS repository_id,
 			virtual_storage,
 			relative_path,
 			NULL AS source_node_storage,
@@ -303,6 +310,7 @@ create_locks AS (
 
 SELECT
 	meta->>'correlation_id',
+	job->>'repository_id',
 	job->>'change',
 	job->>'virtual_storage',
 	job->>'relative_path',
@@ -326,6 +334,7 @@ FROM reconciliation_jobs
 		var j job
 		if err := rows.Scan(
 			&j.CorrelationID,
+			&j.RepositoryID,
 			&j.Change,
 			&j.VirtualStorage,
 			&j.RelativePath,
diff --git a/internal/praefect/reconciler/reconciler_test.go b/internal/praefect/reconciler/reconciler_test.go
index a76adb48f..a74a668f1 100644
--- a/internal/praefect/reconciler/reconciler_test.go
+++ b/internal/praefect/reconciler/reconciler_test.go
@@ -12,6 +12,7 @@ import (
 	"gitlab.com/gitlab-org/gitaly/v14/internal/helper"
 	"gitlab.com/gitlab-org/gitaly/v14/internal/middleware/metadatahandler"
 	"gitlab.com/gitlab-org/gitaly/v14/internal/praefect"
+	"gitlab.com/gitlab-org/gitaly/v14/internal/praefect/commonerr"
 	"gitlab.com/gitlab-org/gitaly/v14/internal/praefect/datastore"
 	"gitlab.com/gitlab-org/gitaly/v14/internal/praefect/datastore/glsql"
 	"gitlab.com/gitlab-org/gitaly/v14/internal/testhelper"
@@ -1078,7 +1079,11 @@ func TestReconciler(t *testing.T) {
 						if repo.generation >= 0 {
 							if !repoCreated {
 								repoCreated = true
-								require.NoError(t, rs.CreateRepository(ctx, virtualStorage, relativePath, storage, nil, nil, false, false))
+
+								id, err := rs.ReserveRepositoryID(ctx, virtualStorage, relativePath)
+								require.NoError(t, err)
+
+								require.NoError(t, rs.CreateRepository(ctx, id, virtualStorage, relativePath, storage, nil, nil, false, false))
 							}
 
 							require.NoError(t, rs.SetGeneration(ctx, virtualStorage, relativePath, storage, repo.generation))
@@ -1149,6 +1154,16 @@ func TestReconciler(t *testing.T) {
 				}
 			}
 
+			// Fill the expected reconciliation jobs with generated repository IDs.
+			for i, job := range tc.reconciliationJobs {
+				id, err := rs.GetRepositoryID(ctx, job.VirtualStorage, job.RelativePath)
+				if err != nil {
+					require.Equal(t, commonerr.NewRepositoryNotFoundError(job.VirtualStorage, job.RelativePath), err)
+				}
+
+				tc.reconciliationJobs[i].RepositoryID = id
+			}
+
 			// run reconcile in two concurrent transactions to ensure everything works
 			// as expected with multiple Praefect's reconciling at the same time
 			firstTx := db.Begin(t)