Tsavorite allocator - tighten the packing of pages

libs/cluster/Server/Replication/PrimaryOps/AofTaskStore.cs

@@ -41,7 +41,7 @@ public AofTaskStore(ClusterProvider clusterProvider, int initialSize = 1, ILogge
                 logPageSizeMask = logPageSize - 1;
                 if (clusterProvider.serverOptions.MainMemoryReplication)
                     clusterProvider.storeWrapper.appendOnlyFile.SafeTailShiftCallback = SafeTailShiftCallback;
-                TruncateLagAddress = clusterProvider.storeWrapper.appendOnlyFile.UnsafeGetReadOnlyLagAddress() - 2 * logPageSize;
+                TruncateLagAddress = clusterProvider.storeWrapper.appendOnlyFile.UnsafeGetReadOnlyAddressLagOffset() - 2 * logPageSize;
             }
             TruncatedUntil = 0;
         }

libs/cluster/Server/Replication/ReplicaOps/ReplicationReplicaAofSync.cs

@@ -38,13 +38,19 @@ public unsafe void ProcessPrimaryStream(byte* record, int recordLength, long pre
 
                 if (clusterProvider.serverOptions.MainMemoryReplication)
                 {
-                    var firstRecordLength = GetFirstAofEntryLength(record);
-                    if (previousAddress > ReplicationOffset ||
-                        currentAddress >= previousAddress + firstRecordLength)
+                    // If the incoming AOF chunk fits in the space between previousAddress and currentAddress (ReplicationOffset),
+                    // an enqueue will result in an offset mismatch. So, we have to first reset the AOF to point to currentAddress.
+                    if (currentAddress > previousAddress)
                     {
-                        logger?.LogWarning("MainMemoryReplication: Skipping from {ReplicaReplicationOffset} to {currentAddress}", ReplicationOffset, currentAddress);
-                        storeWrapper.appendOnlyFile.Initialize(currentAddress, currentAddress);
-                        ReplicationOffset = currentAddress;
+                        if (
+                            (currentAddress % (1 << storeWrapper.appendOnlyFile.UnsafeGetLogPageSizeBits()) != 0) || // the skip was to a non-page-boundary
+                            (currentAddress >= previousAddress + recordLength) // the skip will not be auto-handled by the AOF enqueue
+                            )
+                        {
+                            logger?.LogWarning("MainMemoryReplication: Skipping from {ReplicaReplicationOffset} to {currentAddress}", ReplicationOffset, currentAddress);
+                            storeWrapper.appendOnlyFile.Initialize(currentAddress, currentAddress);
+                            ReplicationOffset = currentAddress;
+                        }
                     }
                 }
 
@@ -56,15 +62,6 @@ public unsafe void ProcessPrimaryStream(byte* record, int recordLength, long pre
                     throw new GarnetException($"Before ProcessPrimaryStream: Replication offset mismatch: ReplicaReplicationOffset {ReplicationOffset}, aof.TailAddress {storeWrapper.appendOnlyFile.TailAddress}", LogLevel.Warning, clientResponse: false);
                 }
 
-                // If there is a gap between the local tail and incoming currentAddress, try to skip local AOF to the next page
-                if (currentAddress >= storeWrapper.appendOnlyFile.TailAddress + recordLength
-                    && storeWrapper.appendOnlyFile.GetPage(currentAddress) == storeWrapper.appendOnlyFile.GetPage(storeWrapper.appendOnlyFile.TailAddress) + 1)
-                {
-                    logger?.LogWarning("SkipPage from {previousAddress} to {currentAddress}, tail is {tailAddress}", previousAddress, currentAddress, storeWrapper.appendOnlyFile.TailAddress);
-                    storeWrapper.appendOnlyFile.UnsafeSkipPage();
-                    logger?.LogWarning("New tail after SkipPage is {tailAddress}", storeWrapper.appendOnlyFile.TailAddress);
-                }
-
                 // Enqueue to AOF
                 _ = clusterProvider.storeWrapper.appendOnlyFile?.UnsafeEnqueueRaw(new Span<byte>(record, recordLength), noCommit: clusterProvider.serverOptions.EnableFastCommit);
 

libs/server/Resp/RespServerSession.cs

@@ -432,7 +432,7 @@ private void ProcessMessages()
                         }
                         else
                         {
-                            if (CanServeSlot(cmd))
+                            if (clusterSession == null || CanServeSlot(cmd))
                                 _ = ProcessBasicCommands(cmd, ref basicGarnetApi);
                         }
                     }

libs/server/Resp/RespServerSessionSlotVerify.cs

@@ -2,6 +2,7 @@
 // Licensed under the MIT license.
 
 using System;
+using System.Diagnostics;
 using Garnet.common;
 
 namespace Garnet.server
@@ -33,9 +34,7 @@ bool NetworkKeyArraySlotVerify(Span<ArgSlice> keys, bool readOnly, int count = -
 
         bool CanServeSlot(RespCommand cmd)
         {
-            // If cluster is disable all commands
-            if (clusterSession == null)
-                return true;
+            Debug.Assert(clusterSession != null);
 
             // Verify slot for command if it falls into data command category
             if (!cmd.IsDataCommand())