Fixing Voron race condition that can happen when a read transaction i…

…s started _while_ the flush is in progress.

This can cause the read transaction to read data that started after it opened, so we need to prevent that.

Raven.Voron/Voron.Tests/Bugs/InvalidReleasesOfScratchPages.cs

@@ -160,7 +160,9 @@ public void AllScratchPagesShouldBeReleased()
 
                 env.FlushLogToDataFile(); // non read nor write transactions, so it should flush and release everything from scratch
 
-                Assert.Equal(0, env.ScratchBufferPool.GetNumberOfAllocations(0));
+                // we keep track of the pages in scratch for one additional transaction, to avoid race
+                // condition with FlushLogToDataFile concurrently with new read transactions
+                Assert.Equal(2, env.ScratchBufferPool.GetNumberOfAllocations(0));
             }
         }
     }

Raven.Voron/Voron.Tests/ScratchBuffer/MutipleScratchBuffersUsage.cs

@@ -37,7 +37,7 @@ public void CanAddContinuallyGrowingValue()
             // also in the meanwhile the free space handling is doing its job so it needs some pages too
             // and we allocate not the exact size but the nearest power of two e.g. we write 257 pages but in scratch we request 512 ones
             int i = 0;
-            while (size < 256 * AbstractPager.PageSize) 
+            while (size < 128 * AbstractPager.PageSize) 
             {
                 using (Env.NewTransaction(TransactionFlags.Read))
                 {

Raven.Voron/Voron/Impl/Compaction/StorageCompaction.cs

@@ -42,7 +42,7 @@ public static void Execute(StorageEnvironmentOptions srcOptions, StorageEnvironm
 
                 compactedEnv.FlushLogToDataFile(allowToFlushOverwrittenPages: true);
 
-                compactedEnv.Journal.Applicator.SyncDataFile(compactedEnv.OldestTransaction);
+                compactedEnv.Journal.Applicator.SyncDataFile();
                 compactedEnv.Journal.Applicator.DeleteCurrentAlreadyFlushedJournal();
 
                 minimalCompactedDataFileSize = compactedEnv.NextPageNumber*AbstractPager.PageSize;

Raven.Voron/Voron/Impl/Journal/WriteAheadJournal.cs

@@ -419,7 +419,7 @@ private IDisposable ForceFlushingPagesOlderThan(long oldestActiveTransaction)
                 });
             }
 
