RavenDB-22972 Dequeued EnvirnmentStateRecord elements should be expli…

…ctly nulled because they might hold references of objects which have finalizers (ClientState)

src/Voron/EnvironmentStateRecord.cs

@@ -18,3 +18,8 @@ public record EnvironmentStateRecord(
     (JournalFile Current, long Last4KWritePosition) Journal,
     object ClientState,
     List<long> SparsePageRanges);
+
+internal sealed class EnvironmentStateRecordHolder
+{
+    public EnvironmentStateRecord EnvStateRecord;
+}

src/Voron/StorageEnvironment.cs

@@ -118,7 +118,7 @@ public static IDisposable GetStaticContext(out ByteStringContext ctx)
 
         internal TestingStuff _forTestingPurposes;
         private EnvironmentStateRecord _currentStateRecord;
-        private readonly ConcurrentQueue<EnvironmentStateRecord> _transactionsToFlush = new();
+        private readonly ConcurrentQueue<EnvironmentStateRecordHolder> _transactionsToFlush = new();
 
         internal EnvironmentStateRecord CurrentStateRecord => _currentStateRecord;
 
@@ -1340,11 +1340,11 @@ public InMemoryStorageState GetInMemoryStorageState(LowLevelTransaction tx)
                 },
                 TransactionsToFlush = _transactionsToFlush.ToList().Select(x => new InMemoryStorageState.EnvironmentStateRecordDetails
                 {
-                    TransactionId = x.TransactionId,
-                    ScratchPagesTable = GetScratchTableSummary(x.ScratchPagesTable),
-                    NextPageNumber = x.NextPageNumber,
-                    WrittenToJournalNumber = x.WrittenToJournalNumber,
-                    ClientStateType = x.ClientState?.GetType().ToString()
+                    TransactionId = x.EnvStateRecord.TransactionId,
+                    ScratchPagesTable = GetScratchTableSummary(x.EnvStateRecord.ScratchPagesTable),
+                    NextPageNumber = x.EnvStateRecord.NextPageNumber,
+                    WrittenToJournalNumber = x.EnvStateRecord.WrittenToJournalNumber,
+                    ClientStateType = x.EnvStateRecord.ClientState?.GetType().ToString()
                 }).ToList(),
                 FlushState = new InMemoryStorageState.FlushStateDetails
                 {
@@ -1661,12 +1661,12 @@ private void UpdateStateOnCommit(LowLevelTransaction tx)
 
             // we don't _have_ to make it using interlocked, but let's publish it immediately
             Interlocked.Exchange(ref _currentStateRecord, updatedState);
-            
+
             // We only want to flush to data pager transactions that have been flushed to the journal.
             // Transactions that _haven't_ been flushed are mostly book-keeping (updating scratch table, etc)
             if (tx.WrittenToJournalNumber >= 0)
             {
-                _transactionsToFlush.Enqueue(updatedState);
+                _transactionsToFlush.Enqueue(new EnvironmentStateRecordHolder { EnvStateRecord = updatedState });
             }
         }
 
@@ -1687,16 +1687,24 @@ internal ApplyLogsToDataFileState TryGetLatestEnvironmentStateToFlush(long uptoT
             while (true)
             {
                 if (_transactionsToFlush.TryPeek(out var maybe) == false ||
-                    maybe.TransactionId >= uptoTxIdExclusive)
+                    maybe.EnvStateRecord.TransactionId >= uptoTxIdExclusive)
                 {
                     if (found == false)
                         return null;
                     Debug.Assert(record is not null);
                     return new ApplyLogsToDataFileState(scratchBuffers, sparseRegions, record);
                 }
-                if (_transactionsToFlush.TryDequeue(out record) == false)
+
+                if (_transactionsToFlush.TryDequeue(out EnvironmentStateRecordHolder recordHolder) == false)
                     throw new InvalidOperationException("Failed to get transaction to flush after already peeked successfully");
 
+                record = recordHolder.EnvStateRecord;
+
+                // single thread is reading from this, so we can be sure that peek + take gets the same value
+                Debug.Assert(ReferenceEquals(record, maybe.EnvStateRecord));
+
+                recordHolder.EnvStateRecord = null; // this way we ensure that _transactionsToFlush won't hold reference in its internal slots preventing GC on EnvironmentStateRecord.ClientState object
+
                 if (record.SparsePageRanges != null)
                 {
                     sparseRegions.AddRange(record.SparsePageRanges);
@@ -1708,8 +1716,6 @@ internal ApplyLogsToDataFileState TryGetLatestEnvironmentStateToFlush(long uptoT
                     scratchBuffers.Add(pageFromScratch);
                 }
 
-                // single thread is reading from this, so we can be sure that peek + take gets the same value
-                Debug.Assert(ReferenceEquals(record, maybe));
                 found = true;
             }
         }