fix downgrade error handling

DowngradeAttemptedError is not correctly handled in defaultReplicator.
This commit fixes the issue and adds logging for cases when a downgrade
was actually attempted.

4 files changed, 77 insertions, 18 deletions

diff --git a/internal/praefect/datastore/repository_memory.go b/internal/praefect/datastore/repository_memory.go
index fb2e70c54..86925935d 100644
--- a/internal/praefect/datastore/repository_memory.go
+++ b/internal/praefect/datastore/repository_memory.go
@@ -163,11 +163,11 @@ func (m *MemoryRepositoryStore) GetReplicatedGeneration(ctx context.Context, vir
 
 	if targetGeneration != GenerationUnknown && targetGeneration >= sourceGeneration {
 		return 0, DowngradeAttemptedError{
-			virtualStorage:      virtualStorage,
-			relativePath:        relativePath,
-			storage:             target,
-			currentGeneration:   targetGeneration,
-			attemptedGeneration: sourceGeneration,
+			VirtualStorage:      virtualStorage,
+			RelativePath:        relativePath,
+			Storage:             target,
+			CurrentGeneration:   targetGeneration,
+			AttemptedGeneration: sourceGeneration,
 		}
 	}
 
diff --git a/internal/praefect/datastore/repository_postgres.go b/internal/praefect/datastore/repository_postgres.go
index c21e4dd6c..a5df50920 100644
--- a/internal/praefect/datastore/repository_postgres.go
+++ b/internal/praefect/datastore/repository_postgres.go
@@ -18,16 +18,16 @@ const GenerationUnknown = -1
 // DowngradeAttemptedError is returned when attempting to get the replicated generation for a source repository
 // that does not upgrade the target repository.
 type DowngradeAttemptedError struct {
-	virtualStorage      string
-	relativePath        string
-	storage             string
-	currentGeneration   int
-	attemptedGeneration int
+	VirtualStorage      string
+	RelativePath        string
+	Storage             string
+	CurrentGeneration   int
+	AttemptedGeneration int
 }
 
 func (err DowngradeAttemptedError) Error() string {
 	return fmt.Sprintf("attempted downgrading %q -> %q -> %q from generation %d to %d",
-		err.virtualStorage, err.relativePath, err.storage, err.currentGeneration, err.attemptedGeneration,
+		err.VirtualStorage, err.RelativePath, err.Storage, err.CurrentGeneration, err.AttemptedGeneration,
 	)
 }
 
@@ -38,7 +38,7 @@ type RepositoryNotExistsError struct {
 	storage        string
 }
 
-// Is checks whetehr the other errors is of the same type.
+// Is checks whether the other errors is of the same type.
 func (err RepositoryNotExistsError) Is(other error) bool {
 	_, ok := other.(RepositoryNotExistsError)
 	return ok
@@ -251,11 +251,11 @@ AND storage = ANY($3)
 
 	if targetGeneration != GenerationUnknown && targetGeneration >= sourceGeneration {
 		return 0, DowngradeAttemptedError{
-			virtualStorage:      virtualStorage,
-			relativePath:        relativePath,
-			storage:             target,
-			currentGeneration:   targetGeneration,
-			attemptedGeneration: sourceGeneration,
+			VirtualStorage:      virtualStorage,
+			RelativePath:        relativePath,
+			Storage:             target,
+			CurrentGeneration:   targetGeneration,
+			AttemptedGeneration: sourceGeneration,
 		}
 	}
 
diff --git a/internal/praefect/replicator.go b/internal/praefect/replicator.go
index 60e5a3a76..f31745381 100644
--- a/internal/praefect/replicator.go
+++ b/internal/praefect/replicator.go
@@ -57,7 +57,14 @@ func (dr defaultReplicator) Replicate(ctx context.Context, event datastore.Repli
 	if err != nil {
 		// Later generation might have already been replicated by an earlier replication job. If that's the case,
 		// we'll simply acknowledge the job. This also prevents accidental downgrades from happening.
-		if errors.Is(err, datastore.DowngradeAttemptedError{}) {
+		var downgradeErr datastore.DowngradeAttemptedError
+		if errors.As(err, &downgradeErr) {
+			message := "repository downgrade prevented"
+			if downgradeErr.CurrentGeneration == downgradeErr.AttemptedGeneration {
+				message = "target repository already on the same generation, skipping replication job"
+			}
+
+			dr.log.WithError(downgradeErr).Info(message)
 			return nil
 		}
 
diff --git a/internal/praefect/replicator_test.go b/internal/praefect/replicator_test.go
index 347637e4f..fd76f8529 100644
--- a/internal/praefect/replicator_test.go
+++ b/internal/praefect/replicator_test.go
@@ -193,6 +193,58 @@ gitaly_praefect_replication_jobs{change_type="update",gitaly_storage="backup",vi
 `)))
 }
 
+func TestReplicatorDowngradeAttempt(t *testing.T) {
+	ctx, cancel := testhelper.Context()
+	defer cancel()
+
+	rs := datastore.NewMemoryRepositoryStore(nil)
+	require.NoError(t, rs.SetGeneration(ctx, "virtual-storage-1", "relative-path-1", "gitaly-1", 0))
+	require.NoError(t, rs.SetGeneration(ctx, "virtual-storage-1", "relative-path-1", "gitaly-2", 0))
+
+	logger := testhelper.DiscardTestLogger(t)
+	hook := test.NewLocal(logger)
+	r := &defaultReplicator{log: logrus.NewEntry(logger), rs: rs}
+
+	require.NoError(t, r.Replicate(ctx, datastore.ReplicationEvent{
+		Job: datastore.ReplicationJob{
+			VirtualStorage:    "virtual-storage-1",
+			RelativePath:      "relative-path-1",
+			SourceNodeStorage: "gitaly-1",
+			TargetNodeStorage: "gitaly-2",
+		},
+	}, nil, nil))
+
+	require.Equal(t, logrus.InfoLevel, hook.LastEntry().Level)
+	require.Equal(t, datastore.DowngradeAttemptedError{
+		VirtualStorage:      "virtual-storage-1",
+		RelativePath:        "relative-path-1",
+		Storage:             "gitaly-2",
+		CurrentGeneration:   0,
+		AttemptedGeneration: 0,
+	}, hook.LastEntry().Data["error"])
+	require.Equal(t, "target repository already on the same generation, skipping replication job", hook.LastEntry().Message)
+
+	require.NoError(t, rs.SetGeneration(ctx, "virtual-storage-1", "relative-path-1", "gitaly-2", 1))
+	require.NoError(t, r.Replicate(ctx, datastore.ReplicationEvent{
+		Job: datastore.ReplicationJob{
+			VirtualStorage:    "virtual-storage-1",
+			RelativePath:      "relative-path-1",
+			SourceNodeStorage: "gitaly-1",
+			TargetNodeStorage: "gitaly-2",
+		},
+	}, nil, nil))
+
+	require.Equal(t, logrus.InfoLevel, hook.LastEntry().Level)
+	require.Equal(t, datastore.DowngradeAttemptedError{
+		VirtualStorage:      "virtual-storage-1",
+		RelativePath:        "relative-path-1",
+		Storage:             "gitaly-2",
+		CurrentGeneration:   1,
+		AttemptedGeneration: 0,
+	}, hook.LastEntry().Data["error"])
+	require.Equal(t, "repository downgrade prevented", hook.LastEntry().Message)
+}
+
 func TestPropagateReplicationJob(t *testing.T) {
 	primaryServer, primarySocketPath, cleanup := runMockRepositoryServer(t)
 	defer cleanup()