-            public void ApplyLogsToDataFile(long oldestActiveTransaction, CancellationToken token, Transaction transaction = null, bool allowToFlushOverwrittenPages = false)
+            public void ApplyLogsToDataFile(CancellationToken token, Transaction transaction = null, bool allowToFlushOverwrittenPages = false)
             {
                 if (token.IsCancellationRequested)
                     return;
@@ -451,6 +451,8 @@ public void ApplyLogsToDataFile(long oldestActiveTransaction, CancellationToken
 
                     long lastFlushedTransactionId = -1;
 
+                    long oldestActiveTransaction = _waj._env.OldestTransaction;
+
                     foreach (var journalFile in jrnls.Where(x => x.Number >= _lastSyncedJournal))
                     {
                         var currentJournalMaxTransactionId = -1L;
@@ -529,7 +531,7 @@ public void ApplyLogsToDataFile(long oldestActiveTransaction, CancellationToken
 
                             using (ForceFlushingPagesOlderThan(oldestActiveTransaction))
                             {
-                                ApplyLogsToDataFile(oldestActiveTransaction, token, transaction, false);
+                                ApplyLogsToDataFile(token, transaction, false);
                             }
                         }
 
@@ -578,7 +580,7 @@ public void ApplyLogsToDataFile(long oldestActiveTransaction, CancellationToken
                     if (_totalWrittenButUnsyncedBytes > DelayedDataFileSynchronizationBytesLimit ||
                         DateTime.UtcNow - _lastDataFileSyncTime > _delayedDataFileSynchronizationTimeLimit)
                     {
-                        SyncDataFile(oldestActiveTransaction);
+                        SyncDataFile();
                     }
 
                 }
@@ -589,11 +591,11 @@ public void ApplyLogsToDataFile(long oldestActiveTransaction, CancellationToken
                 }
             }
 
-            internal void SyncDataFile(long oldestActiveTransaction)
+            internal void SyncDataFile()
             {
                 _waj._dataPager.Sync();
 
-                UpdateFileHeaderAfterDataFileSync(_lastFlushedJournal, oldestActiveTransaction);
+                UpdateFileHeaderAfterDataFileSync(_lastFlushedJournal, _lastSyncedTransactionId);
 
                 foreach (var toDelete in _journalsToDelete.Values)
                 {
@@ -684,17 +686,22 @@ private void EnsureDataPagerSpacing(Transaction transaction, Page last, int numb
 
             private void FreeScratchPages(IEnumerable<JournalFile> unusedJournalFiles, Transaction txw)
             {
+                // we release up to the last read transaction, because there might be new read transactions that are currently
+                // running, that started after the flush
+                var lastSyncedTransactionId = Math.Min(_lastSyncedTransactionId, _waj._env.CurrentReadTransactionId - 1);
+
                 // we have to free pages of the unused journals before the remaining ones that are still in use
                 // to prevent reading from them by any read transaction (read transactions search journals from the newest
                 // to read the most updated version)
+
                 foreach (var journalFile in unusedJournalFiles.OrderBy(x => x.Number))
                 {
-                    journalFile.FreeScratchPagesOlderThan(txw, _lastSyncedTransactionId, forceToFreeAllPages: forcedFlushOfOldPages);
+                    journalFile.FreeScratchPagesOlderThan(txw, lastSyncedTransactionId, forceToFreeAllPages: forcedFlushOfOldPages);
                 }
 
                 foreach (var jrnl in _waj._files.OrderBy(x => x.Number))
                 {
-                    jrnl.FreeScratchPagesOlderThan(txw, _lastSyncedTransactionId, forceToFreeAllPages: forcedFlushOfOldPages);
+                    jrnl.FreeScratchPagesOlderThan(txw, lastSyncedTransactionId, forceToFreeAllPages: forcedFlushOfOldPages);
                 }
             }
 
@@ -716,7 +723,7 @@ private List<JournalFile> GetUnusedJournalFiles(IEnumerable<JournalSnapshot> jrn
                 return unusedJournalFiles;
             }
 
-            public void UpdateFileHeaderAfterDataFileSync(JournalFile file, long oldestActiveTransaction)
+            public void UpdateFileHeaderAfterDataFileSync(JournalFile file, long maxTransactionId)
             {
                 var txHeaders = stackalloc TransactionHeader[2];
                 var readTxHeader = &txHeaders[0];
@@ -729,7 +736,7 @@ public void UpdateFileHeaderAfterDataFileSync(JournalFile file, long oldestActiv
                         break;
                     if (readTxHeader->HeaderMarker != Constants.TransactionHeaderMarker)
                         break;
-                    if (readTxHeader->TransactionId + 1 == oldestActiveTransaction)
+                    if (readTxHeader->TransactionId > maxTransactionId)
                         break;
 
                     lastReadTxHeader = *readTxHeader;

Raven.Voron/Voron/StorageEnvironment.cs

@@ -432,13 +432,14 @@ public Transaction NewTransaction(TransactionFlags flags, TimeSpan? timeout = nu
                         RecordTransactionState(tx, DebugActionType.TransactionStart);
                         tx.RecordTransactionState = RecordTransactionState;
                     }
+
+                    _activeTransactions.Add(tx);
                 }
                 finally
                 {
                     _txCommit.ExitReadLock();
                 }
-
-                _activeTransactions.Add(tx);
+
                 tx.EnsurePagerStateReference(_dataPager.PagerState);
 
                 if (flags == TransactionFlags.ReadWrite)
@@ -461,6 +462,32 @@ private void RecordTransactionState(Transaction tx, DebugActionType state)
             DebugJournal.RecordTransactionAction(tx, state);
         }
 
+        public long CurrentReadTransactionId
+        {
+            get { return Thread.VolatileRead(ref _transactionsCounter); }
+        }
+
+        internal ExitWriteLock PreventNewReadTransactions()
+        {
+            _txCommit.EnterWriteLock();
+            return new ExitWriteLock(_txCommit);
+        }
+
+        public struct ExitWriteLock : IDisposable
+        {
+            readonly ReaderWriterLockSlim _rwls;
+
+            public ExitWriteLock(ReaderWriterLockSlim rwls)
+            {
+                _rwls = rwls;
+            }
+
+            public void Dispose()
+            {
+                _rwls.ExitWriteLock();
+            }
+        }
+
         public long NextWriteTransactionId
         {
             get { return Thread.VolatileRead(ref _transactionsCounter) + 1; }
@@ -470,19 +497,14 @@ private void TransactionAfterCommit(Transaction tx)
         {
             if (_activeTransactions.Contains(tx) == false)
                 return;
-
-            _txCommit.EnterWriteLock();
-            try
+
+            using (PreventNewReadTransactions())
             {
                 if (tx.Committed && tx.FlushedToJournal)
                     _transactionsCounter = tx.Id;
 
                 State = tx.State;
             }
-            finally
-            {
-                _txCommit.ExitWriteLock();
-            }
 
             if (tx.FlushedToJournal == false)
                 return;
@@ -623,7 +645,7 @@ private Task FlushWritesToDataFileAsync()
 
                             try
                             {
-                                _journal.Applicator.ApplyLogsToDataFile(OldestTransaction, _cancellationTokenSource.Token);
+                                _journal.Applicator.ApplyLogsToDataFile(_cancellationTokenSource.Token);
                             }
                             catch (TimeoutException)
                             {
@@ -657,7 +679,7 @@ internal void ForceLogFlushToDataFile(Transaction tx, bool allowToFlushOverwritt
                 _debugJournal.RecordFlushAction(DebugActionType.FlushStart, tx);
             }
 
-            _journal.Applicator.ApplyLogsToDataFile(OldestTransaction, _cancellationTokenSource.Token, tx, allowToFlushOverwrittenPages);
+            _journal.Applicator.ApplyLogsToDataFile(_cancellationTokenSource.Token, tx, allowToFlushOverwrittenPages);
 
             if (IsDebugRecording)
             {