natsdotnet/tests/NATS.Server.Tests/Raft/RaftApplyQueueTests.cs

using NATS.Server.Raft;

namespace NATS.Server.Tests.Raft;

/// <summary>
/// Tests for CommitQueue and commit/processed index tracking in RaftNode.
/// Go reference: raft.go:150-160 (applied/processed fields), raft.go:2100-2150 (ApplyQ).
/// </summary>
public class RaftApplyQueueTests
{
    // -- Helpers --

    private static (RaftNode[] nodes, InMemoryRaftTransport transport) CreateCluster(int size)
    {
        var transport = new InMemoryRaftTransport();
        var nodes = Enumerable.Range(1, size)
            .Select(i => new RaftNode($"n{i}", transport))
            .ToArray();
        foreach (var node in nodes)
        {
            transport.Register(node);
            node.ConfigureCluster(nodes);
        }
        return (nodes, transport);
    }

    private static RaftNode ElectLeader(RaftNode[] nodes)
    {
        var candidate = nodes[0];
        candidate.StartElection(nodes.Length);
        foreach (var voter in nodes.Skip(1))
            candidate.ReceiveVote(voter.GrantVote(candidate.Term, candidate.Id), nodes.Length);
        return candidate;
    }

    // -- CommitQueue<T> unit tests --

    [Fact]
    public async Task Enqueue_and_dequeue_lifecycle()
    {
        var queue = new CommitQueue<RaftLogEntry>();

        var entry = new RaftLogEntry(1, 1, "cmd-1");
        await queue.EnqueueAsync(entry);
        queue.Count.ShouldBe(1);

        var dequeued = await queue.DequeueAsync();
        dequeued.ShouldBe(entry);
        queue.Count.ShouldBe(0);
    }

    [Fact]
    public async Task Multiple_items_dequeue_in_fifo_order()
    {
        var queue = new CommitQueue<RaftLogEntry>();

        var entry1 = new RaftLogEntry(1, 1, "cmd-1");
        var entry2 = new RaftLogEntry(2, 1, "cmd-2");
        var entry3 = new RaftLogEntry(3, 1, "cmd-3");

        await queue.EnqueueAsync(entry1);
        await queue.EnqueueAsync(entry2);
        await queue.EnqueueAsync(entry3);
        queue.Count.ShouldBe(3);

        (await queue.DequeueAsync()).ShouldBe(entry1);
        (await queue.DequeueAsync()).ShouldBe(entry2);
        (await queue.DequeueAsync()).ShouldBe(entry3);
        queue.Count.ShouldBe(0);
    }

    [Fact]
    public void TryDequeue_returns_false_when_empty()
    {
        var queue = new CommitQueue<RaftLogEntry>();
        queue.TryDequeue(out var item).ShouldBeFalse();
        item.ShouldBeNull();
    }

    [Fact]
    public async Task TryDequeue_returns_true_when_item_available()
    {
        var queue = new CommitQueue<RaftLogEntry>();
        var entry = new RaftLogEntry(1, 1, "cmd-1");
        await queue.EnqueueAsync(entry);

        queue.TryDequeue(out var item).ShouldBeTrue();
        item.ShouldBe(entry);
    }

    [Fact]
    public async Task Complete_prevents_further_enqueue()
    {
        var queue = new CommitQueue<RaftLogEntry>();
        await queue.EnqueueAsync(new RaftLogEntry(1, 1, "cmd-1"));
        queue.Complete();

        // After completion, writing should throw ChannelClosedException
        await Should.ThrowAsync<System.Threading.Channels.ChannelClosedException>(
            async () => await queue.EnqueueAsync(new RaftLogEntry(2, 1, "cmd-2")));
    }

    [Fact]
    public async Task Complete_allows_draining_remaining_items()
    {
        var queue = new CommitQueue<RaftLogEntry>();
        var entry = new RaftLogEntry(1, 1, "cmd-1");
        await queue.EnqueueAsync(entry);
        queue.Complete();

        // Should still be able to read remaining items
        var dequeued = await queue.DequeueAsync();
        dequeued.ShouldBe(entry);
    }

    [Fact]
    public void Count_reflects_current_queue_depth()
    {
        var queue = new CommitQueue<RaftLogEntry>();
        queue.Count.ShouldBe(0);
    }

    // -- RaftNode CommitIndex tracking tests --

