Garnet Storage v2

Version.props

@@ -1,6 +1,6 @@
 <Project>
 	<!-- VersionPrefix property for builds and packages -->
 	<PropertyGroup>
-		<VersionPrefix>1.0.83</VersionPrefix>
+		<VersionPrefix>1.0.84</VersionPrefix>
 	</PropertyGroup>
 </Project>

benchmark/BDN.benchmark/Custom/CustomTxnSet.cs

@@ -52,10 +52,10 @@ public override bool Prepare<TGarnetReadApi>(TGarnetReadApi api, ref CustomProce
             valueC = GetNextArg(ref procInput, ref offset);
             valueD = GetNextArg(ref procInput, ref offset);
 
-            AddKey(setA, LockType.Exclusive, isObject: false);
-            AddKey(setB, LockType.Exclusive, isObject: false);
-            AddKey(setC, LockType.Exclusive, isObject: false);
-            AddKey(setD, LockType.Exclusive, isObject: false);
+            AddKey(setA, LockType.Exclusive, StoreType.Main);
+            AddKey(setB, LockType.Exclusive, StoreType.Main);
+            AddKey(setC, LockType.Exclusive, StoreType.Main);
+            AddKey(setD, LockType.Exclusive, StoreType.Main);
 
             return true;
         }

libs/client/ClientSession/GarnetClientSessionReplicationExtensions.cs

@@ -31,6 +31,7 @@ public sealed unsafe partial class GarnetClientSession : IServerHook, IMessageCo
         /// <param name="aofBeginAddress"></param>
         /// <param name="aofTailAddress"></param>
         /// <returns></returns>
