Implement checkpoint modes with docs/tests and reorganize project file layout

tests/CBDD.Tests/Cdc/CdcScalabilityTests.cs

@@ -0,0 +1,121 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+using ZB.MOM.WW.CBDD.Core.CDC;
+using ZB.MOM.WW.CBDD.Shared;
+using Xunit;
+
+namespace ZB.MOM.WW.CBDD.Tests;
+
+public class CdcScalabilityTests : IDisposable
+{
+    private readonly Shared.TestDbContext _db;
+    private readonly string _dbPath;
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="CdcScalabilityTests"/> class.
+    /// </summary>
+    public CdcScalabilityTests()
+    {
+        _dbPath = Path.Combine(Path.GetTempPath(), $"cdc_scaling_{Guid.NewGuid()}.db");
+        _db = new Shared.TestDbContext(_dbPath);
+    }
+
+    /// <summary>
+    /// Verifies CDC dispatch reaches all registered subscribers.
+    /// </summary>
+    [Fact]
+    public async Task Test_Cdc_1000_Subscribers_Receive_Events()
+    {
+        var ct = TestContext.Current.CancellationToken;
+        const int SubscriberCount = 1000;
+        var eventCounts = new int[SubscriberCount];
+        var subscriptions = new List<IDisposable>();
+
+        // 1. Create 1000 subscribers
+        for (int i = 0; i < SubscriberCount; i++)
+        {
+            int index = i;
+            var sub = _db.People.Watch().Subscribe(_ =>
+            {
+                Interlocked.Increment(ref eventCounts[index]);
+            });
+            subscriptions.Add(sub);
+        }
+
+        // 2. Perform some writes
+        _db.People.Insert(new Person { Id = 1, Name = "John", Age = 30 });
+        _db.People.Insert(new Person { Id = 2, Name = "Jane", Age = 25 });
+        _db.SaveChanges();
+
+        // 3. Wait for events to propagate
+        await Task.Delay(1000, ct);
+
+        // 4. Verify all subscribers received both events
+        for (int i = 0; i < SubscriberCount; i++)
+        {
+            eventCounts[i].ShouldBe(2);
+        }
+
+        foreach (var sub in subscriptions) sub.Dispose();
+    }
+
+    /// <summary>
+    /// Verifies a slow subscriber does not block other subscribers.
+    /// </summary>
+    [Fact(Skip = "Performance test - run manually when needed")]
+    public async Task Test_Cdc_Slow_Subscriber_Does_Not_Block_Others()
+    {
+        var ct = TestContext.Current.CancellationToken;
+        var fastEventCount = 0;
+        var slowEventCount = 0;
+
+        // 1. Register a slow subscriber that blocks SYNCHRONOUSLY
+        using var slowSub = _db.People.Watch().Subscribe(_ =>
+        {
+            Interlocked.Increment(ref slowEventCount);
+            // Synchronous block to block the BridgeChannelToObserverAsync loop for this sub
+            Thread.Sleep(2000);
+        });
+
+        // 2. Register a fast subscriber
+        using var fastSub = _db.People.Watch().Subscribe(_ =>
+        {
+            Interlocked.Increment(ref fastEventCount);
+        });
+
+        // 3. Perform a write
+        _db.People.Insert(new Person { Id = 1, Name = "John", Age = 30 });
+        _db.SaveChanges();
+
+        // 4. Verification: Fast subscriber should receive it immediately
+        await Task.Delay(200, ct);
+        fastEventCount.ShouldBe(1);
+        slowEventCount.ShouldBe(1); // Started but not finished or blocking others
+
+        // 5. Perform another write
+        _db.People.Insert(new Person { Id = 2, Name = "Jane", Age = 25 });
+        _db.SaveChanges();
+
+        // 6. Verification: Fast subscriber should receive second event while slow one is still busy
+        await Task.Delay(200, ct);
+        fastEventCount.ShouldBe(2);
+        slowEventCount.ShouldBe(1); // Still processing first one or second one queued in private channel
+
+        // 7. Wait for slow one to eventually catch up
+        await Task.Delay(2500, ct); // Wait for the second one in slow sub to be processed after the first Sleep
+        slowEventCount.ShouldBe(2);
+    }
+
+    /// <summary>
+    /// Disposes test resources and removes temporary files.
+    /// </summary>
+    public void Dispose()
+    {
+        _db.Dispose();
+        if (File.Exists(_dbPath)) File.Delete(_dbPath);
+        var wal = Path.ChangeExtension(_dbPath, ".wal");
+        if (File.Exists(wal)) File.Delete(wal);
+    }
+}

