A voter that hangs holds the leader's lease and writes for MaxReplicationLag × RequestTimeout after a burst of writes (empty AppendEntries never answers ReplicatedWithLeaderTerm) #309

// Three voters on DotNext's TCP transport, the leader lease on. C sits behind a TCP proxy that can make it hang
// (connections accepted, nothing forwarded either way, as a black-holed peer). C hangs during 3 s of writes every 100 ms;
// the writes stop; one write is made 1 s later. The leader's lease is sampled every 10 ms throughout.
using System.Collections.Concurrent;
using System.Diagnostics;
using System.Net;
using System.Net.Sockets;
using DotNext.IO;
using DotNext.Net.Cluster.Consensus.Raft;
using DotNext.Net.Cluster.Consensus.Raft.StateMachine;

var requestTimeout = TimeSpan.FromMilliseconds(500);
await using var nodes = new Nodes();
var leader = await nodes.Start("A", coldStart: true, requestTimeout);
var healthy = await nodes.Start("B", coldStart: false, requestTimeout);
var silent = await nodes.Start("C", coldStart: false, requestTimeout, proxied: true);
using var timeout = new CancellationTokenSource(TimeSpan.FromMinutes(5));
await leader.Cluster.WaitForLeaderAsync(TimeSpan.FromSeconds(30), timeout.Token);
if (!await leader.Cluster.AddMemberAsync(healthy.PublicEndPoint, timeout.Token) || !await leader.Cluster.AddMemberAsync(silent.PublicEndPoint, timeout.Token))
    throw new InvalidOperationException("B or C not added");

while (!Leased(leader.Cluster))
    await Task.Delay(20);

Console.WriteLine($"DotNext.Net.Cluster {typeof(RaftCluster).Assembly.GetName().Version}, TCP transport, 3 voters, IsLeaderLeaseEnabled, " +
    $"election timeout 1000-2000 ms, RequestTimeout {requestTimeout.TotalMilliseconds} ms, MaxReplicationLag 16 (default)");

var clock = Stopwatch.StartNew();
List<(double From, double To)> lapses = [];
using var stopSampling = new CancellationTokenSource();
var sampler = Task.Run(async () =>
{
    double? since = null;
    while (!stopSampling.IsCancellationRequested)
    {
        var now = clock.Elapsed.TotalMilliseconds;
        if (Leased(leader.Cluster))
        {
            if (since is { } began)
            {
                lapses.Add((began, now));
                since = null;
            }
        }
        else
        {
            since ??= now;
        }

        await Task.Delay(10);
    }

    if (since is { } open)
        lapses.Add((open, clock.Elapsed.TotalMilliseconds));
});

var payload = new byte[64];
async Task<int> Writes(TimeSpan duration)
{
    var until = clock.Elapsed + duration;
    var committed = 0;
    while (clock.Elapsed < until)
    {
        await leader.Cluster.ReplicateAsync(new Entry(leader.Cluster.Term, payload), timeout.Token);
        committed++;
        await Task.Delay(100);
    }

    return committed;
}

Console.WriteLine($"{clock.Elapsed.TotalSeconds,5:F1} s  all three answer: writes every 100 ms for 3 s, {await Writes(TimeSpan.FromSeconds(3))} committed");
silent.Proxy!.Hang();
Console.WriteLine($"{clock.Elapsed.TotalSeconds,5:F1} s  C hangs: writes every 100 ms for 3 s, {await Writes(TimeSpan.FromSeconds(3))} committed");
var quiet = clock.Elapsed;
Console.WriteLine($"{quiet.TotalSeconds,5:F1} s  writes stop");
await Task.Delay(TimeSpan.FromSeconds(1));
var write = Stopwatch.StartNew();
Console.WriteLine($"{clock.Elapsed.TotalSeconds,5:F1} s  one write");
await leader.Cluster.ReplicateAsync(new Entry(leader.Cluster.Term, payload), timeout.Token);
Console.WriteLine($"{clock.Elapsed.TotalSeconds,5:F1} s  that write committed, after {write.ElapsedMilliseconds} ms");
await Task.Delay(quiet + TimeSpan.FromSeconds(12) - clock.Elapsed);
stopSampling.Cancel();
await sampler;
Console.WriteLine($"{clock.Elapsed.TotalSeconds,5:F1} s  end; A still leader: {!leader.Cluster.LeadershipToken.IsCancellationRequested}");
Console.WriteLine(lapses.Count == 0 ? "A held its lease throughout" : "A without its lease:");
foreach (var (from, to) in lapses)
    Console.WriteLine($"  {from / 1000,5:F2} s to {to / 1000,5:F2} s: {to - from,6:F0} ms");

