Fix two rare CI failures: ListPushPopStressTest host crash and VectorManager cleanup vs Reset() AVE

Two independent rare CI failures, both surfacing as `Test host process
crashed` and aborting the whole test run.

## 1. `ClusterVectorSetTests.MigrateVectorSetWhileModifyingAsync` — fatal `AccessViolationException` in `VectorManager` cleanup task

### Symptom

```
Passed Garnet.test.cluster.ClusterVectorSetTests.MigrateVectorSetWhileModifyingAsync [12 s]
Fatal error. System.AccessViolationException: Attempted to read or write protected memory.
   at Tsavorite.core.LogRecord.get_Info()
   at Tsavorite.core.LogRecord.get_AllocatedSize()
   at Tsavorite.core.ObjectScanIterator`2[...].GetPhysicalAddressAndAllocatedSize(...)
   at Tsavorite.core.ObjectScanIterator`2[...].GetNext()
   at Tsavorite.core.TsavoriteKVIterator`6[...].PushNext[...](...)
   at Tsavorite.core.TsavoriteKV`2[...].Iterate[...](MainSessionFunctions, ...)
   at Garnet.server.VectorManager+<RunCleanupTaskAsync>d__24.MoveNext()
The active test run was aborted. Reason: Test host process crashed
```

The AVE is a Corrupted-State Exception — `catch (Exception)` in
`RunCleanupTaskAsync` cannot suppress it; the runtime fails fast and the
test host crashes.

### Root cause

`Recovery.Reset()` → `hlogBase.Reset()` (in `AllocatorBase` and the
per-allocator overrides `SpanByte` / `Object` / `TsavoriteLog`) frees pages
by synchronously invoking `OnPagesClosed(...)` and a
`for (i in BufferSize) FreePage(i)` loop. Both paths ultimately call
`ReturnPage(index)`, which sets:

```csharp
pageArrays[index]   = default;
pagePointers[index] = default;   // ★ becomes 0
```

`Reset()`'s docstring promised *"WARNING: assumes that threads have drained
out at this point."* But Garnet's cluster re-attach paths invoke it on a
running store:

* `libs/cluster/Server/Replication/ReplicaOps/ReplicaDisklessSync.cs:100`
* `libs/cluster/Server/Replication/ReplicaOps/ReplicaDiskbasedSync.cs:136`

In both files `storeWrapper.Reset()` is called **before**
`SuspendPrimaryOnlyTasksAsync()`, and even that suspend only drains
`TaskManager` tasks — `VectorManager.cleanupTask` is independent and never
drained.

Once `pagePointers[i] = 0`, the iterator's `GetPhysicalAddress` returns
`0 + offset` — a tiny kernel-page address — and dereferencing it in
`*(RecordInfo*)physicalAddress` raises a fatal AVE.

### The exact interleaving

Production scenario in `MigrateVectorSetWhileModifyingAsync`:

1. Source primary migrates a slot containing a vector set → drops the index → `CleanupDroppedIndex` queues a cleanup-task scan on the source primary.
2. The drop AOF entry replicates to the source's replica, which replays it and **also** queues a cleanup-task scan on the replica.
3. Cluster topology change (post-migration, gossip, or any reason) triggers a replica re-attach → `ReplicaDisklessSync.ReplicateAttachAsync` / `ReplicaDiskbasedSync.ReplicateAttachAsync` calls `storeWrapper.Reset()`.
4. The replica's cleanup task is still mid-iterate over the main store → AVE.

Thread-level interleaving:

