natsdotnet/tests/NATS.Server.JetStream.Tests/JetStream/ConsumerDeliveryParityTests.cs

// Ported from golang/nats-server/server/jetstream_consumer_test.go
// Covers: consumer creation, deliver policies (All, Last, New, ByStartSequence, ByStartTime),
// and ack policies (None, Explicit, All) as modelled in the .NET port.
//
// Go reference tests:
//   TestJetStreamConsumerCreate               (~line 2967)
//   TestJetStreamConsumerWithStartTime        (~line 3160)
//   TestJetStreamConsumerMaxDeliveries        (~line 3265)
//   TestJetStreamConsumerAckFloorFill         (~line 3404)
//   TestJetStreamConsumerReplayRateNoAck      (~line 4505)

using System.Text;
using NATS.Server.JetStream;
using NATS.Server.JetStream.Consumers;
using NATS.Server.JetStream.Models;
using NATS.Server.JetStream.Storage;

namespace NATS.Server.JetStream.Tests.JetStream;

/// <summary>
/// Consumer delivery parity tests ported from the Go reference implementation.
/// These tests exercise push/pull delivery, deliver policies, and ack policies against
/// the in-process ConsumerManager + StreamManager, mirroring the semantics validated in
/// golang/nats-server/server/jetstream_consumer_test.go.
/// </summary>
public class ConsumerDeliveryParityTests
{
    // -------------------------------------------------------------------------
    // Test 1 – Pull consumer with DeliverPolicy.All returns all published msgs
    //
    // Go reference: TestJetStreamConsumerCreate – verifies that a durable pull
    // consumer created with default settings fetches all stored messages in
    // sequence order.
    // -------------------------------------------------------------------------
    [Fact]
    public async Task Pull_consumer_deliver_all_returns_messages_in_sequence_order()
    {
        var streams = new StreamManager();
        streams.CreateOrUpdate(new StreamConfig
        {
            Name = "ORDERS",
            Subjects = ["orders.*"],
        }).Error.ShouldBeNull();

        var consumers = new ConsumerManager();
        consumers.CreateOrUpdate("ORDERS", new ConsumerConfig
        {
            DurableName = "PULL",
            DeliverPolicy = DeliverPolicy.All,
        }).Error.ShouldBeNull();

        streams.Capture("orders.created", "msg-1"u8.ToArray());
        streams.Capture("orders.updated", "msg-2"u8.ToArray());
        streams.Capture("orders.created", "msg-3"u8.ToArray());

        var batch = await consumers.FetchAsync("ORDERS", "PULL", 3, streams, default);

        batch.Messages.Count.ShouldBe(3);
        batch.Messages[0].Sequence.ShouldBe((ulong)1);
        batch.Messages[1].Sequence.ShouldBe((ulong)2);
        batch.Messages[2].Sequence.ShouldBe((ulong)3);
    }

    // -------------------------------------------------------------------------
    // Test 2 – Deliver policy Last starts at the final stored sequence
    //
    // Go reference: TestJetStreamConsumerWithMultipleStartOptions – verifies
    // that DeliverLast causes the consumer cursor to begin at the last message
    // in the stream rather than seq 1.
    // -------------------------------------------------------------------------
    [Fact]
    public async Task Pull_consumer_deliver_last_starts_at_final_sequence()
    {
        var streams = new StreamManager();
        streams.CreateOrUpdate(new StreamConfig
        {
            Name = "ORDERS",
            Subjects = ["orders.*"],
        }).Error.ShouldBeNull();

        streams.Capture("orders.a", "first"u8.ToArray());
        streams.Capture("orders.b", "second"u8.ToArray());
        streams.Capture("orders.c", "third"u8.ToArray());

        var consumers = new ConsumerManager();
        consumers.CreateOrUpdate("ORDERS", new ConsumerConfig
        {
            DurableName = "LAST",
            DeliverPolicy = DeliverPolicy.Last,
        }).Error.ShouldBeNull();

        var batch = await consumers.FetchAsync("ORDERS", "LAST", 5, streams, default);

        // DeliverLast cursor resolves to sequence 3 (last stored).
        batch.Messages.Count.ShouldBe(1);
        batch.Messages[0].Sequence.ShouldBe((ulong)3);
    }

