avoid cache invalidation when updating replicationOffset

Author: vazois

libs/cluster/Server/Replication/ReplicaOps/AOFReplay/ReplicaReplayDriver.cs

@@ -210,6 +210,7 @@ internal unsafe void ConsumeDirect(byte* record, int recordLength, long currentA
             ValidateSublogIndex(physicalSublogIdx);
             replicationManager.SetSublogReplicationOffset(physicalSublogIdx, currentAddress);
             var ptr = record;
+            var replicationOffset = currentAddress;
             // logger?.LogError("[{physicalSublogIdx}] = {currentAddress} -> {nextAddress}", physicalSublogIdx, currentAddress, nextAddress);
             while (ptr < record + recordLength)
             {
@@ -224,7 +225,7 @@ internal unsafe void ConsumeDirect(byte* record, int recordLength, long currentA
                     if (isCheckpointStart)
                     {
                         // This is safe to be updated in parallel given that each sublog replay taks will update its own slot with corresponding address of the checkpoint marker
-                        replicationManager.ReplicationCheckpointStartOffset[physicalSublogIdx] = replicationManager.GetSublogReplicationOffset(physicalSublogIdx);
+                        replicationManager.ReplicationCheckpointStartOffset[physicalSublogIdx] = replicationOffset;
                     }
                     entryLength += TsavoriteLog.UnsafeAlign(payloadLength);
                 }
@@ -240,11 +241,13 @@ internal unsafe void ConsumeDirect(byte* record, int recordLength, long currentA
                     entryLength += TsavoriteLog.UnsafeAlign(-payloadLength);
                 }
                 ptr += entryLength;
-                replicationManager.IncrementSublogReplicationOffset(physicalSublogIdx, entryLength);
+                replicationOffset += entryLength;
             }
             // logger?.LogError("[{physicalSublogIdx}] = {currentAddress} -> {nextAddress}", physicalSublogIdx, currentAddress, nextAddress);
 
-            if (replicationManager.GetSublogReplicationOffset(physicalSublogIdx) != nextAddress)
+            replicationManager.SetSublogReplicationOffset(physicalSublogIdx, replicationOffset);
+
+            if (replicationOffset != nextAddress)
             {
                 logger?.LogError("ReplicaReplayTask.Consume NextAddress Mismatch sublogIdx: {sublogIdx}; recordLength:{recordLength}; currentAddress:{currentAddress}; nextAddress:{nextAddress}; replicationOffset:{ReplicationOffset}", physicalSublogIdx, recordLength, currentAddress, nextAddress, replicationManager.ReplicationOffset[physicalSublogIdx]);
                 throw new GarnetException("Failed validating integrity of replay", LogLevel.Warning, clientResponse: false);