Add AOF replication for RangeIndex (BF-Tree) migration by tiagonapoli · Pull Request #1911 · microsoft/garnet

tiagonapoli · 2026-07-05T23:40:37Z

What

Adds AOF replication for migrated RangeIndex (BF-Tree) keys, so a migrated
index reaches replicas and survives crash recovery with its full data — not just
an empty tree rebuilt from the stub.

Why

When a node receives a migrated RI key, PublishMigratedIndex created the store
record via a RICREATE RMW that auto-logged only the stub to the AOF. On AOF
replay, HandleRangeIndexCreateReplay rebuilds an empty tree from the stub
config, so the migrated data never reached secondaries or survived recovery
(the TODO(RI) at RangeIndexManager.Migration.cs). A BF-Tree file can also
exceed a single AOF page, so it must be chunked.

How

RISTREAM — a new internal, AOF-only pseudo-command (never parsed from the
network), intercepted in AofProcessor.StoreRMW alongside
RICREATE/RISET/RIDEL.
Primary/publish (ReplicateRangeIndexStream): the reassembled BfTree file is
streamed into the AOF as chunked RISTREAM entries (reusing
RangeIndexChunkedSerializer), each entry keyed by the RI key and no larger than
DefaultMigrationChunkSize; the final chunk is flagged.
Replay (HandleRangeIndexStreamReplay): a per-key
RangeIndexChunkedDeserializer reassembles the file across interleaved AOF
entries (all chunks for one key hash to a single virtual sublog, so they arrive
in order on one replay task), then publishes via PublishMigratedIndex — which
re-streams on a chained replica. Incomplete reassembly (crash mid-stream) is
dropped by CleanupIncompleteStreamReassembly.
RICREATE suppression: the local RICREATE RMW carries the
StreamedPublishLogArg sentinel so WriteLogRMW skips its auto-log. The
RISTREAM stream is the single AOF source of truth for a migrated key; keeping a
competing empty-tree RICREATE entry would cause data loss on replay in either
ordering (it would block the publish via the KeyExists gate, or clobber the
reassembled tree).

Source-side deletion of the migrated key already replicates via the normal
DELETE -> AOF path (out of scope here).

Tests

ClusterMigrateRangeIndexReplicatesToReplicaViaAof — a migrated DISK-backed RI
key replicates to the destination primary's replica via AOF; the replica serves
the full data.
ClusterMigrateRangeIndexRecoversFromAof — the destination primary reconstructs
the migrated key from its own AOF on restart.
Full migrate.rangeindex suite (21) green; RangeIndexChunkedSerializerTests
(57) and RangeIndexMigrationReceiveStateTests (2) green; full solution builds
with 0 warnings.

Notes

MIGRATE REPLACE for RI keys remains unsupported (pre-existing TODO).
Preview feature (--enable-range-index-preview).

Draft: opening for review; not yet linked to a tracking issue.

Stream a migrated RangeIndex's BfTree file into the AOF as chunked RISTREAM entries (reusing RangeIndexChunkedSerializer), reassemble and publish on AOF replay (replica replication and crash recovery). Suppress the local RICREATE AOF auto-log so the chunk stream is the single AOF source of truth for a migrated key. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

- Cluster test: migrated RI key replicates to the destination primary's replica via the chunked RISTREAM AOF stream. - Cluster test: migrated RI key is reconstructed from the destination primary's own AOF on crash recovery. - Document the RISTREAM AOF stream replication path and RICREATE auto-log suppression in the RangeIndex dev docs. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

…eam restart - Add RangeIndexAofStreamChunkSize server option (validated < AofPageSize), threaded through RangeIndexManager and the cluster test harness. - Mark the first chunk of a RISTREAM stream; on replay, a new stream's first chunk supersedes any incomplete per-key reassembly left by a failed/retried migration, so the retry reassembles cleanly. - Add RangeIndex_Migration_Throw_After_Stream_Chunk injection point. - Correct the misleading chained-replica re-stream comment/doc. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

