Replication job acknowledge removes 'completed' and 'dead' events.

Benchmark refactored as 'maxRuns' parameter has no effect on execution time.
Log entry added to track events that would be removed on acknowledgment.

Closes: https://gitlab.com/gitlab-org/gitaly/-/issues/2941

1 files changed, 13 insertions, 10 deletions

diff --git a/internal/praefect/replicator.go b/internal/praefect/replicator.go
index 416c31cf3..60e5a3a76 100644
--- a/internal/praefect/replicator.go
+++ b/internal/praefect/replicator.go
@@ -556,23 +556,26 @@ func (r ReplMgr) startHealthUpdate(ctx context.Context, logger logrus.FieldLogge
 }
 
 func (r ReplMgr) handleNodeEvent(ctx context.Context, logger logrus.FieldLogger, shard nodes.Shard, target nodes.Node, event datastore.ReplicationEvent) datastore.JobState {
-	logger = logger.WithFields(logrus.Fields{
+	ctxLogger := logger.WithFields(logrus.Fields{
 		logWithReplJobID: event.ID,
 		logWithCorrID:    getCorrelationID(event.Meta),
 	})
-	logger.Info("processing replication job")
-	err := r.processReplicationEvent(ctx, event, shard, target.GetConnection())
-	if err == nil {
-		return datastore.JobStateCompleted
-	}
+	ctxLogger.Info("processing replication job")
+
+	if err := r.processReplicationEvent(ctx, event, shard, target.GetConnection()); err != nil {
+		ctxLogger.WithError(err).Error("replication job failed")
 
-	logger.WithError(err).Error("replication job failed")
+		if event.Attempt <= 0 {
+			logger.WithField("event", event).Info("handled event would be deleted")
+			return datastore.JobStateDead
+		}
 
-	if event.Attempt <= 0 {
-		return datastore.JobStateDead
+		return datastore.JobStateFailed
 	}
 
-	return datastore.JobStateFailed
+	logger.WithField("event", event).Info("handled event would be deleted")
+
+	return datastore.JobStateCompleted
 }
 
 func (r ReplMgr) processReplicationEvent(ctx context.Context, event datastore.ReplicationEvent, shard nodes.Shard, targetCC *grpc.ClientConn) error {