```
Thread A: VectorManager cleanup task                 Thread B: storeWrapper.Reset()
─────────────────────────────────────────            ─────────────────────────────────
loop session.Iterate(callbacks)
  PushNext → ObjectScanIterator.GetNext()
    epoch.Resume()              ◄── enter at epoch E
    headAddress = HeadAddress   (still old value)
    LoadPageIfNeeded(...)       (cur >= head → in-mem)
    physicalAddress =
        pagePointers[pageIdx] + offset
                                                      Recovery.Reset()
                                                        hlogBase.Reset()
                                                          HeadAddress ← TailAddress
                                                          OnPagesClosed(...)
                                                            FreePage(p)
                                                              ReturnPage(p)
                                                                pagePointers[p] = 0   ◄── ★
                                                          // override loop:
                                                          for i in BufferSize:
                                                            FreePage(i)
                                                              ReturnPage(i)
                                                                pagePointers[i] = 0
    *(RecordInfo*)physicalAddress  ◄── ☠ AVE
       (LogRecord.GetInfo /
        LogRecord.AllocatedSize)
```

### Why epoch protection didn't catch this

Tsavorite's normal eviction path defers page-freeing through:

```csharp
epoch.BumpCurrentEpoch(() => OnPagesClosed(newAddr));
```

`BumpCurrentEpoch` queues the action and only fires it after
`SafeToReclaimEpoch` has advanced past the prior epoch — i.e. after every
thread that was holding the prior epoch has either suspended or moved on.
That's why scan iterators are safe against normal eviction.

`Reset()` skipped that mechanism in two places:

1. `AllocatorBase.Reset()` invoked `OnPagesClosed(newBeginAddress)` directly.
2. The per-allocator overrides had a `for (i in BufferSize) FreePage(i)`
   loop that ran **after** `base.Reset()` returned — also without epoch
   protection. **This second loop is the actual point of failure**: even
   if `OnPagesClosed` were deferred, the leftover (tail) page is freed by
   the override loop while a reader could still be reading it.

### The fix (Tsavorite layer)

`AllocatorBase.Reset()` defers ALL page-close + page-free work through
`BumpCurrentEpoch` and waits on a `ManualResetEventSlim` signalled by the
deferred action — no polling:

```csharp
using var resetComplete = new ManualResetEventSlim(initialState: false);

// If caller was already epoch-protected, our prior epoch is what the action
// will be waiting on — release it before waiting and re-acquire after.
var wasProtected = epoch.ThisInstanceProtected();
if (!wasProtected)
    epoch.Resume();   // BumpCurrentEpoch requires a protected caller

try
{
    epoch.BumpCurrentEpoch(() =>
    {
        try
        {
            if (headShifted) OnPagesClosed(newBeginAddress);
            FreeAllAllocatedPages();
        }
        finally { resetComplete.Set(); }   // never deadlock if action throws
    });
}
finally { epoch.Suspend(); }   // unconditionally so the action can fire

resetComplete.Wait();

if (wasProtected) epoch.Resume();
```

Each per-allocator override (`SpanByte` / `Object` / `TsavoriteLog`) moves
its `FreePage(i)` loop into a new `FreeAllAllocatedPages()` virtual so the
loop runs inside the deferred action:

```csharp
public override void Reset() { base.Reset(); Initialize(); }

protected override void FreeAllAllocatedPages()
{
    for (int index = 0; index < BufferSize; index++)
        if (IsAllocated(index)) FreePage(index);
}
```

### Why this is safe

* The deferred action runs only after `SafeToReclaimEpoch ≥ priorEpoch`,
  i.e. after every iterator that was inside `GetNext` at the moment
  `Reset()` was called has either suspended or advanced. By the time
  `pagePointers[i] = 0` executes, no thread is reading `pagePointers[i]`.
* Iterators that re-enter `GetNext` after `HeadAddress` was shifted see
  `currentAddress < headAddress` and route through the buffered disk frame
  instead of `pagePointers` — so they don't touch the cleared array.
* `Reset()` blocks until the deferred work has actually run, preserving
  its synchronous contract (the override's `Initialize()` after `Reset()`
  observes a fully freed page set).

### Test vs. product

