ensure that we actually update the last-write-time so we don't constantly...

ensure that we actually update the last-write-time so we don't constantly heartbeat; pre-emptively set last-write-time *before* sending heartbeat, to prevent over-runs; track physical's write-state

StackExchange.Redis/PhysicalBridge.cs

@@ -263,6 +263,7 @@ internal void KeepAlive()
                 msg.SetInternalCall();
                 Multiplexer.Trace("Enqueue: " + msg);
                 Multiplexer.OnInfoMessage($"heartbeat '{msg.CommandAndKey}' on '{PhysicalName}' (v{features.Version})");
+                physical?.UpdateLastWriteTime(); // pre-emptively
                 var result = TryWrite(msg, ServerEndPoint.IsSlave);
 
                 if (result != WriteResult.Success)
@@ -375,6 +376,7 @@ private void AbandonPendingBacklog(Exception ex)
         internal void OnFullyEstablished(PhysicalConnection connection)
         {
             Trace("OnFullyEstablished");
+            connection?.SetIdle();
             if (physical == connection && !isDisposed && ChangeState(State.ConnectedEstablishing, State.ConnectedEstablished))
             {
                 reportNextFailure = reconfigureNextFailure = true;
@@ -562,6 +564,7 @@ internal WriteResult WriteMessageTakingWriteLock(PhysicalConnection physical, Me
                 {
                     Multiplexer?.OnInfoMessage("reentrant call to WriteMessageTakingWriteLock for " + message.CommandAndKey);
                 }
+                physical.SetWriting();
                 var messageIsSent = false;
                 if (message is IMultiMessage)
                 {
@@ -593,7 +596,8 @@ internal WriteResult WriteMessageTakingWriteLock(PhysicalConnection physical, Me
                 {
                     result = WriteMessageToServerInsideWriteLock(physical, message);
                 }
-                result = physical.WakeWriterAndCheckForThrottle();
+                result = physical.FlushSync();
+                physical.SetIdle();
             }
             finally
             {

StackExchange.Redis/PhysicalConnection.cs

@@ -269,17 +269,20 @@ public void Dispose()
         private async Task AwaitedFlush(ValueTask<FlushResult> flush)
         {
             await flush;
-            Interlocked.Exchange(ref lastWriteTickCount, Environment.TickCount);
+            _writeStatus = WriteStatus.Flushed;
+            UpdateLastWriteTime();
         }
-
+        internal void UpdateLastWriteTime() => Interlocked.Exchange(ref lastWriteTickCount, Environment.TickCount);
         public Task FlushAsync()
         {
             var tmp = _ioPipe?.Output;
             if (tmp != null)
             {
+                _writeStatus = WriteStatus.Flushing;
                 var flush = tmp.FlushAsync();
                 if (!flush.IsCompletedSuccessfully) return AwaitedFlush(flush);
-                Interlocked.Exchange(ref lastWriteTickCount, Environment.TickCount);
+                _writeStatus = WriteStatus.Flushed;
+                UpdateLastWriteTime();
             }
             return Task.CompletedTask;
         }
@@ -342,6 +345,7 @@ public Task FlushAsync()
                         if (bridge != null)
                         {
                             exMessage.Append(" on ").Append(Format.ToString(bridge.ServerEndPoint?.EndPoint)).Append("/").Append(connectionType)
+                                .Append(", ").Append(_writeStatus)
                                 .Append(", last: ").Append(bridge.LastCommand);
 
                             data.Add(Tuple.Create("FailureType", failureType.ToString()));
@@ -411,11 +415,21 @@ void add(string lk, string sk, string v)
             Shutdown();
         }
 
-        public override string ToString()
+        internal void SetIdle() => _writeStatus = WriteStatus.Idle;
+        internal void SetWriting() => _writeStatus = WriteStatus.Writing;
+
+        private WriteStatus _writeStatus;
+        private enum WriteStatus
         {
-            return physicalName;
+            Initializing,
+            Idle,
+            Writing,
+            Flushing,
+            Flushed,
         }
 
+        public override string ToString() => $"{physicalName} ({_writeStatus})";
+
         internal static void IdentifyFailureType(Exception exception, ref ConnectionFailureType failureType)
         {
             if (exception != null && failureType == ConnectionFailureType.InternalFailure)
@@ -799,15 +813,20 @@ internal static int WriteRaw(Span<byte> span, long value, bool withLengthPrefix
             return WriteCrlf(span, offset);
         }
 
-        internal WriteResult WakeWriterAndCheckForThrottle()
+        internal WriteResult FlushSync(bool throwOnFailure = false)
         {
+            var tmp = _ioPipe?.Output;
+            if (tmp == null) return WriteResult.NoConnectionAvailable;
             try
             {
-                var flush = _ioPipe.Output.FlushAsync();
+                _writeStatus = WriteStatus.Flushing;
+                var flush = tmp.FlushAsync();
                 if (!flush.IsCompletedSuccessfully) flush.AsTask().Wait();
+                _writeStatus = WriteStatus.Flushed;
+                UpdateLastWriteTime();
                 return WriteResult.Success;
             }
-            catch (ConnectionResetException ex)
+            catch (ConnectionResetException ex) when (!throwOnFailure)
             {
                 RecordConnectionFailed(ConnectionFailureType.SocketClosed, ex);
                 return WriteResult.WriteFailure;

StackExchange.Redis/RedisTransaction.cs

@@ -274,7 +274,8 @@ public IEnumerable<Message> GetMessages(PhysicalConnection connection)
                                 sb.AppendLine("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();
+                                connection.FlushSync(true); // make sure they get sent, so we can check for QUEUED (and the pre-conditions if necessary)
+
                                 if (Monitor.Wait(lastBox, multiplexer.TimeoutMilliseconds))
                                 {
                                     if (!AreAllConditionsSatisfied(multiplexer))
@@ -330,7 +331,7 @@ public IEnumerable<Message> GetMessages(PhysicalConnection connection)
                                 sb.AppendLine("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)
+                                connection.FlushSync(true); // make sure they get sent, so we can check for QUEUED (and the pre-conditions if necessary)
                                 if (Monitor.Wait(lastBox, multiplexer.TimeoutMilliseconds))
                                 {
                                     if (!AreAllConditionsSatisfied(multiplexer))