natsdotnet/tests/NATS.Server.JetStream.Tests/JetStream/Storage/FileStoreStreamStoreTests.cs

// Reference: golang/nats-server/server/filestore.go
// Tests in this file:
//   StoreRawMsg_stores_at_specified_sequence  — IStreamStore.StoreRawMsg preserves caller seq/ts
//   LoadPrevMsg_returns_message_before_seq   — IStreamStore.LoadPrevMsg backward scan
//   Type_returns_file                        — IStreamStore.Type() returns StorageType.File
//   Stop_prevents_further_writes             — IStreamStore.Stop() sets _stopped flag

using NATS.Server.JetStream.Storage;
using StorageType = NATS.Server.JetStream.Models.StorageType;

namespace NATS.Server.JetStream.Tests.JetStream.Storage;

/// <summary>
/// Tests for IStreamStore methods added to FileStore in Batch 1:
/// StoreRawMsg, LoadPrevMsg, Type, and Stop.
/// </summary>
public sealed class FileStoreStreamStoreTests : IDisposable
{
    private readonly string _root;

    public FileStoreStreamStoreTests()
    {
        _root = Path.Combine(Path.GetTempPath(), $"nats-js-sstest-{Guid.NewGuid():N}");
        Directory.CreateDirectory(_root);
    }

    public void Dispose()
    {
        // Best-effort cleanup of temp directory. If it fails (e.g. open handles on CI),
        // the OS will clean it up on the next reboot. Letting it throw would suppress
        // the real test failure so we absorb IO errors explicitly.
        if (!Directory.Exists(_root))
            return;

        try
        {
            Directory.Delete(_root, recursive: true);
        }
        catch (IOException ex)
        {
            // Open file handles (common on Windows CI) — log and continue.
            Console.Error.WriteLine($"[FileStoreStreamStoreTests] Dispose: {ex.Message}");
        }
        catch (UnauthorizedAccessException ex)
        {
            // Read-only files left by the test — log and continue.
            Console.Error.WriteLine($"[FileStoreStreamStoreTests] Dispose: {ex.Message}");
        }
    }

    private FileStore CreateStore(string subDir, FileStoreOptions? opts = null)
    {
        var dir = Path.Combine(_root, subDir);
        Directory.CreateDirectory(dir);
        var o = opts ?? new FileStoreOptions();
        o.Directory = dir;
        return new FileStore(o);
    }

    // -------------------------------------------------------------------------
    // StoreRawMsg
    // -------------------------------------------------------------------------

    // Go: filestore.go storeRawMsg — caller specifies seq and ts; store must not
    // auto-increment, and _last must be updated to Math.Max(_last, seq).
    [Fact]
    public void StoreRawMsg_stores_at_specified_sequence()
    {
        using var store = CreateStore("raw-seq");
        IStreamStore ss = store;

        var subject = "events.raw";
        var data = "hello raw"u8.ToArray();
        // Use a specific Unix nanosecond timestamp.
        var tsNs = new DateTimeOffset(2024, 6, 1, 12, 0, 0, TimeSpan.Zero).ToUnixTimeMilliseconds() * 1_000_000L;
        const ulong targetSeq = 42UL;

        ss.StoreRawMsg(subject, null, data, targetSeq, tsNs, 0, false);

        // Verify by loading the message back via LoadMsg.
        var sm = ss.LoadMsg(targetSeq, null);
        sm.Subject.ShouldBe(subject);
        sm.Sequence.ShouldBe(targetSeq);
        sm.Timestamp.ShouldBe(tsNs);
    }

    [Fact]
    public void StoreRawMsg_updates_last_watermark()
    {
        using var store = CreateStore("raw-wm");
        IStreamStore ss = store;

        // Store a message at seq 100 — _last should become 100.
        ss.StoreRawMsg("foo", null, "x"u8.ToArray(), 100UL, 1_000_000L, 0, false);

        var state = new StreamState();
        ss.FastState(ref state);
        state.LastSeq.ShouldBe(100UL);
    }

    [Fact]
    public void StoreRawMsg_does_not_decrement_last_for_lower_seq()
    {
        using var store = CreateStore("raw-order");
        IStreamStore ss = store;

        // Write seq 50 first, then seq 30 (out-of-order replication scenario).
        ss.StoreRawMsg("foo", null, "x"u8.ToArray(), 50UL, 1_000_000L, 0, false);
        ss.StoreRawMsg("bar", null, "y"u8.ToArray(), 30UL, 2_000_000L, 0, false);

        var state = new StreamState();
        ss.FastState(ref state);
        // _last should remain 50, not go down to 30.
        state.LastSeq.ShouldBe(50UL);
    }

    [Fact]
    public void StoreRawMsg_preserves_caller_timestamp()
    {
        using var store = CreateStore("raw-ts");
        IStreamStore ss = store;

        var subject = "ts.test";
        var data = "payload"u8.ToArray();
        // A deterministic Unix nanosecond timestamp.
        var tsNs = 1_717_238_400_000_000_000L; // 2024-06-01 00:00:00 UTC in ns

        ss.StoreRawMsg(subject, null, data, 7UL, tsNs, 0, false);

        var sm = ss.LoadMsg(7UL, null);
        sm.Timestamp.ShouldBe(tsNs);
    }

