Replace SubscribeEvictions with IRecordTriggers.OnEvict for heap-size tracking

Garnet's `CacheSizeTracker` used `LogAccessor.SubscribeEvictions` to drive a
`LogSizeTracker.OnNext` observer that, on every page eviction, allocated a
scan iterator and walked the page to sum `MemoryUtils.CalculateHeapMemorySize`
over each non-null / non-closed record. This is heavyweight for a hot path
(buffer-pool allocation, iterator bookkeeping, epoch resume/suspend, and a
virtual dispatch per page).

This PR migrates the per-page heap-size decrement onto the per-record
`IRecordTriggers.OnEvict` hook introduced in #1695, which the object
allocator already walks during `EvictRecordsInRange`. That collapses the
"scan iterator + observer + sum" path into a single per-record callback
directly on the record we would have visited anyway.

### Tsavorite changes

- Add `EvictionSource { MainLog, ReadCache }` and thread it through
  `IRecordTriggers.OnEvict`, `IStoreFunctions.OnEvict`, and
  `IAllocator.EvictRecordsInRange`. Garnet uses this to route decrements to
  the correct counter (`AddHeapSize` vs `AddReadCacheHeapSize`).
- Change `IRecordTriggers.CallOnEvict` / `IStoreFunctions.CallOnEvict` from
  a property to `CallOnEvict(EvictionSource)` so the allocator can skip the
  per-record eviction walk entirely when the application has no work on
  that side (e.g. a read-cache-only budget should not force walks of the
  main-log allocator).
- Add `AllocatorSettings.IsReadCache` and `AllocatorBase.IsReadCache`, set
  to `true` by `Tsavorite.cs` when constructing the read-cache allocator.
  This is the cleanest way to distinguish the two allocators at
  `OnPagesClosedWorker` time without relying on the `evictCallback`
  sentinel, and is used to pass the correct `EvictionSource` to `CallOnEvict`.
- `AllocatorBase.EvictPageForRecovery` routes through the per-record
  `EvictRecordsInRange` when `storeFunctions.CallOnEvict(source)` is set.
  The legacy `MemoryPageScan(observer)` path is preserved as a fallback
  for consumers that still use `SubscribeEvictions`.
- Collapse `AllocatorBase`'s constructor to accept `AllocatorSettings`
  directly instead of unpacking individual fields at each concrete
  allocator (`ObjectAllocatorImpl`, `SpanByteAllocatorImpl`,
  `TsavoriteLogAllocatorImpl`).
- Tighten `ObjectAllocatorImpl.EvictRecordsInRange` to match
  `ObjectScanIterator`'s single-page invariant: clip `stopAddress` to the
  start page, bail on `offset == 0`, and document that both callers
  (`OnPagesClosedWorker`, `EvictPageForRecovery`) hand single-page ranges.

### Garnet changes

- `GarnetRecordTriggers.CallOnEvict(EvictionSource)` returns true only for
  the sides that are actually configured (`mainLogTracker` and/or
  `readCacheTracker` non-null), avoiding pointless per-record walks on the
  untracked allocator.
- `GarnetRecordTriggers.OnEvict(ref LogRecord, EvictionSource)` computes
  `MemoryUtils.CalculateHeapMemorySize(in logRecord)` and dispatches to
  `AddHeapSize(-size)` or `AddReadCacheHeapSize(-size)` based on the
  source. This goes through the standard asserting path — the counter must
  never undershoot zero.
- `CacheSizeTracker.Initialize` replaces the two `SubscribeEvictions`
  calls with `SetLogSizeTracker` calls so the fast-path size tracking
  during `TryCopyToTail`, `TryCopyToReadCache`, and object-page growth
  (`UpdateSize`, `IncrementSize`) continues to work unchanged.

### MainStore heap-tracking fix (root cause of negative counter)

Routing the eviction decrement through the asserting `AddHeapSize` path
surfaced a pre-existing gap: `MainStore` record-creation paths never
emitted a positive heap-size bump when a record's key or value spilled to
overflow (large inline field backed by a heap-allocated byte array). The
legacy observer path masked this silently because `OnNext` did a raw
decrement with no assertion — the counter quietly went negative on every
HLL sparse→dense transition (≈-12 KB per key) and accumulated drift
elsewhere.

