natsnet/dotnet/tests/ZB.MOM.NatsNet.Server.Tests/ImplBacklog/ConcurrencyTests2.Impltests.cs

using System.Collections.Concurrent;
using System.Diagnostics;
using System.Reflection;
using Shouldly;
using ZB.MOM.NatsNet.Server;

namespace ZB.MOM.NatsNet.Server.Tests.ImplBacklog;

public sealed partial class ConcurrencyTests2
{
    [Fact] // T:2491
    public void NoRaceFileStoreMsgLoadNextMsgMultiPerf_ShouldSucceed()
    {
        WithStore((fs, _) =>
        {
            for (var i = 0; i < 150; i++)
                fs.StoreMsg($"ln.{i % 6}", null, "x"u8.ToArray(), 0);

            var errors = new ConcurrentQueue<Exception>();
            Parallel.For(0, 400, _ =>
            {
                try
                {
                    var (sm, _) = fs.LoadNextMsgMulti(new[] { "ln.1", "ln.*" }, 1, null);
                    if (sm != null)
                        sm.Subject.ShouldStartWith("ln.");
                }
                catch (Exception ex)
                {
                    errors.Enqueue(ex);
                }
            });

            errors.ShouldBeEmpty();
            fs.State().Msgs.ShouldBeGreaterThan(0UL);
        });
    }

    [Fact] // T:2501
    public void NoRaceFileStoreMsgLimitsAndOldRecoverState_ShouldSucceed()
    {
        var root = NewRoot();
        Directory.CreateDirectory(root);

        try
        {
            var cfg = DefaultStreamConfig(maxMsgs: 60);
            var fs1 = JetStreamFileStore.NewFileStore(new FileStoreConfig { StoreDir = root }, cfg);
            Parallel.For(0, 180, i => fs1.StoreMsg($"lm.{i % 4}", null, "x"u8.ToArray(), 0));
            fs1.Stop();

            var fs2 = JetStreamFileStore.NewFileStore(new FileStoreConfig { StoreDir = root }, cfg);
            var (seq, _) = fs2.StoreMsg("lm.tail", null, "tail"u8.ToArray(), 0);
            seq.ShouldBeGreaterThan(0UL);
            fs2.State().Msgs.ShouldBeLessThanOrEqualTo((ulong)cfg.MaxMsgs);
            fs2.Stop();
        }
        finally
        {
            Directory.Delete(root, recursive: true);
        }
    }

    [Fact] // T:2476
    public void NoRaceFilestoreBinaryStreamSnapshotEncodingLargeGaps_ShouldSucceed()
    {
        WithStore((fs, _) =>
        {
            const int numMsgs = 5000;
            var payload = new byte[128];

            fs.StoreMsg("zzz", null, payload, 0).Seq.ShouldBe(1UL);
            for (var i = 2; i < numMsgs; i++)
            {
                var (seq, _) = fs.StoreMsg("zzz", null, null, 0);
                seq.ShouldBeGreaterThan(1UL);
                fs.RemoveMsg(seq).Removed.ShouldBeTrue();
            }

            fs.StoreMsg("zzz", null, payload, 0).Seq.ShouldBe((ulong)numMsgs);
            Should.NotThrow(() => InvokePrivate(fs, "SyncBlocks"));

            var (snapshot, err) = fs.EncodedStreamState(0);
            err.ShouldBeNull();
            StoreParity.IsEncodedStreamState(snapshot).ShouldBeTrue();
            snapshot.Length.ShouldBeLessThan(2048);

            var state = fs.State();
            state.FirstSeq.ShouldBe(1UL);
            state.LastSeq.ShouldBe((ulong)numMsgs);
            state.Msgs.ShouldBe(2UL);
            state.NumDeleted.ShouldBe(numMsgs - 2);
        });
    }

    [Fact] // T:2480
    public void NoRaceFileStoreLargeMsgsAndFirstMatching_ShouldSucceed()
    {
        var cfg = DefaultStreamConfig();
        cfg.Subjects = [">"];

        WithStore((fs, _) =>
        {
            for (var i = 0; i < 4_000; i++)
                fs.StoreMsg($"foo.bar.{i}", null, null, 0);
            for (var i = 0; i < 4_000; i++)
                fs.StoreMsg($"foo.baz.{i}", null, null, 0);

            var blocks = InvokePrivate<int>(fs, "NumMsgBlocks");
            blocks.ShouldBeGreaterThanOrEqualTo(1);

            var start = fs.State().FirstSeq;
            for (var seq = start; seq < start + 7_600; seq++)
                fs.RemoveMsg(seq).Removed.ShouldBeTrue();

            var sw = System.Diagnostics.Stopwatch.StartNew();
            var (sm, _) = fs.LoadNextMsg("*.baz.*", true, start, null);
            sw.Stop();

            sm.ShouldNotBeNull();
            sm!.Subject.ShouldContain(".baz.");
            sw.ElapsedMilliseconds.ShouldBeLessThan(50);
        }, cfg);
    }