    [Fact]
    public async Task CommitIndex_advances_when_proposal_succeeds_quorum()
    {
        var (nodes, _) = CreateCluster(3);
        var leader = ElectLeader(nodes);
        leader.CommitIndex.ShouldBe(0);

        var index1 = await leader.ProposeAsync("cmd-1", default);
        leader.CommitIndex.ShouldBe(index1);

        var index2 = await leader.ProposeAsync("cmd-2", default);
        leader.CommitIndex.ShouldBe(index2);
        index2.ShouldBeGreaterThan(index1);
    }

    [Fact]
    public async Task CommitIndex_starts_at_zero()
    {
        var node = new RaftNode("n1");
        node.CommitIndex.ShouldBe(0);
        await Task.CompletedTask;
    }

    // -- RaftNode ProcessedIndex tracking tests --

    [Fact]
    public void ProcessedIndex_starts_at_zero()
    {
        var node = new RaftNode("n1");
        node.ProcessedIndex.ShouldBe(0);
    }

    [Fact]
    public void MarkProcessed_advances_ProcessedIndex()
    {
        var node = new RaftNode("n1");
        node.MarkProcessed(5);
        node.ProcessedIndex.ShouldBe(5);
    }

    [Fact]
    public void MarkProcessed_does_not_go_backward()
    {
        var node = new RaftNode("n1");
        node.MarkProcessed(10);
        node.MarkProcessed(5);
        node.ProcessedIndex.ShouldBe(10);
    }

    [Fact]
    public async Task ProcessedIndex_tracks_separately_from_CommitIndex()
    {
        var (nodes, _) = CreateCluster(3);
        var leader = ElectLeader(nodes);

        var index1 = await leader.ProposeAsync("cmd-1", default);
        var index2 = await leader.ProposeAsync("cmd-2", default);

        // CommitIndex should have advanced
        leader.CommitIndex.ShouldBe(index2);

        // ProcessedIndex stays at 0 until explicitly marked
        leader.ProcessedIndex.ShouldBe(0);

        // Simulate state machine processing one entry
        leader.MarkProcessed(index1);
        leader.ProcessedIndex.ShouldBe(index1);

        // CommitIndex is still ahead of ProcessedIndex
        leader.CommitIndex.ShouldBeGreaterThan(leader.ProcessedIndex);

        // Process the second entry
        leader.MarkProcessed(index2);
        leader.ProcessedIndex.ShouldBe(index2);
        leader.ProcessedIndex.ShouldBe(leader.CommitIndex);
    }

    // -- CommitQueue integration with RaftNode --

    [Fact]
    public async Task CommitQueue_receives_entries_after_successful_quorum()
    {
        var (nodes, _) = CreateCluster(3);
        var leader = ElectLeader(nodes);

        var index1 = await leader.ProposeAsync("cmd-1", default);
        var index2 = await leader.ProposeAsync("cmd-2", default);

        // CommitQueue should have 2 entries
        leader.CommitQueue.Count.ShouldBe(2);

        // Dequeue and verify order
        var entry1 = await leader.CommitQueue.DequeueAsync();
        entry1.Index.ShouldBe(index1);
        entry1.Command.ShouldBe("cmd-1");

        var entry2 = await leader.CommitQueue.DequeueAsync();
        entry2.Index.ShouldBe(index2);
        entry2.Command.ShouldBe("cmd-2");
    }

    [Fact]
    public async Task CommitQueue_entries_match_committed_log_entries()
    {
        var (nodes, _) = CreateCluster(3);
        var leader = ElectLeader(nodes);

        await leader.ProposeAsync("alpha", default);
        await leader.ProposeAsync("beta", default);
        await leader.ProposeAsync("gamma", default);

        // Drain the commit queue and verify entries match log
        for (int i = 0; i < 3; i++)
        {
            var committed = await leader.CommitQueue.DequeueAsync();
            committed.ShouldBe(leader.Log.Entries[i]);
        }
    }

    [Fact]
    public async Task Non_leader_proposal_throws_and_does_not_affect_commit_queue()
    {
        var node = new RaftNode("follower");
        node.CommitQueue.Count.ShouldBe(0);

        await Should.ThrowAsync<InvalidOperationException>(
            async () => await node.ProposeAsync("cmd", default));

        node.CommitQueue.Count.ShouldBe(0);
        node.CommitIndex.ShouldBe(0);
    }
}