test: make the concurrency tests independent of the runner's load

tests/BLite.Tests/CdcScalabilityTests.cs

@@ -37,7 +37,9 @@ public async Task Test_Cdc_1000_Subscribers_Receive_Events()
         await _db.SaveChangesAsync();
 
         // 3. Wait for events to propagate
-        await Task.Delay(1000);
+        await LoadTolerant.WaitUntilAsync(
+            () => Enumerable.Range(0, SubscriberCount).All(i => Volatile.Read(ref eventCounts[i]) >= 2),
+            "every subscriber has received both events");
 
         // 4. Verify all subscribers received both events
         for (int i = 0; i < SubscriberCount; i++)
@@ -54,41 +56,54 @@ public async Task Test_Cdc_Slow_Subscriber_Does_Not_Block_Others()
         var fastEventCount = 0;
         var slowEventCount = 0;
 
-        // 1. Register a slow subscriber that blocks SYNCHRONOUSLY
-        using var slowSub = _db.People.Watch().Subscribe(_ => 
+        // 1. Register a slow subscriber that blocks SYNCHRONOUSLY, for as long as the test
+        //    keeps the gate closed, so that the ordering below does not depend on timing.
+        var slowGate = new ManualResetEventSlim(false);
+        using var slowSub = _db.People.Watch().Subscribe(_ =>
         {
             Interlocked.Increment(ref slowEventCount);
             // Synchronous block to block the BridgeChannelToObserverAsync loop for this sub
-            Thread.Sleep(2000); 
+            slowGate.Wait();
         });
 
         // 2. Register a fast subscriber
-        using var fastSub = _db.People.Watch().Subscribe(_ => 
+        using var fastSub = _db.People.Watch().Subscribe(_ =>
         {
             Interlocked.Increment(ref fastEventCount);
         });
 
-        // 3. Perform a write
-        await _db.People.InsertAsync(new Person { Id = 1, Name = "John", Age = 30 });
-        await _db.SaveChangesAsync();
-
-        // 4. Verification: Fast subscriber should receive it immediately
-        await Task.Delay(200);
-        Assert.Equal(1, fastEventCount);
-        Assert.Equal(1, slowEventCount); // Started but not finished or blocking others
-
-        // 5. Perform another write
-        await _db.People.InsertAsync(new Person { Id = 2, Name = "Jane", Age = 25 });
-        await _db.SaveChangesAsync();
-
-        // 6. Verification: Fast subscriber should receive second event while slow one is still busy
-        await Task.Delay(200);
-        Assert.Equal(2, fastEventCount);
-        Assert.Equal(1, slowEventCount); // Still processing first one or second one queued in private channel
-
-        // 7. Wait for slow one to eventually catch up
-        await Task.Delay(2500); // Wait for the second one in slow sub to be processed after the first Sleep
-        Assert.Equal(2, slowEventCount);
+        try
+        {
+            // 3. Perform a write
+            await _db.People.InsertAsync(new Person { Id = 1, Name = "John", Age = 30 });
+            await _db.SaveChangesAsync();
+
+            // 4. Verification: both subscribers receive it; the slow one then blocks on the gate
+            await LoadTolerant.WaitUntilAsync(
+                () => Volatile.Read(ref fastEventCount) == 1 && Volatile.Read(ref slowEventCount) == 1,
+                "both subscribers have received the first event");
+
+            // 5. Perform another write
+            await _db.People.InsertAsync(new Person { Id = 2, Name = "Jane", Age = 25 });
+            await _db.SaveChangesAsync();
+
+            // 6. Verification: the fast subscriber receives the second event while the slow one
+            //    is still blocked in the first; the second one waits in its private channel
+            await LoadTolerant.WaitUntilAsync(
+                () => Volatile.Read(ref fastEventCount) == 2,
+                "the fast subscriber has received the second event");
+            Assert.Equal(1, Volatile.Read(ref slowEventCount));
+
+            // 7. Release the slow one: it catches up
+            slowGate.Set();
+            await LoadTolerant.WaitUntilAsync(
+                () => Volatile.Read(ref slowEventCount) == 2,
+                "the slow subscriber has caught up");
+        }
+        finally
+        {
+            slowGate.Set(); // never leave the bridge loop blocked
+        }
     }
 
     public void Dispose()

tests/BLite.Tests/ConcurrentConsistencyTests.cs

@@ -32,7 +32,7 @@ public class ConcurrentConsistencyTests : IDisposable
     public ConcurrentConsistencyTests()
     {
         _dbPath = Path.Combine(Path.GetTempPath(), $"blite_consistency_{Guid.NewGuid()}.db");
-        _engine = new BLiteEngine(_dbPath, PageFileConfig.Server(_dbPath));
+        _engine = new BLiteEngine(_dbPath, LoadTolerant.Server(_dbPath));
     }
 
     public void Dispose()
@@ -145,11 +145,13 @@ public async Task ConcurrentInsertAndRead_8Threads_NoCorruption()
             await session.CommitAsync();
         })).ToArray();
 
