Let fatal exceptions when getting locked row count bubble up (#971)

* Let fatal exceptions when getting locked row count bubble up

* Check for deadlock

src/SqlBindingUtilities.cs

@@ -262,6 +262,19 @@ internal static bool IsFatalSqlException(this Exception e)
                 || (lowerMessage.Contains("connection") && (lowerMessage.Contains("broken") || lowerMessage.Contains("closed")));
         }
 
+        /// <summary>
+        /// Whether the exception is a SqlClient deadlock exception (Error Number 1205)
+        /// </summary>
+        /// <param name="e"></param>
+        /// <returns></returns>
+        internal static bool IsDeadlockException(this Exception e)
+        {
+            // See https://learn.microsoft.com/sql/relational-databases/errors-events/mssqlserver-1205-database-engine-error
+            // Transaction (Process ID %d) was deadlocked on %.*ls resources with another process and has been chosen as the deadlock victim. Rerun the transaction.
+            return (e as SqlException)?.Number == 1205
+                || (e.InnerException as SqlException)?.Number == 1205;
+        }
+
         /// <summary>
         /// Checks whether the connection state is currently Broken or Closed
         /// </summary>

src/TriggerBinding/SqlTableChangeMonitor.cs

@@ -242,6 +242,10 @@ private async Task RunChangeConsumptionLoopAsync()
                             this._logger.LogError($"Fatal SQL Client exception processing changes. Will attempt to reestablish connection in {this._pollingIntervalInMs}ms. Exception = {e.Message}");
                             forceReconnect = true;
                         }
+                        catch (Exception e) when (e.IsDeadlockException())
+                        {
+                            // Deadlocks aren't fatal and don't need a reconnection so just let the loop try again after the normal delay
+                        }
                         await Task.Delay(TimeSpan.FromMilliseconds(this._pollingIntervalInMs), token);
                     }
                 }
@@ -860,8 +864,17 @@ LEFT OUTER JOIN {this._bracketedLeasesTableName} AS l ON {leasesTableJoinConditi
             {
                 this._logger.LogError($"Failed to query count of lease locked changes for table '{this._userTable.FullName}' due to exception: {ex.GetType()}. Exception message: {ex.Message}");
                 TelemetryInstance.TrackException(TelemetryErrorName.GetLeaseLockedRowCount, ex, null, new Dictionary<TelemetryMeasureName, double>() { { TelemetryMeasureName.GetLockedRowCountDurationMs, getLockedRowCountDurationMs } });
-                // This is currently only used for debugging, so return a -1 instead of throwing since it isn't necessary to get the value
-                leaseLockedRowsCount = -1;
+                // This is currently only used for debugging, so ignore the exception if we can. If the error is a fatal one though then the connection or transaction will be
+                // unusable so we have to let this bubble up so we can attempt to reconnect
+                if (ex.IsFatalSqlException() || ex.IsDeadlockException() || connection.IsBrokenOrClosed())
+                {
+                    throw;
+                }
+                else
+                {
+                    // If it's non-fatal though return a -1 instead of throwing since it isn't necessary to get the value
+                    leaseLockedRowsCount = -1;
+                }
             }
             return leaseLockedRowsCount;
         }