diff --git a/tests/BLite.Tests/CdcScalabilityTests.cs b/tests/BLite.Tests/CdcScalabilityTests.cs index cd3406bc..3a0c39e2 100644 --- a/tests/BLite.Tests/CdcScalabilityTests.cs +++ b/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() diff --git a/tests/BLite.Tests/ConcurrentConsistencyTests.cs b/tests/BLite.Tests/ConcurrentConsistencyTests.cs index e13cdaf9..b4dcff22 100644 --- a/tests/BLite.Tests/ConcurrentConsistencyTests.cs +++ b/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(); diff --git a/tests/BLite.Tests/CrossCollectionWriteRaceTests.cs b/tests/BLite.Tests/CrossCollectionWriteRaceTests.cs index 8a715e99..3bcc80a2 100644 --- a/tests/BLite.Tests/CrossCollectionWriteRaceTests.cs +++ b/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(); diff --git a/tests/BLite.Tests/IndexPageReleaseTests.cs b/tests/BLite.Tests/IndexPageReleaseTests.cs index df8d2353..62d69af5 100644 --- a/tests/BLite.Tests/IndexPageReleaseTests.cs +++ b/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"); diff --git a/tests/BLite.Tests/LoadTolerant.cs b/tests/BLite.Tests/LoadTolerant.cs new file mode 100644 index 00000000..414a3dd1 --- /dev/null +++ b/tests/BLite.Tests/LoadTolerant.cs @@ -0,0 +1,42 @@ +using BLite.Core.Storage; + +namespace BLite.Tests; + +/// +/// Settings for tests whose subject is correctness under concurrency, not lock timeouts. +/// +/// 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 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. +/// +/// +internal static class LoadTolerant +{ + /// Lock timeouts a loaded runner does not reach; only a deadlock does. + public static readonly LockTimeout LockTimeout = LockTimeout.From(readTimeoutMs: 30_000, writeTimeoutMs: 30_000); + + /// with . + public static PageFileConfig SingleFile => PageFileConfig.Default with { LockTimeout = LockTimeout }; + + /// with . + public static PageFileConfig Server(string databasePath) => PageFileConfig.Server(databasePath, SingleFile); + + /// + /// Polls until it holds, instead of sleeping a fixed time and + /// hoping the background work is done by then. Fails with after + /// . + /// + public static async Task WaitUntilAsync(Func 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); + } + } +} diff --git a/tests/BLite.Tests/WalLockTimeoutTests.cs b/tests/BLite.Tests/WalLockTimeoutTests.cs index e3d566cc..5e7fe33f 100644 --- a/tests/BLite.Tests/WalLockTimeoutTests.cs +++ b/tests/BLite.Tests/WalLockTimeoutTests.cs @@ -15,6 +15,19 @@ public class WalLockTimeoutTests : IDisposable /// private const int TestWriteTimeoutMs = 100; + /// + /// How much later than 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. + /// + private const int SchedulingSlackMs = 2000; + + /// + /// Write timeout for the cancellation test: long enough that a cancellation token due in + /// 10 ms wins even when the runner delays both timers. + /// + 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( 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