    [Fact]
    public void StoreRawMsg_throws_after_stop()
    {
        using var store = CreateStore("raw-stop");
        IStreamStore ss = store;

        ss.Stop();

        Should.Throw<ObjectDisposedException>(() =>
            ss.StoreRawMsg("foo", null, "x"u8.ToArray(), 1UL, 1_000_000L, 0, false));
    }

    // -------------------------------------------------------------------------
    // LoadPrevMsg
    // -------------------------------------------------------------------------

    // Go: filestore.go LoadPrevMsg — walks backward from start-1 to first.
    [Fact]
    public void LoadPrevMsg_returns_message_before_seq()
    {
        using var store = CreateStore("prev-basic");
        IStreamStore ss = store;

        // Write 3 messages at seqs 1, 2, 3.
        ss.StoreMsg("a", null, "msg1"u8.ToArray(), 0);
        ss.StoreMsg("b", null, "msg2"u8.ToArray(), 0);
        ss.StoreMsg("c", null, "msg3"u8.ToArray(), 0);

        // LoadPrevMsg(3) should return seq 2.
        var sm = ss.LoadPrevMsg(3UL, null);
        sm.Sequence.ShouldBe(2UL);
        sm.Subject.ShouldBe("b");
    }

    [Fact]
    public void LoadPrevMsg_skips_deleted_message()
    {
        using var store = CreateStore("prev-skip");
        IStreamStore ss = store;

        ss.StoreMsg("a", null, "msg1"u8.ToArray(), 0);  // seq 1
        ss.StoreMsg("b", null, "msg2"u8.ToArray(), 0);  // seq 2
        ss.StoreMsg("c", null, "msg3"u8.ToArray(), 0);  // seq 3

        // Delete seq 2 — LoadPrevMsg(3) must skip it and return seq 1.
        ss.RemoveMsg(2UL);

        var sm = ss.LoadPrevMsg(3UL, null);
        sm.Sequence.ShouldBe(1UL);
        sm.Subject.ShouldBe("a");
    }

    [Fact]
    public void LoadPrevMsg_throws_when_no_message_before_seq()
    {
        using var store = CreateStore("prev-none");
        IStreamStore ss = store;

        ss.StoreMsg("a", null, "msg1"u8.ToArray(), 0);  // seq 1

        // LoadPrevMsg(1) — nothing before seq 1.
        Should.Throw<KeyNotFoundException>(() => ss.LoadPrevMsg(1UL, null));
    }

    [Fact]
    public void LoadPrevMsg_reuses_provided_container()
    {
        using var store = CreateStore("prev-reuse");
        IStreamStore ss = store;

        ss.StoreMsg("x", null, "d1"u8.ToArray(), 0);  // seq 1
        ss.StoreMsg("y", null, "d2"u8.ToArray(), 0);  // seq 2

        var container = new StoreMsg();
        var result = ss.LoadPrevMsg(2UL, container);

        // Should return the same object reference.
        result.ShouldBeSameAs(container);
        container.Sequence.ShouldBe(1UL);
    }

    // -------------------------------------------------------------------------
    // Type
    // -------------------------------------------------------------------------

    // Go: filestore.go fileStore.Type — returns StorageType.File.
    [Fact]
    public void Type_returns_file()
    {
        using var store = CreateStore("type");
        IStreamStore ss = store;

        ss.Type().ShouldBe(StorageType.File);
    }

    // -------------------------------------------------------------------------
    // Stop
    // -------------------------------------------------------------------------

    // Go: filestore.go fileStore.Stop — flushes and marks as stopped.
    [Fact]
    public void Stop_prevents_further_writes_via_StoreMsg()
    {
        using var store = CreateStore("stop-storemsg");
        IStreamStore ss = store;

        ss.StoreMsg("ok", null, "before"u8.ToArray(), 0);

        ss.Stop();

        Should.Throw<ObjectDisposedException>(() =>
            ss.StoreMsg("fail", null, "after"u8.ToArray(), 0));
    }

    [Fact]
    public async Task Stop_prevents_further_writes_via_AppendAsync()
    {
        using var store = CreateStore("stop-append");

        await store.AppendAsync("ok", "before"u8.ToArray(), CancellationToken.None);

        ((IStreamStore)store).Stop();

        await Should.ThrowAsync<ObjectDisposedException>(() =>
            store.AppendAsync("fail", "after"u8.ToArray(), CancellationToken.None).AsTask());
    }

    [Fact]
    public void Stop_is_idempotent()
    {
        using var store = CreateStore("stop-idem");
        IStreamStore ss = store;

        ss.Stop();

        // Second Stop() must not throw.
        var ex = Record.Exception(() => ss.Stop());
        ex.ShouldBeNull();
    }

    [Fact]
    public void Stop_preserves_messages_on_disk()
    {
        var dir = Path.Combine(_root, "stop-persist");
        Directory.CreateDirectory(dir);

        FileStore CreateWithDir() => new FileStore(new FileStoreOptions { Directory = dir });

        // Write a message, stop the store.
        using (var store = CreateWithDir())
        {
            ((IStreamStore)store).StoreMsg("saved", null, "payload"u8.ToArray(), 0);
            ((IStreamStore)store).Stop();
        }

        // Re-open and verify the message survived.
        using var recovered = CreateWithDir();
        var sm = ((IStreamStore)recovered).LoadMsg(1UL, null);
        sm.Subject.ShouldBe("saved");
    }
}