Deadlock after Connect: conditional transaction holds the write lock while the handshake's INFO replication write blocks the reader (RedisTimeoutException on UNWATCH) #3348

// Repro: a conditional transaction sent on a multiplexer whose connect has just returned can stall until the
// timeout and fail on UNWATCH.
//
// usage: dotnet run -c Release -- <endpoint> [mode=fresh] [parallel=16] [iterations=500] [--timeout=5000] [--min-threads=N]
//   fresh     each task calls Connect and immediately executes the conditional transaction; stops at the first timeout
//   async     like fresh, but each task connects with ConnectAsync
//   control   the same connects, all awaited first, then the same transactions
//   ping      ConnectAsync, then PingAsync, then the transaction
//   delay     ConnectAsync, then a 1 s delay, then the transaction
//   nocond    like async, but the transaction has no condition
//   compare   runs fresh, async, control, ping, delay and nocond in turn every iteration, counts timeouts and stalls
//             per mode and never stops early; compare:fresh,async runs only the listed modes
//   reconnect one long-lived multiplexer runs conditional transactions in a loop while its connections are killed
//             with CLIENT KILL TYPE normal; <iterations> is the number of kills
//
// A round that takes longer than half the timeout is a stall. It prints "STALL pid=<pid>" once, so a stack can be taken
// meanwhile. A stall ends either in a RedisTimeoutException on UNWATCH or in a transaction that completes late.

using System.Diagnostics;
using System.Reflection;
using StackExchange.Redis;

var positional = args.Where(a => !a.StartsWith("--")).ToArray();
var named = args.Where(a => a.StartsWith("--")).Select(a => a[2..].Split('=', 2)).ToDictionary(p => p[0], p => p.Length > 1 ? p[1] : "");

var endpoint = positional.Length > 0 ? positional[0] : "127.0.0.1:6379";
var mode = positional.Length > 1 ? positional[1] : "fresh";
var parallel = positional.Length > 2 ? int.Parse(positional[2]) : 16;
var iterations = positional.Length > 3 ? int.Parse(positional[3]) : 500;
var timeout = named.TryGetValue("timeout", out var t) ? int.Parse(t) : 5000;

if (named.TryGetValue("min-threads", out var minThreads))
{
    ThreadPool.SetMinThreads(int.Parse(minThreads), int.Parse(minThreads));
}

ThreadPool.GetMinThreads(out var minWorker, out var minIo);
var library = typeof(ConnectionMultiplexer).Assembly
    .GetCustomAttribute<AssemblyInformationalVersionAttribute>()!.InformationalVersion;
Console.WriteLine($"StackExchange.Redis {library}, runtime {Environment.Version}, {Environment.OSVersion}, {Environment.ProcessorCount} logical CPUs");
Console.WriteLine($"endpoint {endpoint}, mode {mode}, parallel {parallel}, timeout {timeout} ms, min threads {minWorker}/{minIo}, pid {Environment.ProcessId}");

using var admin = ConnectionMultiplexer.Connect(Options(allowAdmin: true));
var adminDb = admin.GetDatabase();
var clock = Stopwatch.StartNew();

if (mode == "reconnect")
{
    return await ReconnectAsync();
}

string[] modes = mode switch
{
    "compare" => ["fresh", "async", "control", "ping", "delay", "nocond"],
    _ when mode.StartsWith("compare:") => mode["compare:".Length..].Split(','),
    _ => [mode],
};
var compare = mode.StartsWith("compare");
var stalls = modes.ToDictionary(m => m, _ => 0);
var wrong = modes.ToDictionary(m => m, _ => 0);
var failures = modes.ToDictionary(m => m, _ => 0);
var rounds = modes.ToDictionary(m => m, _ => 0);

