AyCode.Core/AyCode.Services.Server.Tests/SignalRs/AcBinaryProtocolThreadPoolTests.cs

using System.IO.Pipelines;
using AyCode.Services.SignalRs;
using Microsoft.AspNetCore.SignalR.Protocol;

namespace AyCode.Services.Server.Tests.SignalRs;

/// <summary>
/// Thread-pool starvation regression tests for the AsyncSegment paths
/// (<c>ACCORE-SBP-I-T3W9</c>: blocking waits on .NET thread-pool threads deadlock the pool under
/// concurrent load + real network latency — production-only, "stuck loading → 60 s timeout").
///
/// <para><b>Deser (receive) side — fixed.</b> The streaming deser blocks in
/// <c>ManualResetEventSlim.Wait()</c> between network-delivered chunks. It now runs off the .NET
/// thread pool on a never-queue + reuse dedicated executor (<see cref="AcBinaryDeserExecutor"/>) so
/// the blocking wait never consumes a pool thread (which the producer's receive-continuations need
/// → circular dependency). These tests assert that executor contract directly and should all pass.</para>
///
/// <para><b>Ser (send) side — TBD.</b> The per-chunk <c>SyncFlush</c> in <c>WriteMessageChunked</c>
/// blocks a dispatch (pool) thread under client backpressure. The send-side test below asserts the
/// DESIRED non-blocking behaviour and is <c>[Ignore]</c>d until that fix lands.</para>
/// </summary>
[TestClass]
public class AcBinaryProtocolThreadPoolTests
{
    // ---------------------------------------------------------------------------------------------
    // Deser (receive) side — AcBinaryDeserExecutor contract. Expected: all pass.
    // ---------------------------------------------------------------------------------------------

    [TestMethod]
    public void DeserExecutor_RunsOffThreadPool()
    {
        var onPool = (bool)AcBinaryDeserExecutor.Run(() => (object?)Thread.CurrentThread.IsThreadPoolThread).GetAwaiter().GetResult()!;

        Assert.IsFalse(onPool,
            "Streaming deser must run OFF the .NET thread pool (dedicated thread) — the blocking " +
            "MRES.Wait() would otherwise pin a pool thread the producer's receive-continuation needs " +
            "→ circular-dependency starvation (ACCORE-SBP-I-T3W9).");
    }

    [TestMethod]
    public void DeserExecutor_NeverQueues_AllConcurrentWorkStartsImmediately()
    {
        const int n = 32;
        using var allStarted = new CountdownEvent(n);
        using var release = new ManualResetEventSlim(false);

        var tasks = new Task<object?>[n];
        for (var i = 0; i < n; i++)
        {
            tasks[i] = AcBinaryDeserExecutor.Run(() =>
            {
                allStarted.Signal();
                release.Wait();

                return (object?)null;
            });
        }

        // Every item must START even though none has finished — proves the executor NEVER queues:
        // each submit immediately gets a thread (reuse idle or grow). A bounded-queue executor
        // would stall here (excess items wait behind busy workers) and the countdown never reaches 0.
        // 5 s is generous for 32 thread spawns (~ms) yet fails fast on a queue regression.
        Assert.IsTrue(allStarted.Wait(TimeSpan.FromSeconds(5)),
            $"Only {n - allStarted.CurrentCount}/{n} work items started — executor queued instead of " +
            "growing on demand (would reintroduce the [202] GetResult-on-queued-deser coupling).");

        release.Set();
        Task.WaitAll(tasks);
    }

    [TestMethod]
    public void DeserExecutor_ReusesThreads_AcrossSequentialWork()
    {
        var ids = new HashSet<int>();
        for (var i = 0; i < 50; i++)
        {
            var id = (int)AcBinaryDeserExecutor.Run(() => (object?)Environment.CurrentManagedThreadId).GetAwaiter().GetResult()!;
            ids.Add(id);
        }

        // Sequential work: each completes before the next is submitted → the same idle worker is
        // reused (no per-message thread churn). The worker pushes itself to idle right after
        // SetResult (1 instruction) while the test thread runs ~10+ (incl. a TCS alloc) before the
        // next TryPop, so the worker almost always wins the race → 1-2 distinct threads in practice.
        // <= 4 leaves a small margin for the rare preempt-in-window case; more than that is suspect.
        Assert.IsTrue(ids.Count <= 4, $"Expected thread reuse across sequential work, got {ids.Count} distinct threads / 50 — churn instead of reuse.");
    }

