diff --git a/src/TriggerBinding/SqlTableChangeMonitor.cs b/src/TriggerBinding/SqlTableChangeMonitor.cs index 7295ca9b..4f739fcc 100644 --- a/src/TriggerBinding/SqlTableChangeMonitor.cs +++ b/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 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} diff --git a/src/TriggerBinding/SqlTriggerListener.cs b/src/TriggerBinding/SqlTriggerListener.cs index 52e376ad..eaf4aaf1 100644 --- a/src/TriggerBinding/SqlTriggerListener.cs +++ b/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 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 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 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 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}, diff --git a/src/TriggerBinding/SqlTriggerMetricsProvider.cs b/src/TriggerBinding/SqlTriggerMetricsProvider.cs index 4411e31d..7ce880b5 100644 --- a/src/TriggerBinding/SqlTriggerMetricsProvider.cs +++ b/src/TriggerBinding/SqlTriggerMetricsProvider.cs @@ -63,6 +63,8 @@ private async Task 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} diff --git a/src/TriggerBinding/SqlTriggerUtils.cs b/src/TriggerBinding/SqlTriggerUtils.cs index ef3ca7a1..59f6305a 100644 --- a/src/TriggerBinding/SqlTriggerUtils.cs +++ b/src/TriggerBinding/SqlTriggerUtils.cs @@ -16,6 +16,24 @@ namespace Microsoft.Azure.WebJobs.Extensions.Sql public static class SqlTriggerUtils { + /// + /// 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. + /// + /// The SQL connection + /// The transaction to acquire the lock on + /// Logger for logging the command + /// Cancellation token + 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); + } + } + /// /// Gets the names and types of primary key columns of the user table. ///