for (var iteration = 1; iteration <= iterations; iteration++)
{
    foreach (var current in modes)
    {
        try
        {
            var elapsed = await RoundAsync(current);
            rounds[current]++;
            if (elapsed.TotalMilliseconds > timeout / 2.0)
            {
                stalls[current]++;
                Console.WriteLine($"{current}: stall at iteration {iteration}, completed after {elapsed.TotalMilliseconds:F0} ms");
            }
        }
        catch (RedisConnectionException ex)
        {
            // Opening thousands of short-lived connections can run a client machine out of local ports; such a
            // round is not counted, and the loop waits for the ports to be released
            Console.WriteLine($"{current}: connect failed at iteration {iteration} ({ex.Message.Split('.')[0]}), round not counted");
            await Task.Delay(TimeSpan.FromSeconds(60));
        }
        catch (WrongOutcomeException ex)
        {
            rounds[current]++;
            wrong[current]++;
            Console.WriteLine($"{current}: wrong outcome at iteration {iteration}: {ex.Message}");
        }
        catch (RedisTimeoutException ex)
        {
            rounds[current]++;
            failures[current]++;
            stalls[current]++;
            Console.WriteLine($"{current}: timeout at iteration {iteration} after {clock.Elapsed.TotalSeconds:F0} s");
            Console.WriteLine(ex.Message);
            if (!compare)
            {
                return 1;
            }
        }
    }

    if (iteration % 100 == 0)
    {
        Console.WriteLine($"iteration {iteration} ({clock.Elapsed.TotalSeconds:F0} s): " + Summary());
    }
}

Console.WriteLine($"done, {iterations} iterations ({clock.Elapsed.TotalSeconds:F0} s): " + Summary());
return failures.Values.Sum() == 0 ? 0 : 1;

string Summary() => string.Join(", ", modes.Select(m => $"{m} {stalls[m]} stalls ({failures[m]} timeouts), {wrong[m]} wrong outcomes in {rounds[m]} rounds"));

ConfigurationOptions Options(bool allowAdmin = false)
{
    var options = ConfigurationOptions.Parse(endpoint);
    options.SyncTimeout = timeout;
    options.AsyncTimeout = timeout;
    options.AllowAdmin = allowAdmin;
    return options;
}

async Task<RedisKey> NewKeyAsync()
{
    RedisKey key = $"repro:{Guid.NewGuid():N}";
    await adminDb.HashSetAsync(key, "state", "open");
    await adminDb.KeyExpireAsync(key, TimeSpan.FromMinutes(1));
    return key;
}

Task<bool> Transaction(IConnectionMultiplexer connection, RedisKey key, int i, bool conditional = true)
{
    var transaction = connection.GetDatabase().CreateTransaction();
    if (conditional)
    {
        transaction.AddCondition(Condition.HashExists(key, "state"));
        transaction.AddCondition(Condition.HashNotExists(key, "winner"));
    }
    _ = transaction.HashSetAsync(key, "winner", i);
    return transaction.ExecuteAsync();
}