    // -------------------------------------------------------------------------
    // Test 3 – Deliver policy New skips all messages present at first-fetch time
    //
    // Go reference: TestJetStreamConsumerDeliverNewNotConsumingBeforeRestart
    // (~line 6213) – validates that DeliverNew positions the cursor past the
    // last stored sequence so that messages already in the stream when the
    // consumer first fetches are not returned.
    //
    // In the .NET port the initial sequence is resolved on the first FetchAsync
    // call (when NextSequence == 1). DeliverPolicy.New sets the cursor to
    // lastSeq + 1, so every message present at fetch time is skipped and only
    // subsequent publishes are visible.
    // -------------------------------------------------------------------------
    [Fact]
    public async Task Pull_consumer_deliver_new_skips_messages_present_at_first_fetch()
    {
        var streams = new StreamManager();
        streams.CreateOrUpdate(new StreamConfig
        {
            Name = "ORDERS",
            Subjects = ["orders.*"],
        }).Error.ShouldBeNull();

        streams.Capture("orders.a", "pre-1"u8.ToArray());
        streams.Capture("orders.b", "pre-2"u8.ToArray());

        var consumers = new ConsumerManager();
        consumers.CreateOrUpdate("ORDERS", new ConsumerConfig
        {
            DurableName = "NEW",
            DeliverPolicy = DeliverPolicy.New,
        }).Error.ShouldBeNull();

        // First fetch: resolves cursor to lastSeq+1 = 3, which has no message yet.
        var empty = await consumers.FetchAsync("ORDERS", "NEW", 5, streams, default);
        empty.Messages.Count.ShouldBe(0);

        // Now publish a new message – this is the "new" message after the cursor.
        streams.Capture("orders.c", "post-1"u8.ToArray());

        // Second fetch: cursor is already at 3, the newly published message is at 3.
        var batch = await consumers.FetchAsync("ORDERS", "NEW", 5, streams, default);
        batch.Messages.Count.ShouldBe(1);
        batch.Messages[0].Sequence.ShouldBe((ulong)3);
    }

    // -------------------------------------------------------------------------
    // Test 4 – Deliver policy ByStartTime resolves cursor at the correct seq
    //
    // Go reference: TestJetStreamConsumerWithStartTime (~line 3160) – publishes
    // messages before a recorded timestamp, then creates a consumer with
    // DeliverByStartTime and verifies the first delivered sequence matches the
    // first message after that timestamp.
    // -------------------------------------------------------------------------
    [Fact]
    public async Task Pull_consumer_deliver_by_start_time_resolves_correct_starting_sequence()
    {
        var streams = new StreamManager();
        streams.CreateOrUpdate(new StreamConfig
        {
            Name = "ORDERS",
            Subjects = ["orders.*"],
        }).Error.ShouldBeNull();

        streams.Capture("orders.a", "before-1"u8.ToArray());
        streams.Capture("orders.b", "before-2"u8.ToArray());

        // Brief pause so that stored timestamps of pre-existing messages are
        // strictly before the cut point we are about to record.
        await Task.Delay(10);
        var startTime = DateTime.UtcNow;

        streams.Capture("orders.c", "after-1"u8.ToArray());
        streams.Capture("orders.d", "after-2"u8.ToArray());

        var consumers = new ConsumerManager();
        consumers.CreateOrUpdate("ORDERS", new ConsumerConfig
        {
            DurableName = "BYTIME",
            DeliverPolicy = DeliverPolicy.ByStartTime,
            OptStartTimeUtc = startTime,
        }).Error.ShouldBeNull();

        var batch = await consumers.FetchAsync("ORDERS", "BYTIME", 5, streams, default);

        // Only messages with timestamp >= startTime should be returned.
        batch.Messages.Count.ShouldBe(2);
        batch.Messages.All(m => m.Sequence >= 3).ShouldBeTrue();
    }

    // -------------------------------------------------------------------------
    // Test 5 – AckAll advances the ack floor and blocks re-delivery of acked msgs
    //
    // Go reference: TestJetStreamConsumerAckFloorFill (~line 3404) – publishes
    // four messages, acks all via AckAll on seq 4, and then verifies that a
    // subsequent fetch returns zero messages because every sequence is at or
    // below the ack floor.
    // -------------------------------------------------------------------------
    [Fact]
    public async Task Explicit_ack_all_advances_floor_and_suppresses_redelivery()
    {
        var streams = new StreamManager();
        streams.CreateOrUpdate(new StreamConfig
        {
            Name = "ORDERS",
            Subjects = ["orders.*"],
        }).Error.ShouldBeNull();

        var consumers = new ConsumerManager();
        consumers.CreateOrUpdate("ORDERS", new ConsumerConfig
        {
            DurableName = "ACK",
            AckPolicy = AckPolicy.Explicit,
            AckWaitMs = 100,
        }).Error.ShouldBeNull();

        for (var i = 1; i <= 4; i++)
            streams.Capture("orders.created", Encoding.UTF8.GetBytes($"msg-{i}"));

        var first = await consumers.FetchAsync("ORDERS", "ACK", 4, streams, default);
        first.Messages.Count.ShouldBe(4);

        // AckAll up to sequence 4 should advance floor and clear all pending.
        consumers.AckAll("ORDERS", "ACK", 4);

        // A subsequent fetch must return no messages because the ack floor
        // now covers all published sequences and there are no new messages.
        var second = await consumers.FetchAsync("ORDERS", "ACK", 4, streams, default);
        second.Messages.Count.ShouldBe(0);
    }
}