`MainStore` now emits balanced heap accounting at all create/update/delete
entry points that go through a typed function callback rather than the
shared `SessionFunctionsUtils` writer (which already tracks deltas):

- `RMWMethods.PostInitialUpdater`:
  `+logRecord.CalculateHeapMemorySize()` for the freshly-created record.
- `RMWMethods.PostCopyUpdater`: `+dstLogRecord.CalculateHeapMemorySize()`
  for the new record only. The source is not subtracted — the
  `TSourceLogRecord` may be an in-memory main-log record (whose heap is
  tracked in `mainLogTracker` and will leak upward by a bounded sealed-
  source amount, parity with `ObjectStore`'s `ClearSource=false` branch
  and the legacy observer path), a read-cache record (heap lives in
  `readCacheTracker`, unrelated to this write), or a pending-IO
  `DiskLogRecord` (heap never counted in any tracker). Subtracting
  unconditionally would undercount in the last two cases and drive the
  counter negative.
- `RMWMethods.InPlaceUpdater`: `GetValueHeapMemorySize()` pre/post delta
  on `Succeeded`, so value-heap changes (e.g. APPEND, SETRANGE, or any
  path that triggers `ReallocateValueOverflow` /
  `ConvertInlineToOverflow` / `ConvertOverflowToInline`) are tracked.
- `UpsertMethods.PostInitialWriter` (all three overloads):
  `+logRecord.CalculateHeapMemorySize()` for SET-style inserts that
  create a new record through the Upsert path.
- `DeleteMethods.InPlaceDeleter`:
  `-logRecord.CalculateHeapMemorySize()` before the wrapper sets
  Tombstone. After tombstone is set, `CalculateHeapMemorySize` short-
  circuits to zero and `EvictRecordsInRange` skips the record, so
  without this decrement the creation-side increments would leak for
  every DEL of an overflow record.

The `HyperLogLogPFADD_LTM` suite is the regression that first surfaced
this — HLL sparse→dense goes through CopyUpdater and allocates ≈12 KB of
overflow for the dense representation. With the fix the counter stays
balanced and the assertion `heapSize.Total >= 0` holds throughout.

### Parity

Behavior at every record state encountered during page eviction is
bit-for-bit identical to the prior `SubscribeEvictions` path:

| Record state                                     | Old path (iterator)                | New path (per-record)        | Delta |
| ------------------------------------------------ | ---------------------------------- | ---------------------------- | ----- |
| Valid, !Sealed, !Tombstone, !IsNull              | yielded → -CalculateHeapMemorySize | OnEvict → -Calculate...      | same  |
| `IsNull`                                         | skipped by iterator                | skipped by filter            | 0     |
| `SkipOnScan` (Invalid or Sealed)                 | skipped by iterator                | skipped by filter            | 0     |
| `Tombstone` (post-delete, already ValueIsInline) | yielded → 0 via `!Info.Tombstone`  | skipped by filter → 0        | 0     |

### Hot-path cost

Fast paths (SET / GET / INCR on inline-sized values) pay only a few
aggressive-inlined inline-bit checks; `GetValueHeapMemorySize` on an
inline record early-returns 0 before any tracker call, and a
`heap != 0` guard skips the `AddHeapSize` dispatch entirely. Tracker
work occurs only when a record genuinely created, resized, or freed
an overflow/object allocation — which was already a heavyweight event.

### Testing

All on net10.0 Debug:

- `HyperLogLogTests` (incl. `HyperLogLogPFADD_LTM{32,4096}`,
  `HyperLogLogTestPFMERGE_LTM_*`) ✓
- `CacheSizeTrackerTests` ✓
- `RespListTests` (incl. `ListPushPopStressTest` ×10),
  `RespHashTests`, `RespSetTests`, `RespSortedSetTests`,
  `RespEtagTests`, `RespBitmapTests`, `RespTests.Set*`, `RespTests.Del*`:
  517/517 ✓
- Clean build on both `Garnet.slnx` and `Tsavorite.slnx` (0 warnings,
  0 errors), `dotnet format --verify-no-changes` clean.

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

Author: badrishc