Strictly, `Reset()`'s docstring put the burden on callers. The cluster
re-attach paths violate that — they call `Reset()` before draining the
`VectorManager` cleanup task, and `SuspendPrimaryOnlyTasksAsync()` doesn't
cover it. The alternative would be to drain every background reader at
every `Reset()` callsite, but we chose to make `Reset()` itself epoch-safe
because the contract was implicit, callsites are scattered, and Tsavorite
already has the right primitive (`epoch.BumpCurrentEpoch`) — the normal
eviction path uses it. This makes the safety property **enforced** rather
than **assumed**, and protects any future caller / background reader.

### Repro

`test/Garnet.test/VectorCleanupVsResetRaceTests.cs` — adds 4 000 vectors,
drops the set (queues a full-keyspace cleanup scan), then spams
`storeWrapper.Reset()` for 5 s.

* **Without the fix:** crashes the host on every iteration with the exact
  production stack (`LogRecord.get_Info` → `ObjectScanIterator.GetNext` →
  `VectorManager.RunCleanupTaskAsync`).
* **With the fix:** all 5 `[Repeat]` iterations pass (~2 700 resets per
  iteration concurrent with the cleanup iterator), no AVE.

## 2. `RespListTests.ListPushPopStressTest` — host crash on rare `RedisTimeoutException`

### Symptom

```
Unhandled exception. StackExchange.Redis.RedisTimeoutException: Timeout performing LPUSH (30000ms)
   at StackExchange.Redis.ConnectionMultiplexer.ExecuteSyncImpl[T](...)
   at StackExchange.Redis.RedisDatabase.ListLeftPush(...)
   at Garnet.test.RespListTests.<>c__DisplayClass39_1.<ListPushPopStressTest>b__0()
The active test run was aborted. Reason: Test host process crashed
```

### Root cause (two compounding issues)

1. **Worker threads created via `new Thread(() => ...)` had no try/catch.**
   In modern .NET an unhandled exception in a manually-created `Thread`
   terminates the process, so a single transient `RedisTimeoutException`
   aborted the entire test run.

2. **All 20 sync workers shared a single `ConnectionMultiplexer`.** Every
   command went through one socket and one background writer. Under CI
   load + lowMemory eviction overhead the writer falls behind and
   accumulates queued messages until SyncTimeout (30s) trips. The failure
   diagnostics confirmed this: `mc: 1/1, qs: 20, bw: SpinningDown`.

### Fix

* Pre-create one `ConnectionMultiplexer` per worker on the main thread.
  Each thread now owns its own socket, eliminating the single-writer
  bottleneck. Pre-creating also avoids a 20-way connect storm racing
  `ConnectTimeout`.
* Wrap each worker body in try/catch; capture exceptions into a
  `ConcurrentBag`, signal stop, exit cleanly. No more host crash.
* Throw the aggregate **before** the post-checks so a real timeout isn't
  masked by secondary "list not empty" assertion noise.
* Route the deadline-exceeded path through the failure bag too.

## Files

```
libs/storage/Tsavorite/cs/src/core/Allocator/AllocatorBase.cs           | 76 +++++++++++++++++++++++--
libs/storage/Tsavorite/cs/src/core/Allocator/ObjectAllocatorImpl.cs     |  7 ++-
libs/storage/Tsavorite/cs/src/core/Allocator/SpanByteAllocatorImpl.cs   |  7 ++-
libs/storage/Tsavorite/cs/src/core/Allocator/TsavoriteLogAllocatorImpl.cs |  7 ++-
test/Garnet.test/RespListTests.cs                                       | 124 +++++++++++++++++++++++++--------------
test/Garnet.test/VectorCleanupVsResetRaceTests.cs                       | new
```

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: badrishc

libs/storage/Tsavorite/cs/src/core/Allocator/AllocatorBase.cs

