[BFTree] Add RangeIndex cluster migration

.github/workflows/ci.yml

@@ -159,7 +159,7 @@ jobs:
         os: [ ubuntu-latest, windows-latest ]
         framework: [ 'net8.0' , 'net10.0']
         configuration: [ 'Debug', 'Release' ]
-        test: [ 'Garnet.test.cluster', 'Garnet.test.cluster.migrate', 'Garnet.test.cluster.replication', 'Garnet.test.cluster.replication.tls', 'Garnet.test.cluster.replication.disklesssync', 'Garnet.test.cluster.replication.rangeindex', 'Garnet.test.cluster.vectorsets', 'Garnet.test.cluster.multilog' ]
+        test: [ 'Garnet.test.cluster', 'Garnet.test.cluster.migrate', 'Garnet.test.cluster.migrate.rangeindex', 'Garnet.test.cluster.replication', 'Garnet.test.cluster.replication.tls', 'Garnet.test.cluster.replication.disklesssync', 'Garnet.test.cluster.replication.rangeindex', 'Garnet.test.cluster.vectorsets', 'Garnet.test.cluster.multilog' ]
     if: needs.changes.outputs.garnet == 'true'
     steps:
       - name: Check out code
@@ -175,7 +175,7 @@ jobs:
       - name: Install dependencies
         run: dotnet restore
       - name: Run tests ${{ matrix.test }}
-        run: dotnet test test/cluster/${{ matrix.test }} -f ${{ matrix.framework }} --configuration ${{ matrix.configuration }} --logger "console;verbosity=detailed" --logger trx --results-directory "GarnetTestResults-${{ matrix.os }}-${{ matrix.framework }}-${{ matrix.configuration }}-${{ matrix.test }}" -- NUnit.DisplayName=FullName
+        run: dotnet test test/cluster/${{ matrix.test }} -f ${{ matrix.framework }} --configuration ${{ matrix.configuration }} --logger "console;verbosity=detailed" --logger trx --blame-crash --blame-crash-dump-type full --blame-hang-timeout 30m --blame-hang-dump-type full --results-directory "GarnetTestResults-${{ matrix.os }}-${{ matrix.framework }}-${{ matrix.configuration }}-${{ matrix.test }}" -- NUnit.DisplayName=FullName
         timeout-minutes: 45 
       - name: Upload test results
         uses: actions/upload-artifact@v7

Garnet.slnx

@@ -87,6 +87,7 @@
     <Project Path="test/standalone/Garnet.test.vectorset/Garnet.test.vectorset.csproj" />
     <Project Path="test/cluster/Garnet.test.cluster/Garnet.test.cluster.csproj" />
     <Project Path="test/cluster/Garnet.test.cluster.migrate/Garnet.test.cluster.migrate.csproj" />
+    <Project Path="test/cluster/Garnet.test.cluster.migrate.rangeindex/Garnet.test.cluster.migrate.rangeindex.csproj" />
     <Project Path="test/cluster/Garnet.test.cluster.multilog/Garnet.test.cluster.multilog.csproj" />
     <Project Path="test/cluster/Garnet.test.cluster.multilog.diskless/Garnet.test.cluster.multilog.diskless.csproj" />
     <Project Path="test/cluster/Garnet.test.cluster.replication/Garnet.test.cluster.replication.csproj" />

libs/client/ClientSession/GarnetClientSessionIncremental.cs

@@ -42,6 +42,12 @@ public enum MigrationRecordSpanType : byte
         /// Bespoke encoding for Vector Set indexes.
         /// </summary>
         VectorSetIndex = 3,
+
+        /// <summary>
+        /// Chunked serialization stream for a RangeIndex key during migration.
+        /// The receiver uses a state machine to track the in-progress stream.
+        /// </summary>
+        SerializedRangeIndexStream = 4,
     }
 
     public sealed unsafe partial class GarnetClientSession : IServerHook, IMessageConsumer

libs/cluster/Server/Migration/MigrateOperation.cs

@@ -5,6 +5,7 @@
 using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Diagnostics;
+using System.Text;
 using System.Threading.Tasks;
 using Garnet.client;
 using Garnet.server;
@@ -22,15 +23,20 @@ internal sealed partial class MigrateOperation
             public StoreScan storeScan;
 
             private readonly ConcurrentDictionary<byte[], byte[]> vectorSetsIndexKeysToMigrate;