async Task<TimeSpan> RoundAsync(string current)
{
    var key = await NewKeyAsync();
    var connections = new ConnectionMultiplexer?[parallel];
    var round = Stopwatch.StartNew();
    using var stallWatch = new Timer(_ => Console.WriteLine($"STALL pid={Environment.ProcessId} mode={current}"), null, timeout / 2, Timeout.Infinite);
    try
    {
        bool[] results;
        switch (current)
        {
            case "fresh":
                results = await Task.WhenAll(Enumerable.Range(0, parallel).Select(i => Task.Run(() =>
                {
                    connections[i] = ConnectionMultiplexer.Connect(Options());
                    return Transaction(connections[i]!, key, i);
                })));
                break;
            case "async":
                results = await Task.WhenAll(Enumerable.Range(0, parallel).Select(i => Task.Run(async () =>
                {
                    connections[i] = await ConnectionMultiplexer.ConnectAsync(Options());
                    return await Transaction(connections[i]!, key, i);
                })));
                break;
            case "control":
                await Task.WhenAll(Enumerable.Range(0, parallel).Select(i => Task.Run(() =>
                {
                    connections[i] = ConnectionMultiplexer.Connect(Options());
                })));
                results = await Task.WhenAll(Enumerable.Range(0, parallel).Select(i => Task.Run(() => Transaction(connections[i]!, key, i))));
                break;
            case "ping":
                results = await Task.WhenAll(Enumerable.Range(0, parallel).Select(i => Task.Run(async () =>
                {
                    connections[i] = await ConnectionMultiplexer.ConnectAsync(Options());
                    await connections[i]!.GetDatabase().PingAsync();
                    return await Transaction(connections[i]!, key, i);
                })));
                break;
            case "delay":
                results = await Task.WhenAll(Enumerable.Range(0, parallel).Select(i => Task.Run(async () =>
                {
                    connections[i] = await ConnectionMultiplexer.ConnectAsync(Options());
                    await Task.Delay(TimeSpan.FromSeconds(1));
                    return await Transaction(connections[i]!, key, i);
                })));
                break;
            case "nocond":
                results = await Task.WhenAll(Enumerable.Range(0, parallel).Select(i => Task.Run(async () =>
                {
                    connections[i] = await ConnectionMultiplexer.ConnectAsync(Options());
                    return await Transaction(connections[i]!, key, i, conditional: false);
                })));
                break;
            default:
                throw new ArgumentException($"Unknown mode: {current}");
        }

        var committed = results.Count(r => r);
        var expected = current == "nocond" ? parallel : 1;
        if (committed != expected)
        {
            var stored = await adminDb.HashGetAsync(key, "winner");
            throw new WrongOutcomeException($"{committed} of {parallel} reported committed, expected {expected}; winner field {(stored.IsNull ? "absent" : "present")}; round took {round.Elapsed.TotalMilliseconds:F0} ms");
        }

        // the delay mode waits 1 s by design
        return current == "delay" ? round.Elapsed - TimeSpan.FromSeconds(1) : round.Elapsed;
    }
    finally
    {
        foreach (var connection in connections)
        {
            connection?.Dispose();
        }
    }
}

async Task<int> ReconnectAsync()
{
    using var target = ConnectionMultiplexer.Connect(Options());
    var stop = new CancellationTokenSource();
    var timeouts = 0;
    var connectionErrors = 0;
    var committed = 0L;
    var killsWithTimeout = new HashSet<int>();
    var kill = 0;

    var loops = Enumerable.Range(0, parallel).Select(i => Task.Run(async () =>
    {
        while (!stop.IsCancellationRequested)
        {
            var key = await NewKeyAsync();
            try
            {
                if (await Transaction(target, key, i))
                {
                    Interlocked.Increment(ref committed);
                }
            }
            catch (RedisTimeoutException ex)
            {
                Interlocked.Increment(ref timeouts);
                lock (killsWithTimeout)
                {
                    killsWithTimeout.Add(Volatile.Read(ref kill));
                }
                Console.WriteLine($"timeout after kill {Volatile.Read(ref kill)}: {ex.Message}");
            }
            catch (RedisConnectionException)
            {
                Interlocked.Increment(ref connectionErrors);
            }
        }
    })).ToArray();

    var server = admin.GetServer(admin.GetEndPoints()[0]);
    for (kill = 1; kill <= iterations; kill++)
    {
        await Task.Delay(TimeSpan.FromSeconds(1));
        await server.ExecuteAsync("CLIENT", "KILL", "TYPE", "normal", "SKIPME", "yes");
        await Task.Delay(TimeSpan.FromSeconds(1));
        if (kill % 50 == 0)
        {
            Console.WriteLine($"kill {kill}: {committed} committed, {timeouts} timeouts, {connectionErrors} connection errors");
        }
    }

    stop.Cancel();
    await Task.WhenAll(loops);
    Console.WriteLine($"done, {iterations} kills ({clock.Elapsed.TotalSeconds:F0} s): {committed} committed, {timeouts} timeouts after {killsWithTimeout.Count} distinct kills, {connectionErrors} connection errors");
    return timeouts == 0 ? 0 : 1;
}

sealed class WrongOutcomeException(string message) : Exception(message);