Only take app lock out at beginning of transaction (#983)

src/TriggerBinding/SqlTableChangeMonitor.cs

@@ -12,6 +12,7 @@
 using Microsoft.Azure.WebJobs.Extensions.Sql.Telemetry;
 using static Microsoft.Azure.WebJobs.Extensions.Sql.Telemetry.Telemetry;
 using static Microsoft.Azure.WebJobs.Extensions.Sql.SqlTriggerConstants;
+using static Microsoft.Azure.WebJobs.Extensions.Sql.SqlTriggerUtils;
 using Microsoft.Azure.WebJobs.Host.Executors;
 using Microsoft.Data.SqlClient;
 using Microsoft.Extensions.Logging;
@@ -285,6 +286,8 @@ private async Task GetTableChangesAsync(SqlConnection connection, CancellationTo
                 {
                     try
                     {
+                        await AcquireAppLockAsync(connection, transaction, this._logger, token);
+
                         // Update the version number stored in the global state table if necessary before using it.
                         using (SqlCommand updateTablesPreInvocationCommand = this.BuildUpdateTablesPreInvocation(connection, transaction))
                         {
@@ -525,6 +528,8 @@ private async Task RenewLeasesAsync(SqlConnection connection, CancellationToken
                 {
                     try
                     {
+                        await AcquireAppLockAsync(connection, transaction, this._logger, token);
+
                         SqlCommand renewLeasesCommand = this.BuildRenewLeasesCommand(connection, transaction);
                         if (renewLeasesCommand != null)
                         {
@@ -634,6 +639,8 @@ private async Task ReleaseLeasesAsync(SqlConnection connection, CancellationToke
                     {
                         try
                         {
+                            await AcquireAppLockAsync(connection, transaction, this._logger, token);
+
                             // Release the leases held on "_rowsToRelease".
                             using (SqlCommand releaseLeasesCommand = this.BuildReleaseLeasesCommand(connection, transaction))
                             {
@@ -789,8 +796,6 @@ private static SqlChangeOperation GetChangeOperation(IReadOnlyDictionary<string,
         private SqlCommand BuildUpdateTablesPreInvocation(SqlConnection connection, SqlTransaction transaction)
         {
             string updateTablesPreInvocationQuery = $@"
-                {AppLockStatements}
-
                 DECLARE @min_valid_version bigint;
                 SET @min_valid_version = CHANGE_TRACKING_MIN_VALID_VERSION({this._userTableId});
 
@@ -834,8 +839,6 @@ private SqlCommand BuildGetChangesCommand(SqlConnection connection, SqlTransacti
             // up regardless since we know it should be processed - no need to check the change version.
             // Once a row is successfully processed the LeaseExpirationTime column is set to NULL.
             string getChangesQuery = $@"
-                {AppLockStatements}
-
                 DECLARE @last_sync_version bigint;
                 SELECT @last_sync_version = LastSyncVersion
                 FROM {GlobalStateTableName}
@@ -882,8 +885,6 @@ private async Task<string> GetLeaseLockedOrMaxAttemptRowCountMessage(SqlConnecti
             //  * NULL LeaseExpirationTime OR LeaseExpirationTime <= Current Time
             //  * No attempts remaining (Attempt count = Max attempts)
             string getLeaseLockedOrMaxAttemptRowCountQuery = $@"
-                {AppLockStatements}
-
                 DECLARE @last_sync_version bigint;
                 SELECT @last_sync_version = LastSyncVersion
                 FROM {GlobalStateTableName}
@@ -948,8 +949,6 @@ private SqlCommand BuildAcquireLeasesCommand(SqlConnection connection, SqlTransa
             const string rowDataParameter = "@rowData";
             // Create the merge query that will either update the rows that already exist or insert a new one if it doesn't exist
             string query = $@"
-                    {AppLockStatements}
-
                     WITH {acquireLeasesCte} AS ( SELECT * FROM OPENJSON(@rowData) WITH ({string.Join(",", cteColumnDefinitions)}) )
                     MERGE INTO {this._bracketedLeasesTableName}
                         AS ExistingData
@@ -989,8 +988,6 @@ private SqlCommand BuildRenewLeasesCommand(SqlConnection connection, SqlTransact
                 return null;
             }
             string renewLeasesQuery = $@"
-                {AppLockStatements}
-
                 UPDATE {this._bracketedLeasesTableName}
                 SET {LeasesTableLeaseExpirationTimeColumnName} = DATEADD(second, {LeaseIntervalInSeconds}, SYSDATETIME())
                 WHERE {matchCondition};
@@ -1021,9 +1018,7 @@ private SqlCommand BuildReleaseLeasesCommand(SqlConnection connection, SqlTransa
             const string rowDataParameter = "@rowData";
 
             string releaseLeasesQuery =
-$@"{AppLockStatements}
-
-WITH {releaseLeasesCte} AS ( SELECT * FROM OPENJSON(@rowData) WITH ({string.Join(",", cteColumnDefinitions)}) )
+$@"WITH {releaseLeasesCte} AS ( SELECT * FROM OPENJSON(@rowData) WITH ({string.Join(",", cteColumnDefinitions)}) )
 UPDATE {this._bracketedLeasesTableName}
 SET
     {LeasesTableChangeVersionColumnName} = cte.{SysChangeVersionColumnName},
@@ -1053,8 +1048,6 @@ private SqlCommand BuildUpdateTablesPostInvocation(SqlConnection connection, Sql
             string leasesTableJoinCondition = string.Join(" AND ", this._primaryKeyColumns.Select(col => $"c.{col.name.AsBracketQuotedString()} = l.{col.name.AsBracketQuotedString()}"));
 
             string updateTablesPostInvocationQuery = $@"
-                {AppLockStatements}
-
                 DECLARE @current_last_sync_version bigint;
                 SELECT @current_last_sync_version = LastSyncVersion
                 FROM {GlobalStateTableName}

src/TriggerBinding/SqlTriggerListener.cs

@@ -143,6 +143,8 @@ public async Task StartAsync(CancellationToken cancellationToken)
                     long createdSchemaDurationMs = 0L, createGlobalStateTableDurationMs = 0L, insertGlobalStateTableRowDurationMs = 0L, createLeasesTableDurationMs = 0L;
                     using (SqlTransaction transaction = connection.BeginTransaction(System.Data.IsolationLevel.RepeatableRead))
                     {
+                        await AcquireAppLockAsync(connection, transaction, this._logger, cancellationToken);
+
                         createdSchemaDurationMs = await this.CreateSchemaAsync(connection, transaction, cancellationToken);
                         createGlobalStateTableDurationMs = await this.CreateGlobalStateTableAsync(connection, transaction, cancellationToken);
                         insertGlobalStateTableRowDurationMs = await this.InsertGlobalStateTableRowAsync(connection, transaction, userTableId, cancellationToken);
@@ -283,8 +285,6 @@ FROM sys.columns AS c
         private async Task<long> CreateSchemaAsync(SqlConnection connection, SqlTransaction transaction, CancellationToken cancellationToken)
         {
             string createSchemaQuery = $@"
-                {AppLockStatements}
-
                 IF SCHEMA_ID(N'{SchemaName}') IS NULL
                     EXEC ('CREATE SCHEMA {SchemaName}');
             ";
@@ -328,8 +328,6 @@ IF SCHEMA_ID(N'{SchemaName}') IS NULL
         private async Task<long> CreateGlobalStateTableAsync(SqlConnection connection, SqlTransaction transaction, CancellationToken cancellationToken)
         {
             string createGlobalStateTableQuery = $@"
-                {AppLockStatements}
-
                 IF OBJECT_ID(N'{GlobalStateTableName}', 'U') IS NULL
                     CREATE TABLE {GlobalStateTableName} (
                         UserFunctionID char(16) NOT NULL,
@@ -401,7 +399,6 @@ private async Task<long> InsertGlobalStateTableRowAsync(SqlConnection connection
             }
 
             string insertRowGlobalStateTableQuery = $@"
-                {AppLockStatements}
                 -- For back compatibility copy the lastSyncVersion from _hostIdFunctionId if it exists.
                 IF NOT EXISTS (
                     SELECT * FROM {GlobalStateTableName}
@@ -456,8 +453,6 @@ private async Task<long> CreateLeasesTableAsync(
             // we're actually using the WEBSITE_SITE_NAME one (e.g. leasesTableName is different)
             bool shouldMigrateOldLeasesTable = !string.IsNullOrEmpty(oldLeasesTableName) && oldLeasesTableName != leasesTableName;
             string createLeasesTableQuery = shouldMigrateOldLeasesTable ? $@"
-                {AppLockStatements}
-
                 IF OBJECT_ID(N'{leasesTableName}', 'U') IS NULL
                 BEGIN
                     CREATE TABLE {leasesTableName} (
@@ -479,8 +474,6 @@ INSERT INTO {leasesTableName}
                 End
             " :
             $@"
-                {AppLockStatements}
-
                 IF OBJECT_ID(N'{leasesTableName}', 'U') IS NULL
                     CREATE TABLE {leasesTableName} (
                         {primaryKeysWithTypes},

src/TriggerBinding/SqlTriggerMetricsProvider.cs

@@ -63,6 +63,8 @@ private async Task<long> GetUnprocessedChangeCountAsync()
                     {
                         try
                         {
+                            await AcquireAppLockAsync(connection, transaction, this._logger, CancellationToken.None);
+
                             using (SqlCommand getUnprocessedChangesCommand = this.BuildGetUnprocessedChangesCommand(connection, transaction, primaryKeyColumns, userTableId))
                             {
                                 var commandSw = Stopwatch.StartNew();
@@ -102,8 +104,6 @@ private SqlCommand BuildGetUnprocessedChangesCommand(SqlConnection connection, S
             string leasesTableJoinCondition = string.Join(" AND ", primaryKeyColumns.Select(col => $"c.{col.name.AsBracketQuotedString()} = l.{col.name.AsBracketQuotedString()}"));
             string bracketedLeasesTableName = GetBracketedLeasesTableName(this._userDefinedLeasesTableName, this._userFunctionId, userTableId);
             string getUnprocessedChangesQuery = $@"
-                {AppLockStatements}
-
                 DECLARE @last_sync_version bigint;
                 SELECT @last_sync_version = LastSyncVersion
                 FROM {GlobalStateTableName}

src/TriggerBinding/SqlTriggerUtils.cs

@@ -16,6 +16,24 @@ namespace Microsoft.Azure.WebJobs.Extensions.Sql
     public static class SqlTriggerUtils
     {
 
+        /// <summary>
+        /// Acquires an exclusive application lock on the transaction to prevent deadlocks.
+        /// This should be called once at the beginning of each transaction rather than
+        /// being included in every individual query, since the lock is transaction-scoped
+        /// and subsequent acquisitions within the same transaction are no-ops.
+        /// </summary>
+        /// <param name="connection">The SQL connection</param>
+        /// <param name="transaction">The transaction to acquire the lock on</param>
+        /// <param name="logger">Logger for logging the command</param>
+        /// <param name="cancellationToken">Cancellation token</param>
+        internal static async Task AcquireAppLockAsync(SqlConnection connection, SqlTransaction transaction, ILogger logger, CancellationToken cancellationToken)
+        {
+            using (var command = new SqlCommand(AppLockStatements, connection, transaction))
+            {
+                await command.ExecuteNonQueryAsyncWithLogging(logger, cancellationToken, true);
+            }
+        }
+
         /// <summary>
         /// Gets the names and types of primary key columns of the user table.
         /// </summary>