static bool Leased(RaftCluster cluster) => cluster.TryGetLeaseToken(out var lease) && !lease.IsCancellationRequested;

sealed record Node(RaftCluster Cluster, WriteAheadLog Log, IPEndPoint PublicEndPoint, Proxy? Proxy);

sealed class Nodes : IAsyncDisposable
{
    private readonly List<Node> nodes = [];

    public async Task<Node> Start(string name, bool coldStart, TimeSpan requestTimeout, bool proxied = false)
    {
        var host = FreeEndPoint();
        var proxy = proxied ? new Proxy(host) : null;
        var publicEndPoint = proxy?.EndPoint ?? host;
        var log = new WriteAheadLog(new() { Location = Directory.CreateTempSubdirectory($"repro-{name}-").FullName }, IStateMachine.CreateNoOp());
        var cluster = new RaftCluster(new RaftCluster.TcpConfiguration(host)
        {
            PublicEndPoint = publicEndPoint,
            LowerElectionTimeout = 1000,
            UpperElectionTimeout = 2000,
            RequestTimeout = requestTimeout,
            IsLeaderLeaseEnabled = true,
            ColdStart = coldStart,
            ConfigurationStorage = null,
        }) { AuditTrail = log };
        await cluster.StartAsync(CancellationToken.None);
        var node = new Node(cluster, log, publicEndPoint, proxy);
        nodes.Add(node);
        return node;
    }

    private static IPEndPoint FreeEndPoint()
    {
        var listener = new TcpListener(IPAddress.Loopback, 0);
        listener.Start();
        var port = ((IPEndPoint)listener.LocalEndpoint).Port;
        listener.Stop();
        return new IPEndPoint(IPAddress.Loopback, port);
    }

    public async ValueTask DisposeAsync()
    {
        foreach (var node in nodes)
        {
            node.Proxy?.Dispose();
            using var stop = new CancellationTokenSource(TimeSpan.FromSeconds(3));
            try
            {
                await node.Cluster.StopAsync(stop.Token);
            }
            catch (Exception e) when (e is OperationCanceledException or InvalidOperationException)
            {
            }

            node.Cluster.Dispose();
            await node.Log.DisposeAsync();
        }
    }
}

/// <summary>A TCP proxy in front of a node: passes until told to hang (connections accepted, nothing forwarded either way).</summary>
sealed class Proxy : IDisposable
{
    private readonly TcpListener listener;
    private readonly IPEndPoint target;
    private readonly ConcurrentDictionary<Socket, byte> open = new();
    private volatile bool hanging;

    public Proxy(IPEndPoint target)
    {
        this.target = target;
        listener = new TcpListener(IPAddress.Loopback, 0);
        listener.Start();
        EndPoint = (IPEndPoint)listener.LocalEndpoint;
        _ = Accept();
    }

    public IPEndPoint EndPoint { get; }

    public void Hang() => hanging = true;

    private async Task Accept()
    {
        while (true)
        {
            Socket client;
            try
            {
                client = await listener.AcceptSocketAsync();
            }
            catch (Exception)
            {
                return;
            }

            open[client] = 0;
            if (hanging)
            {
                _ = Pump(client, null);
                continue;
            }

            var server = new Socket(SocketType.Stream, ProtocolType.Tcp);
            try
            {
                await server.ConnectAsync(target);
            }
            catch (Exception)
            {
                Close(client);
                continue;
            }

            open[server] = 0;
            _ = Pump(client, server);
            _ = Pump(server, client);
        }
    }

    // forwards while passing; drops what comes while hanging
    private async Task Pump(Socket from, Socket? to)
    {
        var buffer = new byte[64 * 1024];
        try
        {
            int read;
            while ((read = await from.ReceiveAsync(buffer)) > 0)
            {
                if (!hanging && to is not null)
                    await to.SendAsync(buffer.AsMemory(0, read));
            }
        }
        catch (Exception)
        {
        }

        Close(from);
        if (to is not null)
            Close(to);
    }

    private void Close(Socket socket)
    {
        open.TryRemove(socket, out _);
        socket.Close();
    }

    public void Dispose()
    {
        listener.Stop();
        foreach (var socket in open.Keys)
            Close(socket);
    }
}

readonly record struct Entry(long Term, byte[] Payload) : IRaftLogEntry
{
    bool IDataTransferObject.IsReusable => true;

    long? IDataTransferObject.Length => Payload.Length;

    bool IDataTransferObject.TryGetMemory(out ReadOnlyMemory<byte> memory)
    {
        memory = Payload;
        return true;
    }

    ValueTask IDataTransferObject.WriteToAsync<TWriter>(TWriter writer, CancellationToken token) => writer.Invoke(Payload, token);
}