    [TestMethod]
    [DoNotParallelize] // stresses the shared process-wide thread pool
    public void DeserExecutor_CompletesUnderThreadPoolPressure()
    {
        // Occupy many .NET thread-pool threads with blocking work, then run a deser-style item.
        // Because the executor is off-pool it completes regardless of pool pressure — whereas the
        // old `Task.Run` deser would queue behind the busy pool and the [202] GetResult would block
        // (ACCORE-SBP-I-T3W9). This is the end-to-end resilience guard.
        using var release = new ManualResetEventSlim(false);
        var hogCount = Environment.ProcessorCount * 4;
        var hogs = new Task[hogCount];

        for (var i = 0; i < hogCount; i++) hogs[i] = Task.Run(() => release.Wait());

        try
        {
            var done = AcBinaryDeserExecutor.Run(() => (object?)42);
            Assert.IsTrue(done.Wait(TimeSpan.FromSeconds(5)), "Off-pool deser did not complete under thread-pool pressure — possible pool coupling.");
            Assert.AreEqual(42, (int)done.Result!);
        }
        finally
        {
            release.Set();
            Task.WaitAll(hogs);
        }
    }

    // ---------------------------------------------------------------------------------------------
    // Ser (send) side — known-open. Asserts DESIRED behaviour; [Ignore]d until the send-side fix.
    // ---------------------------------------------------------------------------------------------

    [TestMethod]
    [Ignore("Send-side fix TBD — ACCORE-SBP-I-T3W9. Asserts the DESIRED non-blocking behaviour; " +
            "today WriteMessageChunked blocks the calling (dispatch) thread under client backpressure " +
            "via the per-chunk SyncFlush. Un-ignore when the send-side fix lands.")]
    public void WriteMessageChunked_UnderBackpressure_DoesNotBlockCallingThread()
    {
        // A Pipe with a small pause threshold and NO reader → FlushAsync backpressures immediately
        // (simulates a stuck / slow mobile client). The chunked send's per-chunk SyncFlush then
        // blocks the calling thread. Desired post-fix behaviour: the send does not pin the caller.
        var pipe = new Pipe(new PipeOptions(pauseWriterThreshold: 4096, resumeWriterThreshold: 2048));

        var opts = new AcBinaryHubProtocolOptions
        {
            ProtocolMode = BinaryProtocolMode.AsyncSegment,
            FlushTimeout = TimeSpan.FromSeconds(2), // bound so a stuck flush can't hang the test forever
        };
        var protocol = new AyCodeBinaryHubProtocol(opts);

        // Large non-byte[] data arg → the chunked path engages and produces > pauseThreshold bytes.
        // The 4-arg shape mirrors the OnReceiveMessage(tag, requestId, SignalParams, data) convention
        // the other SignalR tests use — SignalParams here is convention/realism, not functionally
        // required (the large `bigData` arg alone triggers HasStreamableArgs → chunked send).
        var bigData = new string('x', 200_000);
        var msg = new InvocationMessage("OnReceiveMessage", new object?[]
        {
            1, (int?)1, new SignalParams(), bigData
        });

        var writeThread = new Thread(() =>
        {
            try
            {
                protocol.WriteMessage(msg, pipe.Writer);
            }
            catch 
            {
                /* TimeoutException on the bounded flush is acceptable for this probe */ 
            }
        })
        { IsBackground = true };
        writeThread.Start();

        // 2 s margin so a non-blocking send (returns in ms once it no longer waits on the flush) is
        // not falsely flagged on a slow CI; a blocking send stays pinned well past this.
        var finishedPromptly = writeThread.Join(TimeSpan.FromSeconds(2));
        Assert.IsTrue(finishedPromptly,
            "WriteMessageChunked blocked the calling thread under client backpressure (per-chunk " +
            "SyncFlush) — the send-side thread-pool starvation source (ACCORE-SBP-I-T3W9).");
    }
}