scadaproj/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/SnapshotDriftTests.cs

using System.Threading.Channels;
using Microsoft.Extensions.Logging.Abstractions;
using ZB.MOM.WW.LocalDb.Contracts;
using ZB.MOM.WW.LocalDb.Internal;
using ZB.MOM.WW.LocalDb.Replication;
using ZB.MOM.WW.LocalDb.Replication.Internal;

namespace ZB.MOM.WW.LocalDb.Tests;

/// <summary>
/// LocalDb-004 (scadaproj#39): <see cref="ReplicationOptions.FailClosedOnDrift"/> must hold on the
/// snapshot path too. It was checked only for deltas, so a future-stamped snapshot row was applied,
/// pushed the receiver's HLC a year ahead and persisted that in <c>last_seen_hlc</c>.
/// </summary>
public sealed class SnapshotDriftTests : IDisposable
{
    private static readonly TimeSpan RunTimeout = TimeSpan.FromSeconds(15);

    private readonly List<string> _paths = [];
    private readonly List<IDisposable> _disposables = [];

    public void Dispose()
    {
        foreach (var d in _disposables)
            d.Dispose();
        Microsoft.Data.Sqlite.SqliteConnection.ClearAllPools();
        foreach (var p in _paths)
        {
            if (File.Exists(p)) File.Delete(p);
            if (File.Exists(p + "-wal")) File.Delete(p + "-wal");
            if (File.Exists(p + "-shm")) File.Delete(p + "-shm");
        }
    }

    private sealed record Side(SqliteLocalDb Db, OplogStore Store, SyncSession Session);

    private async Task<SqliteLocalDb> NewDbAsync()
    {
        var path = Path.Combine(Path.GetTempPath(), Guid.NewGuid() + ".db");
        _paths.Add(path);
        var db = new SqliteLocalDb(new LocalDbOptions { Path = path });
        _disposables.Add(db);
        await db.ExecuteAsync("CREATE TABLE orders (id INTEGER PRIMARY KEY, sku TEXT, qty INTEGER)");
        db.RegisterReplicated("orders");
        return db;
    }

    // Built exactly as production does (SyncSessionFactory), so the test covers the real wiring of
    // the snapshot applier — the place the options were missing.
    private static Side SessionOver(SqliteLocalDb db, ReplicationOptions options) =>
        new(db, new OplogStore(db, options), SyncSessionFactory.Create(db, options, NullLoggerFactory.Instance));

    private static (SyncDuplex A, SyncDuplex B) DuplexPair()
    {
        var aToB = Channel.CreateUnbounded<SyncMessage>();
        var bToA = Channel.CreateUnbounded<SyncMessage>();
        return (
            new SyncDuplex { Send = (m, ct) => aToB.Writer.WriteAsync(m, ct).AsTask(), Inbox = bToA.Reader },
            new SyncDuplex { Send = (m, ct) => bToA.Writer.WriteAsync(m, ct).AsTask(), Inbox = aToB.Reader });
    }

    private static long HlcAt(DateTimeOffset utc) => utc.ToUnixTimeMilliseconds() << 16;

    private static async Task<List<long>> Ids(SqliteLocalDb db) =>
        (await db.QueryAsync("SELECT id FROM orders ORDER BY id", r => r.GetInt64(0))).ToList();

    private static async Task<long> Scalar(SqliteLocalDb db, string sql) =>
        (await db.QueryAsync(sql, r => r.GetInt64(0)))[0];

    private static async Task WaitForAsync(Func<Task<bool>> predicate, TimeSpan timeout)
    {
        var deadline = DateTime.UtcNow + timeout;
        while (DateTime.UtcNow < deadline)
        {
            if (await predicate()) return;
            await Task.Delay(20);
        }
        throw new TimeoutException("Condition not reached within " + timeout);
    }

    private static async Task SwallowAsync(Task task)
    {
        try { await task; }
        catch (OperationCanceledException) { }
    }

    /// <summary>
    /// A wiped B is healed by a snapshot from A, one of whose rows A stamped while its clock ran a
    /// year ahead. With FailClosedOnDrift that row must be dead-lettered on B and must not move B's
    /// clock; the honest rows still arrive.
    /// </summary>
    [Theory]
    [InlineData(true)]
    [InlineData(false)]
    public async Task FutureStampedSnapshotRow_HonoursFailClosedOnDrift(bool failClosed)
    {
        var options = new ReplicationOptions { FlushInterval = TimeSpan.FromMilliseconds(20), FailClosedOnDrift = failClosed };
        var dbA = await NewDbAsync();
        await dbA.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'ok', 1)");
        await dbA.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (2, 'future', 2)");
        var future = HlcAt(DateTimeOffset.UtcNow.AddDays(365));
        await dbA.ExecuteAsync(
            "UPDATE __localdb_row_version SET hlc = @hlc WHERE table_name = 'orders' AND pk_json = '{\"id\":2}'",
            new { hlc = future });
        var a = SessionOver(dbA, options);
        await a.Store.RecordPeerAckAsync(await a.Store.GetMaxSeqAsync());

        var dbB = await NewDbAsync();
        var b = SessionOver(dbB, options);

        var (da, dbx) = DuplexPair();
        using var cts = new CancellationTokenSource(RunTimeout);
        var runA = a.Session.RunAsync(da, cts.Token);
        var runB = b.Session.RunAsync(dbx, cts.Token);
        await WaitForAsync(async () => await Scalar(dbB, "SELECT last_applied_remote_seq FROM __localdb_peer_state") > 0, RunTimeout);
        cts.Cancel();
        await SwallowAsync(runA);
        await SwallowAsync(runB);

        var lastSeen = await Scalar(dbB, "SELECT last_seen_hlc FROM __localdb_peer_state");
        if (failClosed)
        {
            Assert.Equal(new[] { 1L }, await Ids(dbB));
            Assert.Equal(1, await Scalar(dbB, "SELECT COUNT(*) FROM __localdb_dead_letter WHERE error LIKE '%drift%'"));
            Assert.True(dbB.Clock.Current < future, "the drifted row advanced the receiver's clock");
            Assert.True(lastSeen < future, "the drifted row's HLC was persisted as last_seen_hlc");
        }
        else
        {
            // Warn-only mode applies it, exactly like the delta path.
            Assert.Equal(new[] { 1L, 2L }, await Ids(dbB));
            Assert.Equal(0, await Scalar(dbB, "SELECT COUNT(*) FROM __localdb_dead_letter"));
        }
    }
}