+        /// <seealso cref="T:Garnet.cluster.ClusterSession.NetworkClusterInitiateReplicaSync"/>
         public Task<string> ExecuteReplicaSync(string nodeId, string primary_replid, byte[] checkpointEntryData, long aofBeginAddress, long aofTailAddress)
         {
             var tcs = new TaskCompletionSource<string>(TaskCreationOptions.RunContinuationsAsynchronously);
@@ -112,6 +113,7 @@ public Task<string> ExecuteReplicaSync(string nodeId, string primary_replid, byt
         /// <param name="fileTokenBytes"></param>
         /// <param name="fileType"></param>
         /// <param name="data"></param>
+        /// <seealso cref="T:Garnet.cluster.ClusterSession.NetworkClusterSendCheckpointMetadata"/>
         public Task<string> ExecuteSendCkptMetadata(Memory<byte> fileTokenBytes, int fileType, Memory<byte> data)
         {
             var tcs = new TaskCompletionSource<string>(TaskCreationOptions.RunContinuationsAsynchronously);
@@ -178,6 +180,7 @@ public Task<string> ExecuteSendCkptMetadata(Memory<byte> fileTokenBytes, int fil
         /// <param name="startAddress"></param>
         /// <param name="data"></param>
         /// <param name="segmentId"></param>
+        /// <seealso cref="T:Garnet.cluster.ClusterSession.NetworkClusterSendCheckpointFileSegment"/>
         public Task<string> ExecuteSendFileSegments(Memory<byte> fileTokenBytes, int fileType, long startAddress, Span<byte> data, int segmentId = -1)
         {
             var tcs = new TaskCompletionSource<string>(TaskCreationOptions.RunContinuationsAsynchronously);
@@ -257,19 +260,19 @@ public Task<string> ExecuteSendFileSegments(Memory<byte> fileTokenBytes, int fil
         /// Signal replica to recover
         /// </summary>
         /// <param name="sendStoreCheckpoint"></param>
-        /// <param name="sendObjectStoreCheckpoint"></param>
         /// <param name="replayAOF"></param>
         /// <param name="primary_replid"></param>
         /// <param name="checkpointEntryData"></param>
         /// <param name="beginAddress"></param>
         /// <param name="tailAddress"></param>
         /// <returns></returns>
-        public Task<string> ExecuteBeginReplicaRecover(bool sendStoreCheckpoint, bool sendObjectStoreCheckpoint, bool replayAOF, string primary_replid, byte[] checkpointEntryData, long beginAddress, long tailAddress)
+        /// <seealso cref="T:Garnet.cluster.ClusterSession.NetworkClusterBeginReplicaRecover"/>
+        public Task<string> ExecuteBeginReplicaRecover(bool sendStoreCheckpoint, bool replayAOF, string primary_replid, byte[] checkpointEntryData, long beginAddress, long tailAddress)
         {
             var tcs = new TaskCompletionSource<string>(TaskCreationOptions.RunContinuationsAsynchronously);
             tcsQueue.Enqueue(tcs);
             byte* curr = offset;
-            int arraySize = 9;
+            int arraySize = 8;
 
             while (!RespWriteUtils.TryWriteArrayLength(arraySize, ref curr, end))
             {
@@ -303,46 +306,38 @@ public Task<string> ExecuteBeginReplicaRecover(bool sendStoreCheckpoint, bool se
             offset = curr;
 
             //4
-            while (!RespWriteUtils.TryWriteBulkString(sendObjectStoreCheckpoint ? "1"u8 : "0"u8, ref curr, end))
-            {
-                Flush();
-                curr = offset;
-            }
-            offset = curr;
-
-            //5
             while (!RespWriteUtils.TryWriteBulkString(replayAOF ? "1"u8 : "0"u8, ref curr, end))
             {
                 Flush();
                 curr = offset;
             }
             offset = curr;
 
-            //6
+            //5
             while (!RespWriteUtils.TryWriteAsciiBulkString(primary_replid, ref curr, end))
             {
                 Flush();
                 curr = offset;
             }
             offset = curr;
 
-            //7
+            //6
             while (!RespWriteUtils.TryWriteBulkString(checkpointEntryData, ref curr, end))
             {
                 Flush();
                 curr = offset;
             }
             offset = curr;
 
-            //8
+            //7
             while (!RespWriteUtils.TryWriteArrayItem(beginAddress, ref curr, end))
             {
                 Flush();
                 curr = offset;
             }
             offset = curr;
 
-            //9
+            //8
             while (!RespWriteUtils.TryWriteArrayItem(tailAddress, ref curr, end))
             {
                 Flush();
@@ -360,6 +355,7 @@ public Task<string> ExecuteBeginReplicaRecover(bool sendStoreCheckpoint, bool se
         /// </summary>
         /// <param name="syncMetadata"></param>
         /// <returns></returns>
+        /// <seealso cref="T:Garnet.cluster.ClusterSession.NetworkClusterAttachSync"/>
         public Task<string> ExecuteAttachSync(byte[] syncMetadata)
         {
             var tcs = new TaskCompletionSource<string>(TaskCreationOptions.RunContinuationsAsynchronously);
@@ -408,6 +404,7 @@ public Task<string> ExecuteAttachSync(byte[] syncMetadata)
         /// </summary>
         /// <param name="sourceNodeId"></param>
         /// <param name="isMainStore"></param>
+        /// <seealso cref="T:Garnet.cluster.ClusterSession.NetworkClusterSync"/>
         public void SetClusterSyncHeader(string sourceNodeId, bool isMainStore)
         {
             // Unlike Migration, where we don't know at the time of header initialization if we have a record or not, in Replication 

libs/cluster/Server/ClusterManagerSlotState.cs

@@ -13,10 +13,13 @@
 namespace Garnet.cluster
 {
     using BasicGarnetApi = GarnetApi<BasicContext<RawStringInput, SpanByteAndMemory, long, MainSessionFunctions,
-            /* MainStoreFunctions */ StoreFunctions<SpanByteComparer, SpanByteRecordDisposer>,
-            SpanByteAllocator<StoreFunctions<SpanByteComparer, SpanByteRecordDisposer>>>,
+            /* MainStoreFunctions */ StoreFunctions<SpanByteComparer, DefaultRecordDisposer>,
+            ObjectAllocator<StoreFunctions<SpanByteComparer, DefaultRecordDisposer>>>,
         BasicContext<ObjectInput, GarnetObjectStoreOutput, long, ObjectSessionFunctions,
             /* ObjectStoreFunctions */ StoreFunctions<SpanByteComparer, DefaultRecordDisposer>,
+            ObjectAllocator<StoreFunctions<SpanByteComparer, DefaultRecordDisposer>>>,
+        BasicContext<UnifiedStoreInput, GarnetUnifiedStoreOutput, long, UnifiedSessionFunctions,
+            /* UnifiedStoreFunctions */ StoreFunctions<SpanByteComparer, DefaultRecordDisposer>,
             ObjectAllocator<StoreFunctions<SpanByteComparer, DefaultRecordDisposer>>>>;
 
     /// <summary>
@@ -482,7 +485,7 @@ public static unsafe void DeleteKeysInSlotsFromMainStore(BasicGarnetApi BasicGar
                 var key = iter.Key;
                 var s = HashSlotUtils.HashSlot(key);
                 if (slots.Contains(s))
-                    _ = BasicGarnetApi.DELETE(PinnedSpanByte.FromPinnedSpan(key), StoreType.Main);
+                    _ = BasicGarnetApi.DELETE(PinnedSpanByte.FromPinnedSpan(key));
             }
         }
 
@@ -499,7 +502,7 @@ public static unsafe void DeleteKeysInSlotsFromObjectStore(BasicGarnetApi BasicG
                 var key = iterObject.Key;
                 var s = HashSlotUtils.HashSlot(key);
                 if (slots.Contains(s))
-                    _ = BasicGarnetApi.DELETE(PinnedSpanByte.FromPinnedSpan(key), StoreType.Object);
+                    _ = BasicGarnetApi.DELETE(PinnedSpanByte.FromPinnedSpan(key));
             }
         }
     }

libs/cluster/Server/ClusterManagerWorkerState.cs

@@ -100,6 +100,10 @@ public ReadOnlySpan<byte> TryReset(bool soft, int expirySeconds = 60)
             try
             {
                 SuspendConfigMerge();
+
+                // Reset recovery operations before proceeding with reset
+                clusterProvider.replicationManager.ResetRecovery();
+
                 var resp = CmdStrings.RESP_OK;
                 while (true)
                 {
@@ -113,8 +117,9 @@ public ReadOnlySpan<byte> TryReset(bool soft, int expirySeconds = 60)
                     this.clusterConnectionStore.CloseAll();
 
                     var newNodeId = soft ? current.LocalNodeId : Generator.CreateHexId();
-                    var address = current.LocalNodeIp;
-                    var port = current.LocalNodePort;
+                    var endpoint = clusterProvider.storeWrapper.GetClusterEndpoint();
+                    var address = endpoint.Address.ToString();
+                    var port = endpoint.Port;
 
                     var configEpoch = soft ? current.LocalNodeConfigEpoch : 0;
                     var expiry = DateTimeOffset.UtcNow.Ticks + TimeSpan.FromSeconds(expirySeconds).Ticks;

libs/cluster/Server/ClusterProvider.cs

@@ -16,10 +16,13 @@
 namespace Garnet.cluster
 {
     using BasicGarnetApi = GarnetApi<BasicContext<RawStringInput, SpanByteAndMemory, long, MainSessionFunctions,
-            /* MainStoreFunctions */ StoreFunctions<SpanByteComparer, SpanByteRecordDisposer>,
-            SpanByteAllocator<StoreFunctions<SpanByteComparer, SpanByteRecordDisposer>>>,
+            /* MainStoreFunctions */ StoreFunctions<SpanByteComparer, DefaultRecordDisposer>,
+            ObjectAllocator<StoreFunctions<SpanByteComparer, DefaultRecordDisposer>>>,
         BasicContext<ObjectInput, GarnetObjectStoreOutput, long, ObjectSessionFunctions,
             /* ObjectStoreFunctions */ StoreFunctions<SpanByteComparer, DefaultRecordDisposer>,
+            ObjectAllocator<StoreFunctions<SpanByteComparer, DefaultRecordDisposer>>>,
+        BasicContext<UnifiedStoreInput, GarnetUnifiedStoreOutput, long, UnifiedSessionFunctions,
+            /* UnifiedStoreFunctions */ StoreFunctions<SpanByteComparer, DefaultRecordDisposer>,
             ObjectAllocator<StoreFunctions<SpanByteComparer, DefaultRecordDisposer>>>>;
 
     /// <summary>
@@ -157,7 +160,6 @@ public void FlushConfig()
         public void FlushDB(bool unsafeTruncateLog = false)
         {
             storeWrapper.store.Log.ShiftBeginAddress(storeWrapper.store.Log.TailAddress, truncateLog: unsafeTruncateLog);
-            storeWrapper.objectStore?.Log.ShiftBeginAddress(storeWrapper.objectStore.Log.TailAddress, truncateLog: unsafeTruncateLog);
         }
 
         /// <inheritdoc />
@@ -171,12 +173,6 @@ public void SafeTruncateAOF(bool full, long CheckpointCoveredAofAddress, Guid st
             entry.metadata.storeCheckpointCoveredAofAddress = CheckpointCoveredAofAddress;
             entry.metadata.storePrimaryReplId = replicationManager.PrimaryReplId;
 
-            entry.metadata.objectStoreVersion = serverOptions.DisableObjects ? -1 : storeWrapper.objectStore.CurrentVersion;
-            entry.metadata.objectStoreHlogToken = serverOptions.DisableObjects ? default : objectStoreCheckpointToken;
-            entry.metadata.objectStoreIndexToken = serverOptions.DisableObjects ? default : objectStoreCheckpointToken;
-            entry.metadata.objectCheckpointCoveredAofAddress = CheckpointCoveredAofAddress;
-            entry.metadata.objectStorePrimaryReplId = replicationManager.PrimaryReplId;
-
             // Keep track of checkpoints for replica
             // Used to delete old checkpoints and cleanup and also cleanup during attachment to new primary
             replicationManager.AddCheckpointEntry(entry, full);
@@ -241,8 +237,6 @@ public MetricsItem[] GetReplicationInfo()
                 new("second_repl_offset", replication_offset2),
                 new("store_current_safe_aof_address", clusterEnabled ? replicationManager.StoreCurrentSafeAofAddress.ToString() : "N/A"),
                 new("store_recovered_safe_aof_address", clusterEnabled ? replicationManager.StoreRecoveredSafeAofTailAddress.ToString() : "N/A"),
-                new("object_store_current_safe_aof_address", clusterEnabled && !serverOptions.DisableObjects ? replicationManager.ObjectStoreCurrentSafeAofAddress.ToString() : "N/A"),
-                new("object_store_recovered_safe_aof_address", clusterEnabled && !serverOptions.DisableObjects ? replicationManager.ObjectStoreRecoveredSafeAofTailAddress.ToString() : "N/A"),
                 new("recover_status", replicationManager.currentRecoveryStatus.ToString()),
                 new("last_failover_state", !clusterEnabled ? FailoverUtils.GetFailoverStatus(FailoverStatus.NO_FAILOVER) : failoverManager.GetLastFailoverStatus())
             };
@@ -427,15 +421,10 @@ public void ExtractKeySpecs(RespCommandsInfo commandInfo, RespCommand cmd, ref S
         public void ClusterPublish(RespCommand cmd, ref Span<byte> channel, ref Span<byte> message)
             => clusterManager.TryClusterPublish(cmd, ref channel, ref message);
 
-        internal GarnetClusterCheckpointManager GetReplicationLogCheckpointManager(StoreType storeType)
+        internal GarnetClusterCheckpointManager GetReplicationLogCheckpointManager()
         {
             Debug.Assert(serverOptions.EnableCluster);
-            return storeType switch
-            {
-                StoreType.Main => (GarnetClusterCheckpointManager)storeWrapper.store.CheckpointManager,
-                StoreType.Object => (GarnetClusterCheckpointManager)storeWrapper.objectStore?.CheckpointManager,
-                _ => throw new Exception($"GetCkptManager: unexpected state {storeType}")
-            };
+            return (GarnetClusterCheckpointManager)storeWrapper.store.CheckpointManager;
         }
 
         /// <summary>

libs/cluster/Server/Migration/MigrateSessionKeys.cs

@@ -104,11 +104,8 @@ public bool MigrateKeys()
                     return false;
 
                 // Migrate object store keys
-                if (!clusterProvider.serverOptions.DisableObjects)
-                {
-                    if (!MigrateKeysFromObjectStore())
-                        return false;
-                }
+                if (!MigrateKeysFromObjectStore())
+                    return false;
             }
             catch (Exception ex)
             {

libs/cluster/Server/Migration/MigrateSessionSlots.cs

@@ -33,17 +33,6 @@ public async Task<bool> MigrateSlotsDriverInline()
             var success = await CreateAndRunMigrateTasks(StoreType.Main, storeBeginAddress, storeTailAddress, mainStorePageSize);
             if (!success) return false;
 
-            // Send object store
-            if (!clusterProvider.serverOptions.DisableObjects)
-            {
-                var objectStoreBeginAddress = clusterProvider.storeWrapper.objectStore.Log.BeginAddress;
-                var objectStoreTailAddress = clusterProvider.storeWrapper.objectStore.Log.TailAddress;
-                var objectStorePageSize = 1 << clusterProvider.serverOptions.ObjectStorePageSizeBits();
-                logger?.LogWarning("Object Store migrate scan range [{objectStoreBeginAddress}, {objectStoreTailAddress}]", objectStoreBeginAddress, objectStoreTailAddress);
-                success = await CreateAndRunMigrateTasks(StoreType.Object, objectStoreBeginAddress, objectStoreTailAddress, objectStorePageSize);
-                if (!success) return false;
-            }
-
             return true;
 
             async Task<bool> CreateAndRunMigrateTasks(StoreType storeType, long beginAddress, long tailAddress, int pageSize)