tests/CBDD.Tests/Cdc/CdcTests.cs

@@ -0,0 +1,215 @@
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.IO;
+using System.Linq;
+using System.Threading.Tasks;
+using ZB.MOM.WW.CBDD.Bson;
+using ZB.MOM.WW.CBDD.Core.CDC;
+using ZB.MOM.WW.CBDD.Core.Transactions;
+using ZB.MOM.WW.CBDD.Shared;
+using Xunit;
+
+namespace ZB.MOM.WW.CBDD.Tests;
+
+public class CdcTests : IDisposable
+{
+    private static readonly TimeSpan DefaultEventTimeout = TimeSpan.FromSeconds(3);
+    private static readonly TimeSpan PollInterval = TimeSpan.FromMilliseconds(10);
+
+    private readonly string _dbPath = $"cdc_test_{Guid.NewGuid()}.db";
+    private readonly Shared.TestDbContext _db;
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="CdcTests"/> class.
+    /// </summary>
+    public CdcTests()
+    {
+        _db = new Shared.TestDbContext(_dbPath);
+    }
+
+    /// <summary>
+    /// Verifies that an insert operation publishes a CDC event.
+    /// </summary>
+    [Fact]
+    public async Task Test_Cdc_Basic_Insert_Fires_Event()
+    {
+        var ct = TestContext.Current.CancellationToken;
+        var events = new ConcurrentQueue<ChangeStreamEvent<int, Person>>();
+        using var subscription = _db.People.Watch(capturePayload: true).Subscribe(events.Enqueue);
+
+        var person = new Person { Id = 1, Name = "John", Age = 30 };
+        _db.People.Insert(person);
+        _db.SaveChanges();
+
+        await WaitForEventCountAsync(events, expectedCount: 1, ct);
+
+        var snapshot = events.ToArray();
+        snapshot.Length.ShouldBe(1);
+        snapshot[0].Type.ShouldBe(OperationType.Insert);
+        snapshot[0].DocumentId.ShouldBe(1);
+        snapshot[0].Entity.ShouldNotBeNull();
+        snapshot[0].Entity!.Name.ShouldBe("John");
+    }
+
+    /// <summary>
+    /// Verifies payload is omitted when CDC capture payload is disabled.
+    /// </summary>
+    [Fact]
+    public async Task Test_Cdc_No_Payload_When_Not_Requested()
+    {
+        var ct = TestContext.Current.CancellationToken;
+        var events = new ConcurrentQueue<ChangeStreamEvent<int, Person>>();
+        using var subscription = _db.People.Watch(capturePayload: false).Subscribe(events.Enqueue);
+
+        var person = new Person { Id = 1, Name = "John", Age = 30 };
+        _db.People.Insert(person);
+        _db.SaveChanges();
+
+        await WaitForEventCountAsync(events, expectedCount: 1, ct);
+
+        var snapshot = events.ToArray();
+        snapshot.Length.ShouldBe(1);
+        snapshot[0].Entity.ShouldBeNull();
+    }
+
+    /// <summary>
+    /// Verifies CDC events are published only for committed changes.
+    /// </summary>
+    [Fact]
+    public async Task Test_Cdc_Commit_Only()
+    {
+        var ct = TestContext.Current.CancellationToken;
+        var events = new ConcurrentQueue<ChangeStreamEvent<int, Person>>();
+        using var subscription = _db.People.Watch(capturePayload: true).Subscribe(events.Enqueue);
+
+        using (var txn = _db.BeginTransaction())
+        {
+            _db.People.Insert(new Person { Id = 1, Name = "John" });
+            events.Count.ShouldBe(0); // Not committed yet
+
+            txn.Rollback();
+        }
+
+        await Task.Delay(100, ct);
+        events.Count.ShouldBe(0); // Rolled back
+
+        using (var txn = _db.BeginTransaction())
+        {
+            _db.People.Insert(new Person { Id = 2, Name = "Jane" });
+            txn.Commit();
+        }
+
+        await WaitForEventCountAsync(events, expectedCount: 1, ct);
+        var snapshot = events.ToArray();
+        snapshot.Length.ShouldBe(1);
+        snapshot[0].DocumentId.ShouldBe(2);
+    }
+
+    /// <summary>
+    /// Verifies update and delete operations publish CDC events.
+    /// </summary>
+    [Fact]
+    public async Task Test_Cdc_Update_And_Delete()
+    {
+        var ct = TestContext.Current.CancellationToken;
+        var events = new ConcurrentQueue<ChangeStreamEvent<int, Person>>();
+        using var subscription = _db.People.Watch(capturePayload: true).Subscribe(events.Enqueue);
+
+        var person = new Person { Id = 1, Name = "John", Age = 30 };
+        _db.People.Insert(person);
+        _db.SaveChanges();
+
+        person.Name = "Johnny";
+        _db.People.Update(person);
+        _db.SaveChanges();
+
+        _db.People.Delete(1);
+        _db.SaveChanges();
+
+        await WaitForEventCountAsync(events, expectedCount: 3, ct);
+
+        var snapshot = events.ToArray();
+        snapshot.Length.ShouldBe(3);
+        snapshot[0].Type.ShouldBe(OperationType.Insert);
+        snapshot[1].Type.ShouldBe(OperationType.Update);
+        snapshot[2].Type.ShouldBe(OperationType.Delete);
+
+        snapshot[1].Entity!.Name.ShouldBe("Johnny");
+        snapshot[2].DocumentId.ShouldBe(1);
+    }
+
+    /// <summary>
+    /// Disposes test resources and removes temporary files.
+    /// </summary>
+    public void Dispose()
+    {
+        _db.Dispose();
+        if (File.Exists(_dbPath)) File.Delete(_dbPath);
+        if (File.Exists(_dbPath + "-wal")) File.Delete(_dbPath + "-wal");
+    }
+
+    private static async Task WaitForEventCountAsync(
+        ConcurrentQueue<ChangeStreamEvent<int, Person>> events,
+        int expectedCount,
+        CancellationToken ct)
+    {
+        var sw = Stopwatch.StartNew();
+        while (sw.Elapsed < DefaultEventTimeout)
+        {
+            if (events.Count >= expectedCount)
+            {
+                return;
+            }
+
+            await Task.Delay(PollInterval, ct);
+        }
+
+        events.Count.ShouldBe(expectedCount);
+    }
+}
+
+// Simple helper to avoid System.Reactive dependency in tests
+public static class ObservableExtensions
+{
+    /// <summary>
+    /// Subscribes to an observable sequence using an action callback.
+    /// </summary>
+    /// <typeparam name="T">The event type.</typeparam>
+    /// <param name="observable">The observable sequence.</param>
+    /// <param name="onNext">The callback for next events.</param>
+    /// <returns>An <see cref="IDisposable"/> subscription.</returns>
+    public static IDisposable Subscribe<T>(this IObservable<T> observable, Action<T> onNext)
+    {
+        return observable.Subscribe(new AnonymousObserver<T>(onNext));
+    }
+
+    private class AnonymousObserver<T> : IObserver<T>
+    {
+        private readonly Action<T> _onNext;
+
+        /// <summary>
+        /// Initializes a new instance of the <see cref="AnonymousObserver{T}"/> class.
+        /// </summary>
+        /// <param name="onNext">The callback for next events.</param>
+        public AnonymousObserver(Action<T> onNext) => _onNext = onNext;
+
+        /// <summary>
+        /// Handles completion.
+        /// </summary>
+        public void OnCompleted() { }
+
+        /// <summary>
+        /// Handles an observable error.
+        /// </summary>
+        /// <param name="error">The observed error.</param>
+        public void OnError(Exception error) { }
+
+        /// <summary>
+        /// Handles the next value.
+        /// </summary>
+        /// <param name="value">The observed value.</param>
+        public void OnNext(T value) => _onNext(value);
+    }
+}