+            private readonly ConcurrentDictionary<byte[], byte> rangeIndexKeysToMigrate;
 
             readonly MigrateSession session;
             readonly GarnetClientSession gcs;
             readonly LocalServerSession localServerSession;
 
             public GarnetClientSession Client => gcs;
 
+            public LocalServerSession LocalSession => localServerSession;
+
             public IEnumerable<KeyValuePair<byte[], byte[]>> VectorSets => vectorSetsIndexKeysToMigrate;
 
+            public IEnumerable<byte[]> RangeIndexKeys => rangeIndexKeysToMigrate.Keys;
+
             public void ThrowIfCancelled() => session._cts.Token.ThrowIfCancellationRequested();
 
             public bool Contains(int slot) => session._sslots.Contains(slot);
@@ -46,6 +52,8 @@ public bool ContainsNamespace(ReadOnlySpan<byte> namespaceBytes)
             public void EncounteredVectorSet(byte[] key, byte[] value)
             => vectorSetsIndexKeysToMigrate.TryAdd(key, value);
 
+            public void AddRangeIndexKey(byte[] key) => rangeIndexKeysToMigrate.TryAdd(key, 0);
+
             public MigrateOperation(MigrateSession session, Sketch sketch = null, int batchSize = 1 << 18)
             {
                 this.session = session;
@@ -55,6 +63,7 @@ public MigrateOperation(MigrateSession session, Sketch sketch = null, int batchS
                 storeScan = new StoreScan(this);
                 keysToDelete = [];
                 vectorSetsIndexKeysToMigrate = new(ByteArrayComparer.Instance);
+                rangeIndexKeysToMigrate = new(ByteArrayComparer.Instance);
             }
 
             public async ValueTask<bool> InitializeAsync()
@@ -127,18 +136,14 @@ public async Task<bool> TransmitSlotsAsync()
                 return true;
             }
 
-            public async Task<bool> TransmitKeysAsync(Dictionary<byte[], byte[]> vectorSetKeysToIgnore)
+            public async Task<bool> TransmitKeysAsync(Func<PinnedSpanByte, bool> shouldSkipKey)
             {
                 // Use this for both stores; main store will just use the SpanByteAndMemory directly. We want it to be outside iterations
                 // so we can reuse the SpanByteAndMemory.Memory across iterations.
                 // TODO: initialize 'output' based on gcs curr and end; make sure it has the initial part of the "send" set, and call gcs.IncrementRecordDirect().
                 //       This will still allow SBAM.Memory to be reused.
                 var output = new UnifiedOutput();
 
-#if NET9_0_OR_GREATER
-                var ignoreLookup = vectorSetKeysToIgnore.GetAlternateLookup<ReadOnlySpan<byte>>();
-#endif
-
                 try
                 {
                     var keys = sketch.Keys;
@@ -152,21 +157,9 @@ public async Task<bool> TransmitKeysAsync(Dictionary<byte[], byte[]> vectorSetKe
                         if (keys[i].Item2)
                             continue;
 
-                        var spanByte = keys[i].Item1;
-
-                        // Don't transmit if a Vector Set
-                        var isVectorSet =
-                            vectorSetKeysToIgnore.Count > 0 &&
-#if NET9_0_OR_GREATER
-                            ignoreLookup.ContainsKey(spanByte.ReadOnlySpan);
-#else
-                                vectorSetKeysToIgnore.ContainsKey(spanByte.ToArray());
-#endif
-
-                        if (isVectorSet)
-                        {
+                        // Skip keys that require special handling
+                        if (shouldSkipKey(keys[i].Item1))
                             continue;
-                        }
 
                         if (!await session.WriteOrSendRecordAsync(gcs, localServerSession, keys[i].Item1, ref input, ref output, out var status).ConfigureAwait(false))
                             return false;
@@ -283,7 +276,19 @@ public void DeleteVectorSet(PinnedSpanByte key)
 
                 var delRes = localServerSession.BasicGarnetApi.DELETE(key);
 
-                session.logger?.LogDebug("Deleting Vector Set {key} after migration: {delRes}", System.Text.Encoding.UTF8.GetString(key), delRes);
+                session.logger?.LogDebug("Deleting Vector Set {key} after migration: {delRes}", Encoding.UTF8.GetString(key), delRes);
+            }
+
+            /// <summary>
+            /// Delete a RangeIndex after migration. COPY option is not yet supported for RangeIndex keys.
+            /// </summary>
+            public void DeleteRangeIndex(PinnedSpanByte key)
+            {
+                if (session._copyOption)
+                    return;
+
+                var delRes = localServerSession.BasicGarnetApi.DELETE(key);
+                session.logger?.LogDebug("Deleted RangeIndex key {key} after migration: {delRes}", Encoding.UTF8.GetString(key), delRes);
             }
         }
     }

