Fix GetNext hang: use cached SafeTailAddress with fallback to scan

GetNext was using only the cached SafeTailAddress, which could remain stale
when no consumer-side RefreshSafeTailAddress had run yet. This caused the
iterator to return false prematurely, and since no scan happened in GetNext,
replication consumers would hang waiting for records that were already safe
to read.

Fix: check the cached value first (O(1)). Only call RefreshSafeTailAddress
when the current address has caught up to the cached boundary. This preserves
the fast path (no scan while the cache is ahead) while ensuring correctness
when the cache is stale.

Also restore RefreshSafeTailAddress in WaitAsync's fast path for the same
reason — the cache-only check could miss records and enter SlowWait
unnecessarily.

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

Author: badrishc

libs/storage/Tsavorite/cs/src/core/TsavoriteLog/TsavoriteLogScanIterator.cs

@@ -144,10 +144,7 @@ public ValueTask<bool> WaitAsync(CancellationToken token = default)
                 return SlowWaitAsync(this, token);
             }
 
-            // Check the cached SafeTailAddress only (O(1)). The more expensive
-            // RefreshSafeTailAddress scan is deferred to SlowWaitUncommittedAsync where
-            // per-iterator backoff logic can throttle scan frequency.
-            if (NextAddress < tsavoriteLog.SafeTailAddress)
+            if (NextAddress < tsavoriteLog.SafeTailAddress || NextAddress < tsavoriteLog.RefreshSafeTailAddress())
                 return new ValueTask<bool>(true);
             return SlowWaitUncommittedAsync(token);
         }
@@ -752,9 +749,13 @@ private unsafe bool GetNextInternal(out long physicalAddress, out int entryLengt
                 if (disposed)
                     return false;
 
-                // Use the cached SafeTailAddress for the hot path. Refresh only happens when the
-                // iterator catches up to the cache, via WaitAsync / SlowWaitUncommittedAsync.
-                if ((currentAddress >= endAddress) || (currentAddress >= (scanUncommitted ? tsavoriteLog.SafeTailAddress : tsavoriteLog.CommittedUntilAddress)))
+                // For scanUncommitted: check cached SafeTailAddress first (O(1)). Only scan the
+                // epoch table via RefreshSafeTailAddress when caught up to the cache, avoiding
+                // the O(kTableSize) scan on every record.
+                var boundary = scanUncommitted
+                    ? (currentAddress < tsavoriteLog.SafeTailAddress ? tsavoriteLog.SafeTailAddress : tsavoriteLog.RefreshSafeTailAddress())
+                    : tsavoriteLog.CommittedUntilAddress;
+                if (currentAddress >= endAddress || currentAddress >= boundary)
                     return false;
 
                 if (currentAddress < _headAddress)
@@ -872,9 +873,12 @@ private unsafe bool ExpandGetNextInternal(long startPhysicalAddress, ref int tot
                 if (disposed)
                     return false;
 
-                // Use the cached SafeTailAddress for the hot path. Refresh only happens when the
-                // iterator catches up to the cache, via WaitAsync / SlowWaitUncommittedAsync.
-                if ((currentAddress >= endAddress) || (currentAddress >= (scanUncommitted ? tsavoriteLog.SafeTailAddress : tsavoriteLog.CommittedUntilAddress)))
+                // For scanUncommitted: check cached SafeTailAddress first (O(1)). Only scan the
+                // epoch table via RefreshSafeTailAddress when caught up to the cache.
+                var boundary2 = scanUncommitted
+                    ? (currentAddress < tsavoriteLog.SafeTailAddress ? tsavoriteLog.SafeTailAddress : tsavoriteLog.RefreshSafeTailAddress())
+                    : tsavoriteLog.CommittedUntilAddress;
+                if (currentAddress >= endAddress || currentAddress >= boundary2)
                     return false;
 
                 if (currentAddress < _headAddress)