libs/server/Storage/Functions/GarnetRecordTriggers.cs

@@ -7,12 +7,13 @@ namespace Garnet.server
 {
     /// <summary>
     /// Record lifecycle triggers for Garnet's unified store. Handles per-record cleanup
-    /// on delete via <see cref="IRecordTriggers.OnDispose"/>.
+    /// on delete via <see cref="IRecordTriggers.OnDispose"/> and per-record heap-size
+    /// accounting on page eviction via <see cref="IRecordTriggers.OnEvict"/>.
     /// </summary>
     public readonly struct GarnetRecordTriggers : IRecordTriggers
     {
         /// <summary>
-        /// Cache size tracker for heap size accounting on delete.
+        /// Cache size tracker for heap size accounting on delete and eviction.
         /// Created before the store and initialized after via <see cref="CacheSizeTracker.Initialize"/>.
         /// </summary>
         internal readonly CacheSizeTracker cacheSizeTracker;
@@ -29,7 +30,19 @@ public GarnetRecordTriggers(CacheSizeTracker cacheSizeTracker)
         public bool CallOnFlush => false;
 
         /// <inheritdoc/>
-        public bool CallOnEvict => false;
+        // Drives per-record heap-size decrement on page eviction. Mirrors the work the
+        // legacy SubscribeEvictions → LogSizeTracker.OnNext observer path used to perform
+        // (see CacheSizeTracker.Initialize for the wiring change). Gated per source so
+        // that enabling only a main-log or only a read-cache memory budget does not
+        // force a per-record eviction walk on the other allocator.
+        public bool CallOnEvict(EvictionSource source)
+        {
+            if (cacheSizeTracker is null)
+                return false;
+            return source == EvictionSource.ReadCache
+                ? cacheSizeTracker.readCacheTracker is not null
+                : cacheSizeTracker.mainLogTracker is not null;
+        }
 
         /// <inheritdoc/>
         public bool CallOnDiskRead => false;
@@ -49,5 +62,26 @@ public void OnDispose(ref LogRecord logRecord, DisposeReason reason)
                 cacheSizeTracker?.AddHeapSize(-logRecord.ValueObject.HeapMemorySize);
             }
         }
+
+        /// <inheritdoc/>
+        public void OnEvict(ref LogRecord logRecord, EvictionSource source)
+        {
+            if (cacheSizeTracker is null)
+                return;
+
+            // Decrement heap size by this record's heap contribution. Uses the same sizing
+            // helper that LogSizeTracker.OnNext used to sum over an evicted iterator. Routes
+            // through the standard AddHeapSize/AddReadCacheHeapSize path so the assertion
+            // guarding against negative totals remains in force; creation sites on the main
+            // log (RMW PostInitialUpdater/PostCopyUpdater and in-place grow/shrink) must emit
+            // a matching positive bump so the account stays balanced.
+            var size = MemoryUtils.CalculateHeapMemorySize(in logRecord);
+            if (size == 0)
+                return;
+            if (source == EvictionSource.ReadCache)
+                cacheSizeTracker.AddReadCacheHeapSize(-size);
+            else
+                cacheSizeTracker.AddHeapSize(-size);
+        }
     }
 }

libs/server/Storage/Functions/MainStore/DeleteMethods.cs

