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 lets us collapse
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`).
- 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.
- `AllocatorBase.EvictPageForRecovery` now also routes through the per-record
  `EvictRecordsInRange` when `storeFunctions.CallOnEvict` 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`).

### Garnet changes

- `GarnetRecordTriggers.CallOnEvict` is now gated on `cacheSizeTracker != null`.
  `OnEvict(ref LogRecord, EvictionSource)` computes
  `MemoryUtils.CalculateHeapMemorySize(in logRecord)` and dispatches to
  `AddHeapSize(-size)` or `AddReadCacheHeapSize(-size)` based on the source.
- `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.

### 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     |

Tombstones converge via two independent mechanisms (the old relies on the
`if (!Info.Tombstone)` guard inside `CalculateHeapMemorySize`; the new
short-circuits in the filter), so both yield 0 contribution as required
given the delete-site `OnDispose(Deleted)` already decremented the tracker.

### Testing

All tests pass on net10.0 Debug:

- `CacheSizeTrackerTests` 2/2
- `RespListTests.ListPushPopStressTest` (10 repetitions) ✓
- `RespListTests`, `RespHashTests`, `RespSetTests`, `RespSortedSetTests`:
  467/467
- Tsavorite `DeleteDisposeTests` + `ReadCacheTests`: 228/228, 14 skipped
- Clean build on both `Garnet.slnx` and `Tsavorite.slnx` (0 warnings,
  0 errors).

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,10 @@ 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).
+        public bool CallOnEvict => cacheSizeTracker != null;
 
         /// <inheritdoc/>
         public bool CallOnDiskRead => false;
@@ -49,5 +53,24 @@ 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, so the
+            // total decrement for any evicted page range is identical to the pre-change path.
+            var size = MemoryUtils.CalculateHeapMemorySize(in logRecord);
+            if (size == 0)
+                return;
+
+            if (source == EvictionSource.ReadCache)
+                cacheSizeTracker.AddReadCacheHeapSize(-size);
+            else
+                cacheSizeTracker.AddHeapSize(-size);
+        }
     }
 }

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,19 @@ 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);
+
+            if (storeFunctions.CallOnEvict)
+            {
+                // New per-record path: OnEvict handles heap-size decrement per record. Parity
+                // with the runtime eviction path in OnPagesClosedWorker.
+                _wrapper.EvictRecordsInRange(start, end, IsReadCache ? EvictionSource.ReadCache : EvictionSource.MainLog);
+            }
+            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);
             }
 
@@ -1490,7 +1511,7 @@ private void OnPagesClosedWorker()
 
                     // Notify application of records being evicted — allows cleanup of external resources.
                     if (storeFunctions.CallOnEvict)
-                        _wrapper.EvictRecordsInRange(start, end);
+                        _wrapper.EvictRecordsInRange(start, end, IsReadCache ? EvictionSource.ReadCache : EvictionSource.MainLog);
 
                     // 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);
     }
 }

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

@@ -74,7 +74,7 @@ internal struct ObjectPage
         public override string ToString() => BaseToString($" (LI {LastIssuedFlushedUntilAddress}, OG {OngoingFlushedUntilAddress}, No {NoFlushUntilAddress})");
 
         public ObjectAllocatorImpl(AllocatorSettings settings, TStoreFunctions storeFunctions, Func<object, ObjectAllocator<TStoreFunctions>> wrapperCreator)
-            : base(settings.LogSettings, storeFunctions, wrapperCreator, settings.evictCallback, settings.epoch, settings.flushCallback, settings.logger, transientObjectIdMap: new ObjectIdMap())
+            : base(settings, storeFunctions, wrapperCreator, settings.logger, transientObjectIdMap: new ObjectIdMap())
         {
             objectLogDevice = settings.LogSettings.ObjectLogDevice;
 
@@ -347,7 +347,7 @@ internal void OnDispose(ref DiskLogRecord logRecord, DisposeReason disposeReason
         /// Iterate records in the given logical address range and call <see cref="IStoreFunctions.OnEvict"/>
         /// on each valid, non-tombstoned record. Used during page eviction to allow cleanup of external resources.
         /// </summary>
-        internal void EvictRecordsInRange(long startAddress, long endAddress)
+        internal void EvictRecordsInRange(long startAddress, long endAddress, EvictionSource source)
         {
             // Ensure we start after the page header
             var page = GetPage(startAddress);
@@ -378,7 +378,7 @@ internal void EvictRecordsInRange(long startAddress, long endAddress)
                 }
 
                 // Notify the application that this record is being evicted from memory.
-                storeFunctions.OnEvict(ref logRecord);
+                storeFunctions.OnEvict(ref logRecord, source);
 
                 address += allocatedSize;
             }

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

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

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

@@ -15,7 +15,7 @@ internal sealed unsafe class SpanByteAllocatorImpl<TStoreFunctions> : AllocatorB
         private OverflowPool<PageUnit<Empty>> freePagePool;
 
         public SpanByteAllocatorImpl(AllocatorSettings settings, TStoreFunctions storeFunctions, Func<object, SpanByteAllocator<TStoreFunctions>> wrapperCreator)
-            : base(settings.LogSettings, storeFunctions, wrapperCreator, settings.evictCallback, settings.epoch, settings.flushCallback, settings.logger)
+            : base(settings, storeFunctions, wrapperCreator, settings.logger)
         {
             freePagePool = new OverflowPool<PageUnit<Empty>>(4, p => { });
             pageHeaderSize = PageHeader.Size;

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

@@ -144,6 +144,6 @@ public readonly RecordSizeInfo GetDeleteRecordSize<TKey>(TKey key)
         public void OnDispose(ref DiskLogRecord logRecord, DisposeReason disposeReason) => throw new NotImplementedException("Not implemented for TsavoriteLogAllocator");
 
         /// <inheritdoc/>
-        public void EvictRecordsInRange(long startAddress, long endAddress) { }
+        public void EvictRecordsInRange(long startAddress, long endAddress, EvictionSource source) { }
     }
 }