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

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

namespace ZB.MOM.WW.LocalDb.Tests;

/// <summary>
/// The passive service's single-stream guard under a HALF-OPEN link: the dialling node is gone, but
/// nothing on this side knows it, so the handler sits inside <c>MoveNext</c> forever holding the
/// guard. Through 0.3.1 that made the pair unrecoverable without a restart — every reconnect from
/// the rebooted peer was refused <c>AlreadyExists</c>, indefinitely on an idle pair, while this node
/// went on reporting itself connected. The guard is there to serialize sessions, not to fence out a
/// caller the auth interceptor already vouched for, so the newest dialler wins.
/// </summary>
public sealed class SyncStreamGuardTests : IDisposable
{
    private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(10);

    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 LocalDbSyncService NewService()
    {
        var path = Path.Combine(Path.GetTempPath(), Guid.NewGuid() + ".db");
        _paths.Add(path);
        var db = new Internal.SqliteLocalDb(new LocalDbOptions { Path = path });
        _disposables.Add(db);
        db.ExecuteAsync("CREATE TABLE orders (id INTEGER PRIMARY KEY, sku TEXT)").GetAwaiter().GetResult();
        db.RegisterReplicated("orders");

        return new LocalDbSyncService(
            db,
            Options.Create(new ReplicationOptions { FlushInterval = TimeSpan.FromMilliseconds(20) }),
            NullLoggerFactory.Instance);
    }

    private static ServerCallContext ContextFor(CancellationToken ct) => new FakeCallContext(ct);

    [Fact]
    public async Task WedgedIncumbent_IsDisplacedByANewStream()
    {
        var service = NewService();
        // Eviction is normally generous; the test only needs it to not be the thing under test.
        service.TakeoverTimeout = Timeout;

        // The incumbent's peer crashed: its stream will never deliver another message, and neither
        // end times out on silence — a converged pair is legitimately quiet. The handler parks in
        // the handshake's first read, holding the guard.
        using var incumbentCts = new CancellationTokenSource();
        var incumbentOut = new RecordingWriter();
        var incumbent = service.Sync(new SilentReader(), incumbentOut, ContextFor(incumbentCts.Token));

        // It really is running and really did claim the guard (its handshake is on the wire).
        await incumbentOut.NextAsync(Timeout);
        Assert.False(incumbent.IsCompleted);

        // The rebooted peer dials again. Pre-fix this threw AlreadyExists and the pair stayed wedged.
        using var newcomerCts = new CancellationTokenSource();
        var newcomerOut = new RecordingWriter();
        var newcomer = service.Sync(new SilentReader(), newcomerOut, ContextFor(newcomerCts.Token));

        // The incumbent is torn down — gracefully, not as a fault of this node.
        await incumbent.WaitAsync(Timeout);
        // ...and the new stream is genuinely served: its own handshake goes out.
        var handshake = await newcomerOut.NextAsync(Timeout);
        Assert.Equal(SyncMessage.MsgOneofCase.Handshake, handshake.MsgCase);
        Assert.False(newcomer.IsCompleted);

        newcomerCts.Cancel();
        await SwallowAsync(newcomer);
    }

    [Fact]
    public async Task DisplacedIncumbent_ReleasesTheGuard_SoAThirdStreamIsAlsoServed()
    {
        // Takeover must leave the guard in a state that can be taken over AGAIN — a handover that
        // leaked the slot would simply move the wedge one reconnect down the line.
        var service = NewService();
        service.TakeoverTimeout = Timeout;

        using var firstCts = new CancellationTokenSource();
        var firstOut = new RecordingWriter();
        var first = service.Sync(new SilentReader(), firstOut, ContextFor(firstCts.Token));
        await firstOut.NextAsync(Timeout);

        using var secondCts = new CancellationTokenSource();
        var secondOut = new RecordingWriter();
        var second = service.Sync(new SilentReader(), secondOut, ContextFor(secondCts.Token));
        await first.WaitAsync(Timeout);
        await secondOut.NextAsync(Timeout);

        using var thirdCts = new CancellationTokenSource();
        var thirdOut = new RecordingWriter();
        var third = service.Sync(new SilentReader(), thirdOut, ContextFor(thirdCts.Token));
        await second.WaitAsync(Timeout);
        Assert.Equal(SyncMessage.MsgOneofCase.Handshake, (await thirdOut.NextAsync(Timeout)).MsgCase);

        thirdCts.Cancel();
        await SwallowAsync(third);
    }

    private static async Task SwallowAsync(Task task)
    {
        try { await task.WaitAsync(Timeout); }
        catch (OperationCanceledException) { }
        catch (RpcException) { }
    }

    /// <summary>
    /// The minimum <see cref="ServerCallContext"/> the handler actually reads: its cancellation
    /// token. (Grpc.Core.Api ships no test double, and the ASP.NET Core one is not reachable outside
    /// a live call.)
    /// </summary>
    private sealed class FakeCallContext(CancellationToken ct) : ServerCallContext
    {
        protected override string MethodCore => "/localdb_sync.v1.LocalDbSync/Sync";
        protected override string HostCore => "test";
        protected override string PeerCore => "test";
        protected override DateTime DeadlineCore => DateTime.MaxValue;
        protected override Metadata RequestHeadersCore => [];
        protected override CancellationToken CancellationTokenCore => ct;
        protected override Metadata ResponseTrailersCore { get; } = [];
        protected override Status StatusCore { get; set; }
        protected override WriteOptions? WriteOptionsCore { get; set; }
        protected override AuthContext AuthContextCore => new(null, new Dictionary<string, List<AuthProperty>>());

        protected override ContextPropagationToken CreatePropagationTokenCore(ContextPropagationOptions? options) =>
            throw new NotSupportedException();

        protected override Task WriteResponseHeadersAsyncCore(Metadata responseHeaders) => Task.CompletedTask;
    }

    /// <summary>A stream that never delivers anything and never ends — a half-open link.</summary>
    private sealed class SilentReader : IAsyncStreamReader<SyncMessage>
    {
        public SyncMessage Current => throw new InvalidOperationException("no message");

        public async Task<bool> MoveNext(CancellationToken cancellationToken)
        {
            await Task.Delay(System.Threading.Timeout.Infinite, cancellationToken);
            return false;
        }
    }

    private sealed class RecordingWriter : IServerStreamWriter<SyncMessage>
    {
        private readonly Channel<SyncMessage> _written = Channel.CreateUnbounded<SyncMessage>();

        public WriteOptions? WriteOptions { get; set; }

        public Task WriteAsync(SyncMessage message) => WriteAsync(message, CancellationToken.None);

        public Task WriteAsync(SyncMessage message, CancellationToken cancellationToken)
        {
            _written.Writer.TryWrite(message);
            return Task.CompletedTask;
        }

        public async Task<SyncMessage> NextAsync(TimeSpan timeout)
        {
            using var cts = new CancellationTokenSource(timeout);
            return await _written.Reader.ReadAsync(cts.Token);
        }
    }
}