libs/cluster/Server/Migration/MigrateScanFunctions.cs

@@ -50,7 +50,12 @@ public bool Reader<TSourceLogRecord>(in TSourceLogRecord srcLogRecord, RecordMet
                     // Check if key belongs to slot that is being migrated and if it can be added to our buffer
                     if (migrateOperation.Contains(slot))
                     {
-                        if (srcLogRecord.RecordType == VectorManager.RecordType)
+                        if (srcLogRecord.RecordType == RangeIndexManager.RangeIndexRecordType)
+                        {
+                            // RangeIndex keys need out-of-band migration (snapshot + chunks)
+                            migrateOperation.AddRangeIndexKey(key.ToArray());
+                        }
+                        else if (srcLogRecord.RecordType == VectorManager.RecordType)
                         {
                             // We can't delete the vector set _yet_ nor can we migrate it, 
                             // we just need to remember it to migrate once the associated namespaces are all moved over

libs/cluster/Server/Migration/MigrateSession.RangeIndex.cs

@@ -0,0 +1,168 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT license.
+
+using System;
+using System.Buffers;
+using System.Collections.Generic;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using Garnet.client;
+using Garnet.common;
+using Garnet.server;
+using Microsoft.Extensions.Logging;
+using Tsavorite.core;
+
+namespace Garnet.cluster
+{
+    /// <summary>
+    /// RangeIndex migration support: source-side transmit driver.
+    /// </summary>
+    internal sealed partial class MigrateSession : IDisposable
+    {
+        /// <summary>
+        /// Transmit a single RangeIndex key to the destination node.
+        /// Uses <see cref="RangeIndexManager.SnapshotRangeIndexAndCreateReader"/> to obtain an async
+        /// migration reader that snapshots and streams the BfTree data.
+        /// </summary>
+        private async Task<bool> TransmitRangeIndexAsync(MigrateOperation migrateOperation, byte[] keyBytes, int chunkSize, CancellationToken cancellationToken)
+        {
+            if (!clusterProvider.serverOptions.EnableRangeIndexPreview)
+            {
+                logger?.LogError("TransmitRangeIndexAsync: RangeIndex feature is not enabled, skipping key {key}", Encoding.UTF8.GetString(keyBytes));
+                return false;
+            }
+
+            var rangeIndexManager = clusterProvider.storeWrapper.DefaultDatabase.RangeIndexManager;
+
+            var sessionClient = migrateOperation.Client;
+            var buffer = ArrayPool<byte>.Shared.Rent(chunkSize);
+            var transmitActivity = RangeIndexMigrationActivities.TransmitActivity.StartActivity();
+            try
+            {
+                using var reader = rangeIndexManager.SnapshotRangeIndexAndCreateReader(migrateOperation.LocalSession, keyBytes, chunkSize);
+                transmitActivity.OnSnapshotCompleted(reader.TotalFileBytes);
+
+                while (!reader.IsComplete)
+                {
+                    cancellationToken.ThrowIfCancellationRequested();
+
+                    var payloadLen = await reader.ReadNextChunkAsync(buffer, cancellationToken).ConfigureAwait(false);
+                    if (payloadLen == 0)
+                    {
+                        transmitActivity.OnError("Zero-length payload from reader");
+                        logger?.LogError("TransmitRangeIndexAsync: reader returned zero-length payload with a {Size}-byte buffer for key {key}", chunkSize, Encoding.UTF8.GetString(keyBytes));
+                        return false;
+                    }
+
+                    if (!await WriteOrSendRecordSpanAsync(sessionClient, MigrationRecordSpanType.SerializedRangeIndexStream, buffer.AsSpan(0, payloadLen)).ConfigureAwait(false))
+                    {
+                        transmitActivity.OnError("Failed to write chunk");
+                        logger?.LogError("TransmitRangeIndexAsync: failed to write chunk for key {key}", Encoding.UTF8.GetString(keyBytes));
+                        return false;
+                    }
+
+                    transmitActivity.OnChunkSent(payloadLen);
+                }
+
+                // Force flush and await ACK
+                if (!await HandleMigrateTaskResponseAsync(sessionClient.SendAndResetIterationBuffer()).ConfigureAwait(false))
+                {
+                    transmitActivity.OnError("Flush failed");
+                    logger?.LogError("TransmitRangeIndexAsync: flush failed for key {key}", Encoding.UTF8.GetString(keyBytes));
+                    return false;
+                }
+
+                return true;
+            }
+            catch (Exception ex)
+            {
+                transmitActivity.OnError(ex.ToString());
+                logger?.LogError(ex, "TransmitRangeIndexAsync: error during snapshot or transmission for key {key}", Encoding.UTF8.GetString(keyBytes));
+                return false;
+            }
+            finally
+            {
+                ArrayPool<byte>.Shared.Return(buffer);
+                transmitActivity.EndAndLogActivity(logger, keyBytes);
+            }
+        }
+
+        /// <summary>
+        /// Migrate a batch of RangeIndex keys with sketch protection.
+        /// Adds all keys to the sketch, transitions through TRANSMITTING → DELETING → MIGRATED
+        /// with epoch barriers, ensuring concurrent operations are properly gated.
+        /// </summary>
+        private async Task<bool> MigrateRangeIndexKeysAsync(MigrateOperation migrateOperation, HashSet<byte[]> rangeIndexKeys, CancellationToken cancellationToken)
+        {
+            var migrateActivity = RangeIndexMigrationActivities.MigrateActivity.StartActivity(rangeIndexKeys.Count);
+
+            try
+            {
+                logger?.LogWarning("MigrateRangeIndexKeysAsync: migrating {count} RangeIndex keys", rangeIndexKeys.Count);
+
+                // Add all RI keys to sketch during INITIALIZING (no gating yet)
+                migrateOperation.sketch.Clear();
+                migrateOperation.sketch.SetStatus(SketchStatus.INITIALIZING);
+                foreach (var key in rangeIndexKeys)
+                    migrateOperation.sketch.TryHashAndStore(key);
+
+                ExceptionInjectionHelper.TriggerException(ExceptionInjectionType.RangeIndex_Migration_Before_Transmitting);
+
+                // Block writes during snapshot + transmit
+                migrateOperation.sketch.SetStatus(SketchStatus.TRANSMITTING);
+                await WaitForConfigPropagationAsync().ConfigureAwait(false);
+                migrateActivity.OnTransmitting();
+
+                await ExceptionInjectionHelper.ResetAndWaitAsync(ExceptionInjectionType.RangeIndex_Migration_After_Transmitting).ConfigureAwait(false);
+
+                foreach (var key in rangeIndexKeys)
+                {
+                    cancellationToken.ThrowIfCancellationRequested();
+
+                    if (!await TransmitRangeIndexAsync(migrateOperation, key, RangeIndexManager.DefaultMigrationChunkSize, cancellationToken).ConfigureAwait(false))
+                    {
+                        migrateActivity.OnError($"failed to transmit key {Encoding.UTF8.GetString(key)}");
+                        logger?.LogError("MigrateRangeIndexKeysAsync: failed to migrate RangeIndex key {key}", Encoding.UTF8.GetString(key));
+                        return false;
+                    }
+                }
+
+                ExceptionInjectionHelper.TriggerException(ExceptionInjectionType.RangeIndex_Migration_Before_Deleting);
+
+                // Block reads + writes during delete
+                migrateOperation.sketch.SetStatus(SketchStatus.DELETING);
+                await WaitForConfigPropagationAsync().ConfigureAwait(false);
+                migrateActivity.OnDeleting();
+
+                await ExceptionInjectionHelper.ResetAndWaitAsync(ExceptionInjectionType.RangeIndex_Migration_After_Deleting).ConfigureAwait(false);
+
+                foreach (var key in rangeIndexKeys)
+                {
+                    try
+                    {
+                        unsafe
+                        {
+                            fixed (byte* keyPtr = key)
+                                migrateOperation.DeleteRangeIndex(PinnedSpanByte.FromPinnedPointer(keyPtr, key.Length));
+                        }
+                    }
+                    catch (Exception ex)
+                    {
+                        migrateActivity.OnError($"failed to delete key {Encoding.UTF8.GetString(key)}: {ex}");
+                        logger?.LogError(ex, "MigrateRangeIndexKeysAsync: failed to delete RangeIndex key {key} after migration", Encoding.UTF8.GetString(key));
+                        throw;
+                    }
+                }
+
+                return true;
+            }
+            finally
+            {
+                // Always clean up the sketch, even on failure, to unblock client operations
+                migrateOperation.sketch.Clear();
+                migrateActivity.EndAndLogActivity(logger);
+            }
+        }
+    }
+}