experimental: transaction logging to find this cursed race condition

StackExchange.Redis.Tests/TestBase.cs

@@ -261,6 +261,7 @@ protected IServer GetAnyMaster(ConnectionMultiplexer muxer)
             muxer.ConnectionFailed += OnConnectionFailed;
             muxer.MessageFaulted += (msg, ex, origin) => Writer?.WriteLine($"Faulted from '{origin}': '{msg}' - '{(ex == null ? "(null)" : ex.Message)}'");
             muxer.Connecting += (e, t) => Writer.WriteLine($"Connecting to {Format.ToString(e)} as {t}");
+            muxer.TransactionLog += msg => Writer.WriteLine("tran: " + msg);
             muxer.Resurrecting += (e, t) => Writer.WriteLine($"Resurrecting {Format.ToString(e)} as {t}");
             muxer.Closing += complete => Writer.WriteLine(complete ? "Closed" : "Closing...");
             return muxer;

StackExchange.Redis.Tests/Transactions.cs

@@ -764,7 +764,7 @@ public async Task BasicTranWithSortedSetContainsCondition(bool demandKeyExists,
         [InlineData("", ComparisonType.GreaterThan, 0L, false)]
         public async Task BasicTranWithListLengthCondition(string value, ComparisonType type, long length, bool expectTranResult)
         {
-            using (var muxer = Create())
+            using (var muxer = Create(log: TextWriter.Null))
             {
                 RedisKey key = Me(), key2 = Me() + "2";
                 var db = muxer.GetDatabase();

StackExchange.Redis/StackExchange/Redis/RedisSubscriber.cs

@@ -254,7 +254,7 @@ internal string GetConnectionName(EndPoint endPoint, ConnectionType connectionTy
 
         internal event Action<string, Exception, string> MessageFaulted;
         internal event Action<bool> Closing;
-        internal event Action<string> PreTransactionExec;
+        internal event Action<string> PreTransactionExec, TransactionLog;
         internal event Action<EndPoint, ConnectionType> Connecting;
         internal event Action<EndPoint, ConnectionType> Resurrecting;
 
@@ -282,6 +282,11 @@ internal void OnPreTransactionExec(Message message)
         {
             PreTransactionExec?.Invoke(message.CommandAndKey);
         }
+
+        internal void OnTransactionLog(string message)
+        {
+            TransactionLog?.Invoke(message);
+        }
     }
 
     internal sealed class RedisSubscriber : RedisBase, ISubscriber

StackExchange.Redis/StackExchange/Redis/RedisTransaction.cs

@@ -224,6 +224,11 @@ public override int GetHashSlot(ServerSelectionStrategy serverSelectionStrategy)
             public IEnumerable<Message> GetMessages(PhysicalConnection connection)
             {
                 ResultBox lastBox = null;
+                var bridge = connection.BridgeCouldBeNull;
+                if (bridge == null) throw new ObjectDisposedException(connection.ToString());
+
+                bool explicitCheckForQueued = !bridge.ServerEndPoint.GetFeatures().ExecAbort;
+                var multiplexer = bridge.Multiplexer;
                 try
                 {
                     // Important: if the server supports EXECABORT, then we can check the pre-conditions (pause there),
@@ -234,15 +239,11 @@ public IEnumerable<Message> GetMessages(PhysicalConnection connection)
                     // up-version servers, pre-condition failures exit with UNWATCH; and on down-version servers pre-condition
                     // failures exit with DISCARD - but that's ok : both work fine
 
-                    var bridge = connection.BridgeCouldBeNull;
-                    if (bridge == null) throw new ObjectDisposedException(connection.ToString());
-
-                    bool explicitCheckForQueued = !bridge.ServerEndPoint.GetFeatures().ExecAbort;
-                    var multiplexer = bridge.Multiplexer;
-
                     // PART 1: issue the pre-conditions
                     if (!IsAborted && conditions.Length != 0)
                     {
+                        multiplexer.OnTransactionLog($"issueing conditions...");
+                        int cmdCount = 0;
                         for (int i = 0; i < conditions.Length; i++)
                         {
                             // need to have locked them before sending them
@@ -255,11 +256,15 @@ public IEnumerable<Message> GetMessages(PhysicalConnection connection)
                             {
                                 msg.SetNoRedirect(); // need to keep them in the current context only
                                 yield return msg;
+                                multiplexer.OnTransactionLog($"issuing {msg.CommandAndKey}");
+                                cmdCount++;
                             }
                         }
+                        multiplexer.OnTransactionLog($"issued {conditions.Length} conditions ({cmdCount} commands)");
 
                         if (!explicitCheckForQueued && lastBox != null)
                         {
+                            multiplexer.OnTransactionLog($"checking conditions in the *early* path");
                             // need to get those sent ASAP; if they are stuck in the buffers, we die
                             multiplexer.Trace("Flushing and waiting for precondition responses");
                             connection.FlushAsync().Wait();
@@ -267,11 +272,15 @@ public IEnumerable<Message> GetMessages(PhysicalConnection connection)
                             {
                                 if (!AreAllConditionsSatisfied(multiplexer))
                                     command = RedisCommand.UNWATCH; // somebody isn't happy
+
+                                multiplexer.OnTransactionLog($"after condition check, we are {command}");
                             }
                             else
                             { // timeout running pre-conditions
                                 multiplexer.Trace("Timeout checking preconditions");
                                 command = RedisCommand.UNWATCH;
+
+                                multiplexer.OnTransactionLog($"timeout waiting for conditions, we are {command}");
                             }
                             Monitor.Exit(lastBox);
                             lastBox = null;
@@ -283,12 +292,13 @@ public IEnumerable<Message> GetMessages(PhysicalConnection connection)
                     {
                         multiplexer.Trace("Begining transaction");
                         yield return Message.Create(-1, CommandFlags.None, RedisCommand.MULTI);
+                        multiplexer.OnTransactionLog($"issued MULTI");
                     }
 
                     // PART 3: issue the commands
                     if (!IsAborted && InnerOperations.Length != 0)
                     {
-                        multiplexer.Trace("Issuing transaction operations");
+                        multiplexer.Trace("Issuing operations...");
 
                         foreach (var op in InnerOperations)
                         {
@@ -304,10 +314,15 @@ public IEnumerable<Message> GetMessages(PhysicalConnection connection)
                                 }
                             }
                             yield return op;
+                            multiplexer.OnTransactionLog($"issued {op.CommandAndKey}");
                         }
+                        multiplexer.OnTransactionLog($"issued {InnerOperations.Length} operations");
+
 
                         if (explicitCheckForQueued && lastBox != null)
                         {
+                            multiplexer.OnTransactionLog($"checking conditions in the *late* path");
+
                             multiplexer.Trace("Flushing and waiting for precondition+queued responses");
                             connection.FlushAsync().Wait(); // make sure they get sent, so we can check for QUEUED (and the pre-conditions if necessary)
                             if (Monitor.Wait(lastBox, multiplexer.TimeoutMilliseconds))
@@ -323,17 +338,20 @@ public IEnumerable<Message> GetMessages(PhysicalConnection connection)
                                         if (!op.WasQueued)
                                         {
                                             multiplexer.Trace("Aborting: operation was not queued: " + op.Command);
+                                            multiplexer.OnTransactionLog($"command was not issued: {op.CommandAndKey}");
                                             command = RedisCommand.DISCARD;
                                             break;
                                         }
                                     }
                                 }
                                 multiplexer.Trace("Confirmed: QUEUED x " + InnerOperations.Length);
+                                multiplexer.OnTransactionLog($"after condition check, we are {command}");
                             }
                             else
                             {
                                 multiplexer.Trace("Aborting: timeout checking queued messages");
                                 command = RedisCommand.DISCARD;
+                                multiplexer.OnTransactionLog($"timeout waiting for conditions, we are {command}");
                             }
                             Monitor.Exit(lastBox);
                             lastBox = null;
@@ -346,8 +364,8 @@ public IEnumerable<Message> GetMessages(PhysicalConnection connection)
                 }
                 if (IsAborted)
                 {
+                    multiplexer.OnTransactionLog($"aborting {InnerOperations.Length} wrapped commands...");
                     connection.Trace("Aborting: canceling wrapped messages");
-                    var bridge = connection.BridgeCouldBeNull;
                     foreach (var op in InnerOperations)
                     {
                         op.Wrapped.Cancel();
@@ -355,6 +373,7 @@ public IEnumerable<Message> GetMessages(PhysicalConnection connection)
                     }
                 }
                 connection.Trace("End of transaction: " + Command);
+                multiplexer.OnTransactionLog($"issuing {this.Command}");
                 yield return this; // acts as either an EXEC or an UNWATCH, depending on "aborted"
             }