ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.ResidualBench/AuditContentionBench.cs

using System.Diagnostics;
using Microsoft.Data.SqlClient;
using Microsoft.EntityFrameworkCore;
using ZB.MOM.WW.Audit;
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase;
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Repositories;

namespace ZB.MOM.WW.ScadaBridge.ResidualBench;

/// <summary>
/// ScadaBridge#124: concurrent callers of <see cref="AuditLogRepository.InsertManyIfNotExistsAsync"/>
/// — the shape the ingest singleton, the reconciliation singleton and the cached-telemetry
/// dual-write produce when they overlap. Runs N writers against a throwaway migrated database
/// for a fixed time and reports throughput and the deadlock victims (error 1205) that reached
/// the caller. One writer in N re-sends a packet another writer already inserted, the way a
/// reconciliation pull overlaps live telemetry.
/// </summary>
internal static class AuditContentionBench
{
    private const int PacketSize = 256;

    public static async Task<int> RunAsync(BenchArgs args)
    {
        var admin = args.Get("connection-string", AuditBench.DefaultAdmin);
        var writerCount = args.GetInt("writers", 4);
        var seconds = args.GetInt("seconds", 30);
        var seedRows = args.GetInt("seed-rows", 50_000);
        args.RejectUnknown();

        var dbName = $"ScadaBridgeResidualBench_{Guid.NewGuid():N}"[..40];
        var cs = new SqlConnectionStringBuilder(admin) { InitialCatalog = dbName }.ConnectionString;
        await AuditBench.ExecAsync(admin, $"CREATE DATABASE [{dbName}]; ALTER DATABASE [{dbName}] SET RECOVERY SIMPLE;");
        Console.WriteLine($"# database {dbName}, {writerCount} writers, {seconds} s, {seedRows:N0} seed rows");
        try
        {
            var options = new DbContextOptionsBuilder<ScadaBridgeDbContext>()
                .UseSqlServer(cs, sql => sql.CommandTimeout(600)).Options;
            await using (var ctx = new ScadaBridgeDbContext(options)) await ctx.Database.MigrateAsync();

            var now = DateTime.UtcNow;
            for (var p = 0; p < seedRows / PacketSize; p++)
            {
                var rows = Enumerable.Range(0, PacketSize)
                    .Select(i => AuditBench.Row((long)p * PacketSize + i, now.AddHours(-2).AddMilliseconds((p * PacketSize) + i)))
                    .ToList();
                await using var ctx = new ScadaBridgeDbContext(options);
                await new AuditLogRepository(ctx).InsertManyIfNotExistsAsync(rows);
            }

            long inserted = 0, packets = 0, deadlocks = 0, otherFaults = 0, seq = 1_000_000_000;
            var recent = new System.Collections.Concurrent.ConcurrentQueue<IReadOnlyList<AuditEvent>>();
            using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(seconds));
            var sw = Stopwatch.StartNew();

            async Task Writer(int index)
            {
                while (!cts.IsCancellationRequested)
                {
                    IReadOnlyList<AuditEvent> rows;
                    // Writer 0 replays an already-sent packet now and then (reconciliation overlap).
                    if (index == 0 && writerCount > 1 && Interlocked.Read(ref packets) % 4 == 3 && recent.TryDequeue(out var replay))
                    {
                        rows = replay;
                    }
                    else
                    {
                        var start = Interlocked.Add(ref seq, PacketSize) - PacketSize;
                        var t = DateTime.UtcNow;
                        rows = Enumerable.Range(0, PacketSize)
                            .Select(i => AuditBench.Row(start + i, t.AddMilliseconds(-i)))
                            .ToList();
                        recent.Enqueue(rows);
                        while (recent.Count > 64) recent.TryDequeue(out _);
                    }

                    try
                    {
                        await using var ctx = new ScadaBridgeDbContext(options);
                        Interlocked.Add(ref inserted, await new AuditLogRepository(ctx).InsertManyIfNotExistsAsync(rows));
                        Interlocked.Increment(ref packets);
                    }
                    catch (SqlException ex) when (ex.Number == 1205)
                    {
                        Interlocked.Increment(ref deadlocks);
                    }
                    catch (Exception ex)
                    {
                        if (Interlocked.Increment(ref otherFaults) <= 3) Console.WriteLine($"fault: {ex.GetBaseException().Message}");
                    }
                }
            }

            await Task.WhenAll(Enumerable.Range(0, writerCount).Select(i => Task.Run(() => Writer(i))));
            Console.WriteLine(
                $"{writerCount} writers: {packets:N0} packets, {inserted:N0} rows in {sw.Elapsed.TotalSeconds:F0} s = " +
                $"{inserted / sw.Elapsed.TotalSeconds:N0} rows/s; deadlock victims reaching the caller {deadlocks:N0}; other faults {otherFaults:N0}");
        }
        finally
        {
            SqlConnection.ClearAllPools();
            await AuditBench.ExecAsync(admin, $"ALTER DATABASE [{dbName}] SET SINGLE_USER WITH ROLLBACK IMMEDIATE; DROP DATABASE [{dbName}];");
        }
        return 0;
    }
}