    [Fact] // T:2494
    public void NoRaceFileStoreWriteFullStateUniqueSubjects_ShouldSucceed()
    {
        var cfg = new StreamConfig
        {
            Name = "TEST",
            Storage = StorageType.FileStorage,
            Subjects = ["records.>"],
            MaxMsgs = -1,
            MaxBytes = 15L * 1024 * 1024 * 1024,
            MaxAge = TimeSpan.Zero,
            MaxMsgsPer = 1,
            Discard = DiscardPolicy.DiscardOld,
            Retention = RetentionPolicy.LimitsPolicy,
        };

        WithStore((fs, root) =>
        {
            var payload = Enumerable.Repeat((byte)'Z', 128).ToArray();
            var errors = new ConcurrentQueue<Exception>();
            using var cts = new CancellationTokenSource();

            var writer = Task.Run(async () =>
            {
                while (!cts.Token.IsCancellationRequested)
                {
                    try
                    {
                        var err = InvokePrivate<Exception?>(fs, "WriteFullState");
                        if (err != null)
                            errors.Enqueue(err);
                    }
                    catch (Exception ex)
                    {
                        errors.Enqueue(ex);
                    }

                    try
                    {
                        await Task.Delay(10, cts.Token);
                    }
                    catch (OperationCanceledException)
                    {
                        break;
                    }
                }
            });

            for (var i = 0; i < 2_000; i++)
            {
                var subject = $"records.{Guid.NewGuid():N}.{i % 5}";
                var sw = Stopwatch.StartNew();
                fs.StoreMsg(subject, null, payload, 0).Seq.ShouldBeGreaterThan(0UL);
                sw.Stop();
                sw.Elapsed.ShouldBeLessThan(TimeSpan.FromMilliseconds(500));
            }

            cts.Cancel();
            Should.NotThrow(() => writer.Wait(TimeSpan.FromSeconds(2)));
            errors.ShouldBeEmpty();

            fs.Stop();
            var stateFile = Path.Combine(root, FileStoreDefaults.MsgDir, FileStoreDefaults.StreamStateFile);
            File.Exists(stateFile).ShouldBeTrue();
            new FileInfo(stateFile).Length.ShouldBeGreaterThan(0L);
        }, cfg);
    }

    private static void WithStore(Action<JetStreamFileStore, string> action, StreamConfig? cfg = null)
    {
        var root = NewRoot();
        Directory.CreateDirectory(root);
        JetStreamFileStore? fs = null;

        try
        {
            fs = JetStreamFileStore.NewFileStore(new FileStoreConfig { StoreDir = root }, cfg ?? DefaultStreamConfig());
            action(fs, root);
        }
        finally
        {
            fs?.Stop();
            if (Directory.Exists(root))
                Directory.Delete(root, recursive: true);
        }
    }

    private static StreamConfig DefaultStreamConfig(long maxMsgs = -1)
    {
        return new StreamConfig
        {
            Name = "TEST",
            Storage = StorageType.FileStorage,
            Subjects = ["test.>"],
            MaxMsgs = maxMsgs,
            MaxBytes = -1,
            MaxAge = TimeSpan.Zero,
            MaxMsgsPer = -1,
            Discard = DiscardPolicy.DiscardOld,
            Retention = RetentionPolicy.LimitsPolicy,
        };
    }

    private static void InvokePrivate(object target, string methodName, params object[] args)
    {
        var method = target.GetType().GetMethod(methodName, BindingFlags.Instance | BindingFlags.NonPublic);
        method.ShouldNotBeNull();
        method!.Invoke(target, args);
    }

    private static T InvokePrivate<T>(object target, string methodName, params object[] args)
    {
        var method = target.GetType().GetMethod(methodName, BindingFlags.Instance | BindingFlags.NonPublic);
        method.ShouldNotBeNull();
        var result = method!.Invoke(target, args);
        if (result == null)
            return default!;
        return (T)result;
    }

    private static string NewRoot() => Path.Combine(Path.GetTempPath(), $"impl-fs-c2-{Guid.NewGuid():N}");
}