Ensure that _activeMessage is always cleared, including if an excepti…

…on is thrown while trying to send the message or flush the connection. Leaving _activeMessage set causes WriteMessageInsideLock to always return a "NoConnectionAvailable" error indefinitely (#1374)

Co-authored-by: Nick Craver <craver@stackoverflow.com>

src/StackExchange.Redis/PhysicalBridge.cs

@@ -737,12 +737,15 @@ internal WriteResult WriteMessageTakingWriteLockSync(PhysicalConnection physical
 #pragma warning restore CS0618
                 }
 
-                UnmarkActiveMessage(message);
                 physical.SetIdle();
                 return result;
             }
             catch (Exception ex) { return HandleWriteException(message, ex); }
-            finally { token.Dispose(); }
+            finally
+            {
+                UnmarkActiveMessage(message);
+                token.Dispose();
+            }
 
         }
 
@@ -887,7 +890,6 @@ private void ProcessBacklog()
                             }
 
                             _backlogStatus = BacklogStatus.MarkingInactive;
-                            UnmarkActiveMessage(message);
                             if (result != WriteResult.Success)
                             {
                                 _backlogStatus = BacklogStatus.RecordingWriteFailure;
@@ -901,6 +903,10 @@ private void ProcessBacklog()
                         _backlogStatus = BacklogStatus.RecordingFault;
                         HandleWriteException(message, ex);
                     }
+                    finally
+                    {
+                        UnmarkActiveMessage(message);
+                    }
                 }
                 _backlogStatus = BacklogStatus.SettingIdle;
                 physical.SetIdle();
@@ -985,21 +991,25 @@ internal ValueTask<WriteResult> WriteMessageTakingWriteLockAsync(PhysicalConnect
 
                     result = flush.Result; // we know it was completed, this is fine
                 }
-
-                UnmarkActiveMessage(message);
+
                 physical.SetIdle();
 
                 return new ValueTask<WriteResult>(result);
             }
             catch (Exception ex) { return new ValueTask<WriteResult>(HandleWriteException(message, ex)); }
             finally
             {
-                if (releaseLock & token.Success)
+                if (token.Success)
                 {
+                    UnmarkActiveMessage(message);
+
+                    if (releaseLock)
+                    {
 #if DEBUG
-                    RecordLockDuration(lockTaken);
+                        RecordLockDuration(lockTaken);
 #endif
-                    token.Dispose();
+                        token.Dispose();
+                    }
                 }
             }
         }
@@ -1029,8 +1039,7 @@ private async ValueTask<WriteResult> WriteMessageTakingWriteLockAsync_Awaited(Va
                     {
                         result = await physical.FlushAsync(false).ForAwait();
                     }
-
-                    UnmarkActiveMessage(message);
+
                     physical.SetIdle();
 
 #if DEBUG
@@ -1043,6 +1052,10 @@ private async ValueTask<WriteResult> WriteMessageTakingWriteLockAsync_Awaited(Va
             {
                 return HandleWriteException(message, ex);
             }
+            finally
+            {
+                UnmarkActiveMessage(message);
+            }
         }
 
         private async ValueTask<WriteResult> CompleteWriteAndReleaseLockAsync(LockToken lockToken, ValueTask<WriteResult> flush, Message message, int lockTaken)
@@ -1052,7 +1065,6 @@ private async ValueTask<WriteResult> CompleteWriteAndReleaseLockAsync(LockToken
                 try
                 {
                     var result = await flush.ForAwait();
-                    UnmarkActiveMessage(message);
                     physical.SetIdle();
                     return result;
                 }