CBDDC/tests/ZB.MOM.WW.CDBBC.E2E.Benchmark.Tests/OfflineResyncThroughputBenchmarks.cs

using BenchmarkDotNet.Attributes;
using System.Net;
using System.Net.Sockets;
using ZB.MOM.WW.CBDDC.Core.Network;
using ZB.MOM.WW.CBDDC.Sample.Console;

namespace ZB.MOM.WW.CDBBC.E2E.Benchmark.Tests;

[MemoryDiagnoser]
[SimpleJob(launchCount: 1, warmupCount: 0, iterationCount: 1)]
public class OfflineResyncThroughputBenchmarks
{
    private const int BacklogOperationCount = 100_000;
    private BenchmarkPeerNode _onlineNode = null!;
    private BenchmarkPeerNode _offlineNode = null!;
    private int _runSequence;
    private string _currentPrefix = string.Empty;

    /// <summary>
    ///     Sets up benchmark resources for offline resync scenarios.
    /// </summary>
    [GlobalSetup]
    public Task GlobalSetupAsync()
    {
        return Task.CompletedTask;
    }

    /// <summary>
    ///     Handles benchmark teardown for offline resync scenarios.
    /// </summary>
    [GlobalCleanup]
    public Task GlobalCleanupAsync()
    {
        // Avoid explicit node disposal in BenchmarkDotNet child processes due Surreal embedded callback race.
        // Process teardown releases resources after benchmark completion.
        return Task.CompletedTask;
    }

    /// <summary>
    ///     Prepares write-only workload state for the 100K throughput benchmark.
    /// </summary>
    [IterationSetup(Target = nameof(OfflineBacklogWriteThroughput100k))]
    public void SetupOfflineWriteThroughput()
    {
        _currentPrefix = $"offline-write-{Interlocked.Increment(ref _runSequence):D6}";
        InitializeIterationNodesAsync().GetAwaiter().GetResult();
    }

    /// <summary>
    ///     Measures offline backlog write throughput for 100K operations.
    /// </summary>
    [Benchmark(Description = "Offline backlog write throughput (100K ops)", OperationsPerInvoke = BacklogOperationCount)]
    public async Task OfflineBacklogWriteThroughput100k()
    {
        await WriteBatchAsync(_currentPrefix, BacklogOperationCount);
    }

    /// <summary>
    ///     Prepares nodes and backlog before the re-sync benchmark iteration.
    /// </summary>
    [IterationSetup(Target = nameof(OfflineNodeResyncDurationAfter100kBacklog))]
    public void SetupOfflineResyncBenchmark()
    {
        _currentPrefix = $"offline-resync-{Interlocked.Increment(ref _runSequence):D6}";
        InitializeIterationNodesAsync().GetAwaiter().GetResult();
        WriteBatchAsync(_currentPrefix, BacklogOperationCount).GetAwaiter().GetResult();
    }

    /// <summary>
    ///     Measures re-sync duration after processing a 100K-entry offline backlog.
    /// </summary>
    [Benchmark(Description = "Offline node re-sync duration after 100K backlog")]
    public async Task OfflineNodeResyncDurationAfter100kBacklog()
    {
        await _offlineNode.StartAsync();
        await WaitForReplicationAsync(_currentPrefix, BacklogOperationCount, TimeSpan.FromMinutes(3));
    }

    private async Task WriteBatchAsync(string prefix, int count)
    {
        for (var i = 0; i < count; i++)
        {
            string userId = $"{prefix}-{i:D6}";
            await _onlineNode.UpsertUserAsync(CreateUser(userId));
        }
    }

    private async Task WaitForReplicationAsync(string prefix, int expectedCount, TimeSpan timeout)
    {
        DateTime deadline = DateTime.UtcNow.Add(timeout);
        while (DateTime.UtcNow < deadline)
        {
            if (_offlineNode.CountUsersWithPrefix(prefix) >= expectedCount)
                return;

            await Task.Delay(250);
        }

        int replicatedCount = _offlineNode.CountUsersWithPrefix(prefix);
        throw new TimeoutException(
            $"Timed out waiting for re-sync. Expected {expectedCount}, replicated {replicatedCount}.");
    }

    private static User CreateUser(string userId)
    {
        return new User
        {
            Id = userId,
            Name = $"user-{userId}",
            Age = 30,
            Address = new Address { City = "OfflineBenchmarkCity" }
        };
    }

    private static int GetAvailableTcpPort()
    {
        using var listener = new TcpListener(IPAddress.Loopback, 0);
        listener.Start();
        return ((IPEndPoint)listener.LocalEndpoint).Port;
    }

    private async Task InitializeIterationNodesAsync()
    {
        int onlinePort = GetAvailableTcpPort();
        int offlinePort = GetAvailableTcpPort();
        while (offlinePort == onlinePort)
            offlinePort = GetAvailableTcpPort();

        string clusterToken = Guid.NewGuid().ToString("N");

        _onlineNode = BenchmarkPeerNode.Create(
            "offline-benchmark-online",
            onlinePort,
            clusterToken,
            [
                new KnownPeerConfiguration
                {
                    NodeId = "offline-benchmark-offline",
                    Host = "127.0.0.1",
                    Port = offlinePort
                }
            ]);

        _offlineNode = BenchmarkPeerNode.Create(
            "offline-benchmark-offline",
            offlinePort,
            clusterToken,
            [
                new KnownPeerConfiguration
                {
                    NodeId = "offline-benchmark-online",
                    Host = "127.0.0.1",
                    Port = onlinePort
                }
            ]);

        await _onlineNode.StartAsync();
        await Task.Delay(250);
    }
}