-        // Readers: continuously read seeded documents while writers are active
+        // Readers: continuously read seeded documents while writers are active. Each reader
+        // reads at least once: on a loaded runner the writers can finish before a reader task
+        // is even scheduled.
         var readerTasks = Enumerable.Range(0, readerCount).Select(r => Task.Run(async () =>
         {
             int reads = 0;
-            while (!cts.Token.IsCancellationRequested)
+            do
             {
                 using var session = _engine.OpenSession();
                 var col = session.GetOrCreateCollection(readCollection);
@@ -166,8 +168,7 @@ public async Task ConcurrentInsertAndRead_8Threads_NoCorruption()
                     break;
                 }
                 reads++;
-                if (reads >= 500) break;
-            }
+            } while (reads < 500 && !cts.Token.IsCancellationRequested);
             readSuccesses.Add(reads);
         })).ToArray();
 

tests/BLite.Tests/CrossCollectionWriteRaceTests.cs

@@ -34,7 +34,9 @@ public void Dispose()
         [Fact]
         public async Task Concurrent_Inserts_Into_Different_Collections_LoseNothing()
         {
-            using var db = new TestDbContext(_path);
+            // Three writers share the file-wide write lock, and each holds it through its commit.
+            // The default 500 ms is not enough on a loaded runner (see LoadTolerant).
+            using var db = new TestDbContext(_path, LoadTolerant.SingleFile);
             using var cts = new CancellationTokenSource(Duration);
 
             var failures = new System.Collections.Concurrent.ConcurrentBag<Exception>();

tests/BLite.Tests/IndexPageReleaseTests.cs

@@ -194,7 +194,7 @@ public async Task DynamicCollection_VacuumRebuildIndexes_UnreadableDocument_Fail
     [Fact]
     public async Task CreateDropIndexCycles_Typed_DoNotGrowTheFile()
     {
-        using var db = new TestDbContext(_dbPath);
+        using var db = new TestDbContext(_dbPath, LoadTolerant.SingleFile);
         await db.PeopleV2.InsertBulkAsync(Enumerable.Range(0, 1000).Select(i => new PersonV2 { Name = $"p{i}", Age = i % 90 }));
 
         async Task CycleAsync()
@@ -209,7 +209,7 @@ async Task CycleAsync()
     [Fact]
     public async Task CreateDropIndexCycles_Dynamic_DoNotGrowTheFile()
     {
-        using var engine = new BLiteEngine(_dbPath);
+        using var engine = new BLiteEngine(_dbPath, LoadTolerant.SingleFile);
         var col = engine.GetOrCreateCollection("docs");
         await InsertIndexableDocsAsync(col, 1000);
 
@@ -355,7 +355,9 @@ public async Task Dispose_WaitsForThePendingFreeCheckpoint_LongerThanTheLockTime
     [Fact]
     public async Task DynamicCollection_DropIndex_WaitsForTheCollectionLock()
     {
-        using var engine = new BLiteEngine(_dbPath);
+        // The drop must still be waiting when this test looks at it after Task.Delay(100): on a
+        // loaded runner that continuation can run after the default 500 ms lock timeout.
+        using var engine = new BLiteEngine(_dbPath, LoadTolerant.SingleFile);
         var col = engine.GetOrCreateCollection("docs");
         await InsertIndexableDocsAsync(col, 100);
         await col.CreateIndexAsync("value", "idx_value");

tests/BLite.Tests/LoadTolerant.cs

@@ -0,0 +1,42 @@
+using BLite.Core.Storage;
+
+namespace BLite.Tests;
+
+/// <summary>
+/// Settings for tests whose subject is correctness under concurrency, not lock timeouts.
+/// <para>
+/// <see cref="LockTimeout.Default"/> gives a writer 500 ms to take the file-wide write lock or the
+/// commit lock. On a loaded CI runner (xunit runs test classes in parallel on a few cores, and
+/// other classes block threads) a writer regularly waits longer than that for a commit that is
+/// itself waiting for the disk, and the test fails with a <see cref="TimeoutException"/> that
+/// says nothing about what it checks. These tests open their database with a timeout no loaded
+/// runner reaches, so a timeout can only mean a real deadlock.
+/// </para>
+/// </summary>
+internal static class LoadTolerant
+{
+    /// <summary>Lock timeouts a loaded runner does not reach; only a deadlock does.</summary>
+    public static readonly LockTimeout LockTimeout = LockTimeout.From(readTimeoutMs: 30_000, writeTimeoutMs: 30_000);
+
+    /// <summary><see cref="PageFileConfig.Default"/> with <see cref="LockTimeout"/>.</summary>
+    public static PageFileConfig SingleFile => PageFileConfig.Default with { LockTimeout = LockTimeout };
+
+    /// <summary><see cref="PageFileConfig.Server(string, PageFileConfig?)"/> with <see cref="LockTimeout"/>.</summary>
+    public static PageFileConfig Server(string databasePath) => PageFileConfig.Server(databasePath, SingleFile);
+
+    /// <summary>
+    /// Polls <paramref name="condition"/> until it holds, instead of sleeping a fixed time and
+    /// hoping the background work is done by then. Fails with <paramref name="what"/> after
+    /// <paramref name="timeoutMs"/>.
+    /// </summary>
+    public static async Task WaitUntilAsync(Func<bool> condition, string what, int timeoutMs = 10_000)
+    {
+        var deadline = Environment.TickCount64 + timeoutMs;
+        while (!condition())
+        {
+            if (Environment.TickCount64 >= deadline)
+                Assert.Fail($"Timed out after {timeoutMs} ms waiting until {what}");
+            await Task.Delay(5);
+        }
+    }
+}

tests/BLite.Tests/WalLockTimeoutTests.cs

@@ -15,6 +15,19 @@ public class WalLockTimeoutTests : IDisposable
     /// </summary>
     private const int TestWriteTimeoutMs = 100;
 
+    /// <summary>
+    /// How much later than <see cref="TestWriteTimeoutMs"/> a timed-out wait may return. The timer
+    /// fires late on a loaded runner (770 ms seen in CI for the 100 ms timeout); the bound only
+    /// has to tell the configured timeout from the 5 s production default.
+    /// </summary>
+    private const int SchedulingSlackMs = 2000;
+
+    /// <summary>
+    /// Write timeout for the cancellation test: long enough that a cancellation token due in
+    /// 10 ms wins even when the runner delays both timers.
+    /// </summary>
+    private const int LongWriteTimeoutMs = 10_000;
+
     public WalLockTimeoutTests(ITestOutputHelper output)
     {
         _output = output;
@@ -50,7 +63,7 @@ public async Task WriteBeginRecord_ThrowsTimeoutException_WhenLockIsHeld()
             _output.WriteLine($"TimeoutException after {sw.ElapsedMilliseconds} ms: {ex.Message}");
             Assert.True(sw.ElapsedMilliseconds >= TestWriteTimeoutMs - 20,
                 $"Should wait ~{TestWriteTimeoutMs} ms, was {sw.ElapsedMilliseconds} ms");
-            Assert.True(sw.ElapsedMilliseconds < TestWriteTimeoutMs * 5,
+            Assert.True(sw.ElapsedMilliseconds < TestWriteTimeoutMs + SchedulingSlackMs,
                 $"Should not wait much longer than {TestWriteTimeoutMs} ms, was {sw.ElapsedMilliseconds} ms");
         }
         finally
@@ -164,7 +177,7 @@ public void ReadAll_ThrowsTimeoutException_WhenLockIsHeld()
     [Fact]
     public async Task CancellationToken_AbortsBeforeTimeout()
     {
-        using var wal = new WriteAheadLog(_walPath, writeTimeoutMs: TestWriteTimeoutMs);
+        using var wal = new WriteAheadLog(_walPath, writeTimeoutMs: LongWriteTimeoutMs);
         var semaphore = GetInternalLock(wal);
 
         await semaphore.WaitAsync();
@@ -179,7 +192,7 @@ await Assert.ThrowsAnyAsync<OperationCanceledException>(
             sw.Stop();
 
             _output.WriteLine($"Cancelled after {sw.ElapsedMilliseconds} ms");
-            Assert.True(sw.ElapsedMilliseconds < TestWriteTimeoutMs,
+            Assert.True(sw.ElapsedMilliseconds < LongWriteTimeoutMs,
                 $"Cancellation should fire before the timeout, was {sw.ElapsedMilliseconds} ms");
         }
         finally