…T5/T10)

tiagonapoli · 2026-07-06T00:19:33Z

+        /// replay) or <paramref name="storedProcMode"/> is set. Each chunk is one AOF entry no larger
+        /// than <paramref name="chunkSize"/>, which must fit within a single AOF page.
+        /// </summary>
+        internal unsafe void ReplicateRangeIndexStream(ReadOnlySpan<byte> key, ReadOnlySpan<byte> stub, string filePath,


This is only used in migratio path - why have a bool storedProcMode?

Good catch — removed it. storedProcMode was dead here: migration publish is only driven by the cluster migration-receive session (passes false) and by AOF replay (which always passes a null appendOnlyFile, already short-circuiting before the flag mattered). A migrated-index publish never runs inside a CustomTransactionProcedure, so there was no path that could pass true. Dropped the param from both ReplicateRangeIndexStream and PublishMigratedIndex in 7875d17.

- T2 interleaving, T7 failover-after-migration, T8 checkpoint+recovery. - RangeIndexStreamReplayTests: first-chunk reset of stale reassembly (retry), incomplete cleanup, malformed/truncated drop. - Drop the fragile migration-injection T3 and the unused injection point; add ProcessStreamChunk seam + PendingStreamReassemblyCount. - Remove lowMemory-dependent eviction cluster test (covered by design).

- T4: a replica joining after migration reconstructs the key from the primary AOF during initial sync (manual 2-primary + late attach). - T9: RangeIndexAofStreamChunkSize must be < AofPageSize (config test). - Document the configurable chunk size, first-chunk reassembly reset, and the no-chained-replication note; correct an outdated migration comment. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

The migrated-RangeIndex chunk stream was tagged as a fake RespCommand (RISTREAM), which consumed an ACL/permission-bit slot despite never being network-parseable. Reclassify it as a dedicated internal AofEntryType (RangeIndexStreamChunk = 0x80) dispatched directly by AofProcessor before the generic store paths, and route replay to HandleRangeIndexStreamReplay via a dedicated handler. Behavior is unchanged: same key + chunk + first/last flags on the wire. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

tiagonapoli · 2026-07-06T01:25:31Z

                VectorSetReplayTaskCount = VectorSetReplayTaskCount,
                VectorSetQuantizationTaskCount = VectorSetQuantizationTaskCount,
                EnableRangeIndexPreview = EnableRangeIndexPreview,
+                RangeIndexAofStreamChunkSize = string.IsNullOrEmpty(RangeIndexAofStreamChunkSize)


Default should be min(DefaultMigrationChunkSize, AofChunkSize - Overhead)
Also, should check if the RangeIndexAofStreamChunkSize + overhead will be enough to be in AOfChunkSize

Done in 7875d17. Each chunk becomes one AOF entry whose total size is chunk payload + framing (TsavoriteLog record header + AofHeader + length-prefixed key + StringInput framing), and ValidateAllocatedLength throws if that exceeds one AOF page — so the old chunk < AofPageSize check was insufficient.

Added RangeIndexManager.AofStreamChunkEntryOverhead (a conservative upper bound on that framing, incl. an allowance for the RI key name).

Default now resolves to min(DefaultMigrationChunkSize, AofPage - overhead) (empty config becomes a sentinel resolved in GetAofSettings, where the effective page size is known).

Explicit values are validated against chunk + overhead <= AofPage (error otherwise), and I added a guard that the page is large enough to hold MinChunkSize + overhead.

One nuance: the key name is variable-length, so the overhead bakes in a fixed key allowance rather than the exact key. If you'd prefer this be exact, I can add a runtime clamp in ReplicateRangeIndexStream using the actual key length + appendOnlyFile.Log.UnsafeGetLogPageSizeBits() — let me know.

tiagonapoli · 2026-07-06T01:31:30Z

+        /// Range Index migration stream chunk. Carries one chunk of a migrated RangeIndex's
+        /// serialized BfTree file. This is an internal AOF-only transport entry — it is never
+        /// produced from a client command and has no RESP command; it is enqueued during migration
+        /// publish and replayed via <c>RangeIndexManager.HandleRangeIndexStreamReplay</c>.


Summarized in 74d98bf — now two lines: what it is (a chunk of a migrated Range Index's BF-Tree file) and how it's replayed.

tiagonapoli · 2026-07-06T01:32:39Z

                rangeIndexManager = new RangeIndexManager(
                    riLogRoot: riLogRoot, cprDir: cprDir,
                    storeEpoch: storeEpoch,
+                    aofStreamChunkSize: serverOptions.RangeIndexAofStreamChunkSize,


aofStreamChunkSize is misleading - this should be specific to RangeIndex no?

Renamed the ctor arg to rangeIndexAofStreamChunkSize (and the RangeIndexManager parameter/field to match) in 74d98bf.

tiagonapoli · 2026-07-06T01:33:32Z

-            // otherwise we might loose consistency
-            if (header.opType != AofEntryType.StoreRMW)
+            // otherwise we might loose consistency.
+            // RangeIndexStreamChunk (formerly the StoreRMW-tagged RISTREAM pseudo-command) is


"formerly..." - why mention that if that's in the same PR that got overwritten

Removed the formerly RISTREAM breadcrumb in 74d98bf. You're right — RISTREAM only ever existed within this PR (introduced then replaced), so it never shipped and there's nothing for a future reader to reconcile.

tiagonapoli · 2026-07-06T01:36:44Z

+            RangeIndexManager rangeIndexManager,
+            RespServerSession activeServerSession)
+        {
+            if (rangeIndexManager == null)


Is it possible to check the EnableRangeIndexPreview flag instead?

rangeIndexManager here is db.RangeIndexManager, which is constructed only when EnableRangeIndexPreview is set (GarnetServer.cs) — so rangeIndexManager == null is exactly the !EnableRangeIndexPreview condition. I kept the null-check to stay consistent with the sibling RICREATE/RISET/RIDEL replay handlers right above (they use the same pattern), and because the flag isn't threaded into these static dispatch helpers. Happy to switch all four to an explicit EnableRangeIndexPreview check if you'd prefer that over the null-check convention — just let me know and I'll do it consistently.

tiagonapoli · 2026-07-06T01:37:44Z

+        /// smaller than the AOF page size so each chunk fits in one page. Defaults to
+        /// <see cref="DefaultMigrationChunkSize"/>.
+        /// </summary>
+        private readonly int aofStreamChunkSize;


Rename to something like rangeIndexAofStreamChunkSize - also summarize the comment

Renamed field aofStreamChunkSize -> rangeIndexAofStreamChunkSize and summarized the doc in 74d98bf.

tiagonapoli · 2026-07-06T01:39:33Z

                    parseState.InitializeWithArgument(stubSlice);

                    var input = new StringInput(RespCommand.RICREATE, ref parseState);
+                    // Suppress the auto-AOF-log for this RICREATE: the range index stream enqueued above is


newline before comment

Added the blank line in 74d98bf.

tiagonapoli · 2026-07-06T01:40:11Z

+        /// For a migrated key the chunked <see cref="AofEntryType.RangeIndexStreamChunk"/> stream is the single AOF
+        /// source of truth; a competing RICREATE entry (which carries only the stub → an empty tree on
+        /// replay) would race the stream and cause data loss regardless of replay ordering.


remove - summarize just saying it's used in the migration code path

Summarized in 74d98bf — now just states it suppresses auto-logging the migration-publish RICREATE (the stream is the single AOF source of truth) and that it's used only in the migration code path.

…ng overhead in chunk-size default/validation - ReplicateRangeIndexStream / PublishMigratedIndex: remove the storedProcMode parameter. Migration publish never runs inside a stored procedure, and the AOF-replay path always passes a null appendOnlyFile (which already short-circuits), so the flag was dead. - RangeIndexAofStreamChunkSize: each chunk becomes one AOF entry (chunk payload + framing overhead) that must fit in a single AOF page. Introduce AofStreamChunkEntryOverhead and (a) resolve the auto default to min(DefaultMigrationChunkSize, AofPage - overhead), and (b) validate an explicitly configured size against chunk + overhead <= AofPage, instead of the previous chunk < AofPage check that ignored per-entry overhead. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

- Rename RangeIndexManager ctor param + field aofStreamChunkSize -> rangeIndexAofStreamChunkSize (and the GarnetServer call site) so it is clear the size is RangeIndex-specific. - Summarize the RangeIndexStreamChunk enum doc, the StreamedPublishLogArg doc, and the rangeIndexAofStreamChunkSize field doc. - Drop the "formerly RISTREAM" breadcrumb in AofProcessor (RISTREAM only ever existed within this PR, never shipped). - Add a blank line before the RICREATE suppression comment. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

tiagonapoli · 2026-07-06T02:04:48Z

+                if (RangeIndexAofStreamChunkSize <= 0)
+                {
+                    // Auto default: largest of DefaultMigrationChunkSize and the page budget, capped so
+                    // chunk + overhead fits in one AOF page.
+                    RangeIndexAofStreamChunkSize = (int)Math.Min(RangeIndexManager.DefaultMigrationChunkSize, maxChunkBytes);
+                }


You should do this defaulting first and then check 'RangeIndexAofStreamChunkSize > maxChunkBytes'
What if the default is bigger than allowed

Restructured in 095ace7. The auto default is now resolved first — RangeIndexAofStreamChunkSize = min(DefaultMigrationChunkSize, maxChunkBytes) — and then an unconditional if (RangeIndexAofStreamChunkSize > maxChunkBytes) check runs (no longer an else if). Because the default is clamped to the page budget, it can never be bigger than allowed, so only an oversized explicit value trips the error. That covers the what if the default is bigger than allowed case directly.

tiagonapoli · 2026-07-06T02:06:28Z

+
+        /// <summary>
+        /// One chunk of a migrated Range Index's serialized BF-Tree file — an internal AOF-only
+        /// transport entry (no RESP command), replayed via <c>RangeIndexManager.HandleRangeIndexStreamReplay</c>.


(no RESP command) - remove

Removed in 095ace7.

tiagonapoli · 2026-07-06T02:08:00Z

+        /// <param name="rangeIndexAofStreamChunkSize">Maximum size (bytes) of each chunked
+        /// <see cref="AofEntryType.RangeIndexStreamChunk"/> AOF entry used to replicate a migrated index. Clamped
+        /// up to <see cref="RangeIndexChunkedSerializer.MinChunkSize"/>. Defaults to
+        /// <see cref="DefaultMigrationChunkSize"/>.</param>


Remove defaults to metnion and MinChunksize mention

Removed both the Defaults to... and Clamped up to MinChunkSize mentions from the param doc in 979efd2; documented the new ArgumentOutOfRangeException instead.

tiagonapoli · 2026-07-06T02:08:22Z

+            this.rangeIndexAofStreamChunkSize = rangeIndexAofStreamChunkSize < RangeIndexChunkedSerializer.MinChunkSize
+                ? RangeIndexChunkedSerializer.MinChunkSize
+                : rangeIndexAofStreamChunkSize;


Instead of clamp, throw invalid

Done in 979efd2 — the ctor now throws ArgumentOutOfRangeException when rangeIndexAofStreamChunkSize < MinChunkSize instead of clamping. (The auto default and explicit values are already >= MinChunkSize via the GarnetServerOptions validation, so this only fires on a genuinely invalid programmatic value — fail-fast at construction.)

Restructure the RangeIndexAofStreamChunkSize validation so the auto default is resolved first (clamped to min(DefaultMigrationChunkSize, AofPage - overhead)), then an unconditional chunk > maxChunkBytes check runs. The clamped default can never exceed the budget, so only an oversized explicit value errors — removing the else-if that let the default path skip the check. Also drop "(no RESP command)" from the RangeIndexStreamChunk enum doc. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

… clamping The RangeIndexManager ctor now throws ArgumentOutOfRangeException when rangeIndexAofStreamChunkSize is below MinChunkSize instead of silently clamping it up. Also drop the "Clamped up to MinChunkSize" and "Defaults to DefaultMigrationChunkSize" mentions from the parameter doc and document the new exception. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

tiagonapoli · 2026-07-06T02:18:50Z

+            try
+            {
+                using var fs = new FileStream(filePath, FileMode.Open, FileAccess.Read, FileShare.Read, bufferSize: chunkSize);
+
+                while (!serializer.IsComplete)
+                {
+                    var written = 0;
+                    while (!serializer.IsComplete && written < chunkSize)
+                    {
+                        if (serializer.NeedsFileData)
+                        {
+                            var bytesRead = fs.Read(readBuffer, 0, chunkSize);
+                            if (bytesRead == 0)
+                                throw new GarnetException($"RangeIndex file truncated while streaming to AOF: {serializer.FileDataRemaining} bytes remaining for key");
+                            serializer.SupplyFileData(readBuffer.AsMemory(0, bytesRead));
+                        }
+
+                        var n = serializer.MoveNext(destBuffer.AsSpan(written, chunkSize - written));
+
+                        // No progress: remaining space is too small for the next framing element
+                        // (header/trailer). Flush what we have and start a fresh chunk.
+                        if (n == 0)
+                            break;
+                        written += n;
+                    }
+
+                    if (written == 0)
+                        continue;
+
+                    EnqueueRangeIndexStreamChunk(appendOnlyFile, version, sessionId, key, destBuffer.AsSpan(0, written), isFirst, serializer.IsComplete);
+                    isFirst = false;
+                }


This looks like lots of duplicated code from migration? Possible to improve?

tiagonapoli · 2026-07-06T02:23:20Z

+
+            var deserializer = streamReassembly.GetOrAdd(keyArr, _ => new RangeIndexChunkedDeserializer(DeriveTempMigrationPath(), logger));
+
+            if (!deserializer.ProcessChunk(chunk) || deserializer.HasError)


Look at the activity types that exist for RI migration and create a new activity type for replication here
It will be in streamReassembly and once fully received then it will be complete and logged
Maybe in RemoveAndDisposeStreamReassembly - but will require a reason - either success or some failure

tiagonapoli · 2026-07-06T02:25:28Z

+        /// </summary>
+        internal void CleanupIncompleteStreamReassembly()
+        {
+            foreach (var kvp in streamReassembly)


foreach in concurrent collection can be a problem here?

Add GarnetLog.GetMaxAofEntryOverhead(keyLength, inputFramingBytes), which mirrors TsavoriteLog's Enqueue allocation (record header + AOF entry header + length-prefixed key + input framing + 8-byte alignment) using the larger sharded header. ReplicateRangeIndexStream now asks the log for the exact per-key overhead and clamps the chunk size so every AOF entry fits one page, regardless of key length or future AOF-framing changes. The static AofStreamChunkEntryOverhead constant is now only a startup estimate (default sizing + config validation, before any log exists); the runtime log-based overhead is authoritative. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Replace the magic-0 "unset" sentinel with a nullable int?: the option defaults to null (unset) and is resolved in GetAofSettings to min(DefaultMigrationChunkSize, AofPage - overhead). Explicitly configured values are validated as before. Clearer than overloading 0, which is never a legal chunk size anyway. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Tiago Napoli and others added 4 commits June 29, 2026 22:17

Add corner-case tests: multi-chunk stream, empty key, round-trip (T1/…

d74911d

…T5/T10)