@@ -28,6 +28,14 @@ public void PostInitialDeleter(ref LogRecord logRecord, ref DeleteInfo deleteInf
         /// <inheritdoc />
         public bool InPlaceDeleter(ref LogRecord logRecord, ref DeleteInfo deleteInfo)
         {
+            // Decrement here — before the wrapper sets Tombstone. After Tombstone is set,
+            // CalculateHeapMemorySize short-circuits to zero and EvictRecordsInRange skips
+            // the record, so any matching decrement would be lost and the creation-side
+            // increments in PostInitialUpdater/PostInitialWriter/InPlaceUpdater would leak.
+            var heap = logRecord.CalculateHeapMemorySize();
+            if (heap != 0)
+                functionsState.cacheSizeTracker?.AddHeapSize(-heap);
+
             logRecord.ClearOptionals();
             if (!logRecord.Info.Modified)
                 functionsState.watchVersionMap.IncrementVersion(deleteInfo.KeyHash);

libs/server/Storage/Functions/MainStore/RMWMethods.cs

@@ -398,6 +398,12 @@ public readonly void PostInitialUpdater(ref LogRecord logRecord, in RecordSizeIn
                 input.header.SetExpiredFlag();
                 rmwInfo.UserData |= NeedAofLog; // Mark that we need to write to AOF
             }
+
+            // Account for any heap contribution (key/value overflow) of the newly created record;
+            // balances the decrement emitted by GarnetRecordTriggers.OnEvict on page eviction.
+            var heap = logRecord.CalculateHeapMemorySize();
+            if (heap != 0)
+                functionsState.cacheSizeTracker?.AddHeapSize(heap);
         }
 
         /// <inheritdoc />
@@ -409,6 +415,7 @@ public readonly bool InPlaceUpdater(ref LogRecord logRecord, ref StringInput inp
                 return false;
             }
 
+            var preHeap = logRecord.GetValueHeapMemorySize();
             var ipuResult = InPlaceUpdaterWorker(ref logRecord, ref input, ref output, ref rmwInfo);
             switch (ipuResult)
             {
@@ -419,6 +426,9 @@ public readonly bool InPlaceUpdater(ref LogRecord logRecord, ref StringInput inp
                         functionsState.watchVersionMap.IncrementVersion(rmwInfo.KeyHash);
                     if (functionsState.appendOnlyFile != null)
                         rmwInfo.UserData |= NeedAofLog; // Mark that we need to write to AOF
+                    var delta = logRecord.GetValueHeapMemorySize() - preHeap;
+                    if (delta != 0)
+                        functionsState.cacheSizeTracker?.AddHeapSize(delta);
                     return true;
                 case IPUResult.NotUpdated:
                 default:
@@ -1701,6 +1711,18 @@ public readonly bool PostCopyUpdater<TSourceLogRecord>(in TSourceLogRecord srcLo
             functionsState.watchVersionMap.IncrementVersion(rmwInfo.KeyHash);
             if (functionsState.appendOnlyFile != null)
                 rmwInfo.UserData |= NeedAofLog; // Mark that we need to write to AOF
+
+            // Account for the new record's heap contribution so that the matching OnEvict decrement
+            // balances. We do NOT subtract srcLogRecord's heap: the source may be a main-log record
+            // (whose heap is tracked in mainLogTracker and will leak upward only by a bounded sealed-
+            // source amount — parity with ObjectStore's ClearSource=false branch and the legacy
+            // SubscribeEvictions path), a read-cache record (heap lives in readCacheTracker, unrelated
+            // to this write), or a pending-IO DiskLogRecord (heap never counted in any tracker).
+            // Subtracting unconditionally would undercount in the last two cases and could drive the
+            // counter negative.
+            var newHeap = dstLogRecord.CalculateHeapMemorySize();
+            if (newHeap != 0)
+                functionsState.cacheSizeTracker?.AddHeapSize(newHeap);
             return true;
         }
 

libs/server/Storage/Functions/MainStore/UpsertMethods.cs

@@ -45,6 +45,12 @@ public void PostInitialWriter(ref LogRecord logRecord, in RecordSizeInfo sizeInf
             functionsState.watchVersionMap.IncrementVersion(upsertInfo.KeyHash);
             if (functionsState.appendOnlyFile != null)
                 upsertInfo.UserData |= NeedAofLog; // Mark that we need to write to AOF
+
+            // Account for overflow key/value heap on the freshly-inserted record so the matching
+            // OnEvict decrement balances.
+            var heap = logRecord.CalculateHeapMemorySize();
+            if (heap != 0)
+                functionsState.cacheSizeTracker?.AddHeapSize(heap);
         }
 
         /// <inheritdoc />
@@ -61,6 +67,10 @@ public void PostInitialWriter<TSourceLogRecord>(ref LogRecord logRecord, in Reco
                 Debug.Assert(!inputLogRecord.Info.ValueIsObject, "String store should not be called with IHeapObject");
                 upsertInfo.UserData |= NeedAofLog; // Mark that we need to write to AOF
             }
+
+            var heap = logRecord.CalculateHeapMemorySize();
+            if (heap != 0)
+                functionsState.cacheSizeTracker?.AddHeapSize(heap);
         }
 
         /// <inheritdoc />

libs/server/Storage/SizeTracker/CacheSizeTracker.cs

@@ -65,8 +65,9 @@ public CacheSizeTracker(TsavoriteKV<StoreFunctions, StoreAllocator> store, long
             => Initialize(store, targetSize, readCacheTargetSize, loggerFactory);
 
         /// <summary>
-        /// Initialize the tracker with a store. Subscribes to eviction notifications.
-        /// Called after store creation when using the parameterless constructor.
+        /// Initialize the tracker with a store. Wires the <see cref="LogSizeTracker"/> as the fast-path
+        /// size tracker for copy-to-tail / copy-to-readcache. Per-record heap-size decrement on page
+        /// eviction is driven by <see cref="GarnetRecordTriggers.OnEvict"/>.
         /// </summary>
         public void Initialize(TsavoriteKV<StoreFunctions, StoreAllocator> store, long targetSize, long readCacheTargetSize, ILoggerFactory loggerFactory = null)
         {
@@ -77,14 +78,14 @@ public void Initialize(TsavoriteKV<StoreFunctions, StoreAllocator> store, long t
             {
                 mainLogTracker = new LogSizeTracker<StoreFunctions, StoreAllocator>(store.Log, targetSize,
                         targetSize / HighTargetSizeDeltaFraction, targetSize / LowTargetSizeDeltaFraction, loggerFactory?.CreateLogger("MainLogSizeTracker"));
-                store.Log.SubscribeEvictions(mainLogTracker);
+                store.Log.SetLogSizeTracker(mainLogTracker);
             }
 
             if (store.ReadCache != null && readCacheTargetSize > 0)
             {
                 readCacheTracker = new LogSizeTracker<StoreFunctions, StoreAllocator>(store.ReadCache, readCacheTargetSize,
                         readCacheTargetSize / HighTargetSizeDeltaFraction, readCacheTargetSize / LowTargetSizeDeltaFraction, loggerFactory?.CreateLogger("ReadCacheSizeTracker"));
-                store.ReadCache.SubscribeEvictions(readCacheTracker);
+                store.ReadCache.SetLogSizeTracker(readCacheTracker);
             }
         }
 

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

@@ -205,6 +205,12 @@ protected string BaseToString(string fuaDetails = "")
         /// <summary>Observer for records getting evicted from memory (page closed). May be the same object as <see cref="logSizeTracker"/>.</summary>
         internal IObserver<ITsavoriteScanIterator> onEvictionObserver;
 
+        /// <summary>
+        /// Whether this allocator is the read cache (as opposed to the main hybrid log).
+        /// Set once at construction from <see cref="AllocatorSettings.IsReadCache"/>.
+        /// </summary>
+        internal readonly bool IsReadCache;
+
         /// <summary>Log size tracker; called when an operation at the Tsavorite-internal level adds or removes heap memory size 
         /// (e.g. copying to log tail or read cache, which do not call <see cref="ISessionFunctions{TInputOutput, TContext}"/>).
         /// May be the same object as <see cref="onEvictionObserver"/>.</summary>
@@ -553,9 +559,15 @@ internal void WriteInlinePageAsync<TContext>(IntPtr alignedSourceAddress, ulong
 
         /// <summary>Instantiate base allocator implementation</summary>
         [MethodImpl(MethodImplOptions.NoInlining)]
-        private protected AllocatorBase(LogSettings logSettings, TStoreFunctions storeFunctions, Func<object, TAllocator> wrapperCreator, Action<long, long> evictCallback,
-                LightEpoch epoch, Action<CommitInfo> flushCallback, ILogger logger = null, ObjectIdMap transientObjectIdMap = null)
+        private protected AllocatorBase(AllocatorSettings allocatorSettings, TStoreFunctions storeFunctions, Func<object, TAllocator> wrapperCreator,
+                ILogger logger = null, ObjectIdMap transientObjectIdMap = null)
         {
+            var logSettings = allocatorSettings.LogSettings;
+            var evictCallback = allocatorSettings.evictCallback;
+            var epoch = allocatorSettings.epoch;
+            var flushCallback = allocatorSettings.flushCallback;
+            IsReadCache = allocatorSettings.IsReadCache;
+
             this.storeFunctions = storeFunctions;
             _wrapper = wrapperCreator(this);
 
@@ -1403,10 +1415,20 @@ public void ShiftBeginAddress(long newBeginAddress, bool truncateLog, bool noFlu
         /// <summary>Invokes eviction observer if set and then frees the page.</summary>
         internal void EvictPageForRecovery(long page)
         {
-            if (logSizeTracker is not null)
+            var start = GetLogicalAddressOfStartOfPage(page);
+            var end = GetLogicalAddressOfStartOfPage(page + 1);
+
+            var source = IsReadCache ? EvictionSource.ReadCache : EvictionSource.MainLog;
+            if (storeFunctions.CallOnEvict(source))
+            {
+                // New per-record path: OnEvict handles heap-size decrement per record. Parity
+                // with the runtime eviction path in OnPagesClosedWorker.
+                _wrapper.EvictRecordsInRange(start, end, source);
+            }
+            else if (logSizeTracker is not null)
             {
-                var start = GetLogicalAddressOfStartOfPage(page);
-                var end = GetLogicalAddressOfStartOfPage(page + 1);
+                // Legacy observer path: materialize an iterator and push heap-size decrement to
+                // the LogSizeTracker.OnNext observer (kept for consumers still using SubscribeEvictions).
                 MemoryPageScan(start, end, logSizeTracker);
             }
 
@@ -1489,8 +1511,9 @@ private void OnPagesClosedWorker()
                         MemoryPageScan(start, end, onEvictionObserver);
 
                     // Notify application of records being evicted — allows cleanup of external resources.
-                    if (storeFunctions.CallOnEvict)
-                        _wrapper.EvictRecordsInRange(start, end);
+                    var evictSource = IsReadCache ? EvictionSource.ReadCache : EvictionSource.MainLog;
+                    if (storeFunctions.CallOnEvict(evictSource))
+                        _wrapper.EvictRecordsInRange(start, end, evictSource);
 
                     // If we are using a null storage device, we must also shift BeginAddress (leave it in-memory)
                     if (IsNullDevice)

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

@@ -23,6 +23,12 @@ public struct AllocatorSettings
         /// <summary>The action to call on page eviction; used only for readcache</summary>
         internal Action<long, long> evictCallback;
 
+        /// <summary>
+        /// Whether this allocator is the read cache (as opposed to the main hybrid log).
+        /// Used to tag per-record eviction callbacks so applications can distinguish the source.
+        /// </summary>
+        internal bool IsReadCache;
+
         /// <summary>The action to execute on flush completion; used only for <see cref="TsavoriteLog"/></summary>
         internal Action<CommitInfo> flushCallback;
 

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

@@ -117,6 +117,9 @@ RecordSizeInfo GetDeleteRecordSize<TKey>(TKey key)
         /// <see cref="IRecordTriggers.OnEvict"/> hook for each valid, non-tombstoned record.
         /// Used during page eviction to allow cleanup of external resources.
         /// </summary>
-        void EvictRecordsInRange(long startAddress, long endAddress);
+        /// <param name="startAddress">Start logical address of the range.</param>
+        /// <param name="endAddress">End logical address of the range (exclusive).</param>
+        /// <param name="source">Identifies whether this eviction is from the main log or the read cache.</param>
+        void EvictRecordsInRange(long startAddress, long endAddress, EvictionSource source);
     }
 }

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

@@ -146,6 +146,6 @@ public readonly RecordSizeInfo GetDeleteRecordSize<TKey>(TKey key)
         public readonly void OnDispose(ref DiskLogRecord logRecord, DisposeReason disposeReason) => _this.OnDispose(ref logRecord, disposeReason);
 
         /// <inheritdoc/>
-        public readonly void EvictRecordsInRange(long startAddress, long endAddress) => _this.EvictRecordsInRange(startAddress, endAddress);
+        public readonly void EvictRecordsInRange(long startAddress, long endAddress, EvictionSource source) => _this.EvictRecordsInRange(startAddress, endAddress, source);
     }
 }