@@ -243,7 +243,14 @@ protected abstract void ReadAsync<TContext>(ulong alignedSourceAddress, IntPtr d
         /// <summary>Write page to device (async)</summary>
         protected abstract void WriteAsync<TContext>(long flushPage, DeviceIOCompletionCallback callback, PageAsyncFlushResult<TContext> asyncResult);
 
-        /// <summary>Reset the hybrid log. WARNING: assumes that threads have drained out at this point.</summary>
+        /// <summary>
+        /// Reset the hybrid log. Safe against concurrent iterators / readers: ALL page-close
+        /// and page-free work is deferred via <see cref="LightEpoch.BumpCurrentEpoch(Action)"/>
+        /// so it runs only after all in-flight epoch holders have moved past the current epoch.
+        /// Without this guard a concurrent iterator inside
+        /// <c>SpanByte/ObjectScanIterator.GetNext</c> could observe a freed
+        /// <c>pagePointers[index] == 0</c> and AVE on dereference.
+        /// </summary>
         [MethodImpl(MethodImplOptions.NoInlining)]
         public virtual void Reset()
         {
@@ -253,21 +260,59 @@ public virtual void Reset()
             _ = MonotonicUpdate(ref ReadOnlyAddress, newBeginAddress, out _);
             _ = MonotonicUpdate(ref SafeReadOnlyAddress, newBeginAddress, out _);
 
-            // Shift head address to tail
-            if (MonotonicUpdate(ref HeadAddress, newBeginAddress, out _))
-            {
-                // Close addresses
-                OnPagesClosed(newBeginAddress);
+            // Shift head address to tail. Once HeadAddress is at the new value, the iterator
+            // path checks `currentAddress < headAddress` in LoadPageIfNeeded and falls back to
+            // disk frame buffering instead of using pagePointers — so subsequent epoch entries
+            // by an iterator no longer touch the in-memory page pointers we are about to clear.
+            var headShifted = MonotonicUpdate(ref HeadAddress, newBeginAddress, out _);
+
+            // Defer ALL page closure / freeing through BumpCurrentEpoch so the work runs only
+            // AFTER current epoch holders (e.g. an iterator mid-GetNext) have suspended/advanced.
+            // This is the key safety property: the iterator's GetNext acquires the epoch around
+            // its read of pagePointers[index]; deferring the free past the current epoch ensures
+            // that read completes before the pointer is cleared.
+            //
+            // The action both (a) closes the partially-closed range (so OnPagesClosedWorker
+            // updates ClosedUntilAddress) and (b) frees any remaining allocated pages — which is
+            // what the per-allocator overrides used to do unsafely after base.Reset returned.
+            using var resetComplete = new ManualResetEventSlim(initialState: false);
+
+            // Remember whether the caller was already epoch-protected: we have to release that
+            // protection before waiting (otherwise the action — queued for our prior epoch —
+            // would never fire because we'd still be holding the prior epoch ourselves).
+            var wasProtected = epoch.ThisInstanceProtected();
+            if (!wasProtected)
+                epoch.Resume();   // BumpCurrentEpoch requires a protected caller
 
-                // Wait for pages to get closed
-                while (ClosedUntilAddress < newBeginAddress)
+            try
+            {
+                epoch.BumpCurrentEpoch(() =>
                 {
-                    _ = Thread.Yield();
-                    if (epoch.ThisInstanceProtected())
-                        epoch.ProtectAndDrain();
-                }
+                    try
+                    {
+                        if (headShifted)
+                            OnPagesClosed(newBeginAddress);
+                        FreeAllAllocatedPages();
+                    }
+                    finally
+                    {
+                        // Always signal so Reset() can never deadlock if the action throws.
+                        resetComplete.Set();
+                    }
+                });
+            }
+            finally
+            {
+                // Release the epoch unconditionally so the deferred action can fire (no-spin).
+                epoch.Suspend();
             }
 
+            resetComplete.Wait();
+
+            // Restore the caller's epoch state if they were protected on entry.
+            if (wasProtected)
+                epoch.Resume();
+
             // Update begin address to tail
             _ = MonotonicUpdate(ref BeginAddress, newBeginAddress, out _);
 
@@ -281,6 +326,13 @@ public virtual void Reset()
             device.Reset();
         }
 
+        /// <summary>
+        /// Free any pages still allocated after <see cref="OnPagesClosed"/> has run. Subclasses
+        /// override to call their per-allocator FreePage. Invoked from inside Reset's
+        /// epoch.BumpCurrentEpoch action so it is safe against concurrent iterators.
+        /// </summary>
+        protected virtual void FreeAllAllocatedPages() { }
+
         /// <summary>Asynchronously wraps <see cref="TruncateUntilAddressBlocking(long)"/>.</summary>
         internal void TruncateUntilAddress(long toAddress) => _ = Task.Run(() => TruncateUntilAddressBlocking(toAddress));
 

libs/storage/Tsavorite/cs/src/core/Allocator/ObjectAllocatorImpl.cs

@@ -117,12 +117,17 @@ protected internal override void Initialize()
         public override void Reset()
         {
             base.Reset();
+            Initialize();
+        }
+
+        /// <inheritdoc />
+        protected override void FreeAllAllocatedPages()
+        {
             for (var index = 0; index < BufferSize; index++)
             {
                 if (IsAllocated(index))
                     FreePage(index);
             }
-            Initialize();
         }
 
         /// <summary>Allocate memory page, pinned in memory, and in sector aligned form, if possible</summary>

libs/storage/Tsavorite/cs/src/core/Allocator/SpanByteAllocatorImpl.cs

@@ -26,12 +26,17 @@ public SpanByteAllocatorImpl(AllocatorSettings settings, TStoreFunctions storeFu
         public override void Reset()
         {
             base.Reset();
+            Initialize();
+        }
+
+        /// <inheritdoc />
+        protected override void FreeAllAllocatedPages()
+        {
             for (int index = 0; index < BufferSize; index++)
             {
                 if (IsAllocated(index))
                     FreePage(index);
             }
-            Initialize();
         }
 
         /// <summary>Allocate memory page, pinned in memory, and in sector aligned form, if possible</summary>

libs/storage/Tsavorite/cs/src/core/Allocator/TsavoriteLogAllocatorImpl.cs

@@ -27,12 +27,17 @@ public TsavoriteLogAllocatorImpl(AllocatorSettings settings)
         public override void Reset()
         {
             base.Reset();
+            Initialize();
+        }
+
+        /// <inheritdoc />
+        protected override void FreeAllAllocatedPages()
+        {
             for (var index = 0; index < BufferSize; index++)
             {
                 if (IsAllocated(index))
                     FreePage(index);
             }
-            Initialize();
         }
 
         /// <summary>

test/Garnet.test/RespListTests.cs

@@ -2,6 +2,7 @@
 // Licensed under the MIT license.
 
 using System;
+using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Linq;
 using System.Text;
@@ -1091,59 +1092,106 @@ public void ListPushPopStressTest()
 
             var keyArray = keys.ToArray();
             var stop = new ManualResetEventSlim(false);
+            var failures = new ConcurrentBag<Exception>();
+
+            // Pre-create one ConnectionMultiplexer per worker thread so each "client" has its
+            // own socket. A single shared multiplexer serializes all sync writes through one
+            // background writer; under CI load + lowMemory eviction that writer can fall behind
+            // and accumulate enough queued commands to exceed SyncTimeout. Connecting up front
+            // (instead of inside each thread) also avoids a 20-way connect storm racing
+            // against ConfigurationOptions.ConnectTimeout.
+            var workerCount = keyCount * 2;
+            var workerMuxes = new ConnectionMultiplexer[workerCount];
+            for (int i = 0; i < workerCount; i++)
+                workerMuxes[i] = ConnectionMultiplexer.Connect(TestUtils.GetConfig());
 
-            // Use dedicated threads to avoid threadpool starvation on small CI runners.
-            var threads = new Thread[keyCount * 2];
-            for (int i = 0; i < keyCount; i++)
+            try
             {
-                var key = keyArray[i];
-
-                threads[i * 2] = new Thread(() =>
+                // Use dedicated threads to avoid threadpool starvation on small CI runners.
+                var threads = new Thread[workerCount];
+                for (int i = 0; i < keyCount; i++)
                 {
-                    for (int j = 0; j < ppCount && !stop.IsSet; j++)
-                        db.ListLeftPush(key, j);
-                })
-                { IsBackground = true };
+                    var key = keyArray[i];
+                    var pushDb = workerMuxes[i * 2].GetDatabase(0);
+                    var popDb = workerMuxes[i * 2 + 1].GetDatabase(0);
+
+                    threads[i * 2] = new Thread(() =>
+                    {
+                        try
+                        {
+                            for (int j = 0; j < ppCount && !stop.IsSet; j++)
+                                pushDb.ListLeftPush(key, j);
+                        }
+                        catch (Exception ex)
+                        {
+                            failures.Add(ex);
+                            stop.Set();
+                        }
+                    })
+                    { IsBackground = true };
+
+                    threads[i * 2 + 1] = new Thread(() =>
+                    {
+                        try
+                        {
+                            for (int j = 0; j < ppCount && !stop.IsSet; j++)
+                            {
+                                var value = popDb.ListRightPop(key);
+                                while (value.IsNull && !stop.IsSet)
+                                {
+                                    Thread.Sleep(1);
+                                    value = popDb.ListRightPop(key);
+                                }
+                                if (!stop.IsSet)
+                                    ClassicAssert.IsTrue((int)value >= 0 && (int)value < ppCount, "Pop value inconsistency");
+                            }
+                        }
+                        catch (Exception ex)
+                        {
+                            failures.Add(ex);
+                            stop.Set();
+                        }
+                    })
+                    { IsBackground = true };
+                }
 
-                threads[i * 2 + 1] = new Thread(() =>
+                foreach (var t in threads) t.Start();
+                try
                 {
-                    for (int j = 0; j < ppCount && !stop.IsSet; j++)
+                    var deadline = DateTime.UtcNow.AddMinutes(5);
+                    foreach (var t in threads)
                     {
-                        var value = db.ListRightPop(key);
-                        while (value.IsNull && !stop.IsSet)
+                        var remaining = deadline - DateTime.UtcNow;
+                        if (remaining <= TimeSpan.Zero || !t.Join(remaining))
                         {
-                            Thread.Sleep(1);
-                            value = db.ListRightPop(key);
+                            stop.Set();
+                            failures.Add(new TimeoutException("ListPushPopStressTest timed out after 5 minutes"));
+                            break;
                         }
-                        if (!stop.IsSet)
-                            ClassicAssert.IsTrue((int)value >= 0 && (int)value < ppCount, "Pop value inconsistency");
                     }
-                })
-                { IsBackground = true };
-            }
+                }
+                finally
+                {
+                    stop.Set();
+                    foreach (var t in threads)
+                        t.Join(TimeSpan.FromSeconds(30));
+                }
 
-            foreach (var t in threads) t.Start();
-            try
-            {
-                var deadline = DateTime.UtcNow.AddMinutes(5);
-                foreach (var t in threads)
+                // Surface worker failures BEFORE the post-checks so a transient timeout isn't
+                // masked by secondary "list not empty" assertion noise.
+                if (!failures.IsEmpty)
+                    throw new AggregateException("ListPushPopStressTest worker(s) failed", failures);
+
+                foreach (var key in keyArray)
                 {
-                    var remaining = deadline - DateTime.UtcNow;
-                    if (remaining <= TimeSpan.Zero || !t.Join(remaining))
-                        ClassicAssert.Fail("ListPushPopStressTest timed out after 5 minutes");
+                    var count = db.ListLength(key);
+                    ClassicAssert.AreEqual(0, count);
                 }
             }
             finally
             {
-                stop.Set();
-                foreach (var t in threads)
-                    t.Join(TimeSpan.FromSeconds(30));
-            }
-
-            foreach (var key in keyArray)
-            {
-                var count = db.ListLength(key);
-                ClassicAssert.AreEqual(0, count);
+                foreach (var mux in workerMuxes)
+                    mux?.Dispose();
             }
         }