< Summary - AsyncResponse (Release / net8.0+net10.0 / unit+integration)

Information
Class: AsyncResponse.DurableFlows.SqlServer.SqlServerFlowStateStore
Assembly: AsyncResponse.DurableFlows.SqlServer
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/DurableFlows/AsyncResponse.DurableFlows.SqlServer/SqlServerDurableFlows.cs
Line coverage
99%
Covered lines: 171
Uncovered lines: 1
Coverable lines: 172
Total lines: 344
Line coverage: 99.4%
Branch coverage
81%
Covered branches: 13
Total branches: 16
Branch coverage: 81.2%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor(...)100%11100%
LoadAsync()100%11100%
TryCreateAsync()100%22100%
TryUpdateAsync()100%22100%
TryAcquireLeaseAsync(...)100%11100%
TryRenewLeaseAsync(...)100%11100%
ReleaseLeaseAsync()100%11100%
TryDeleteAsync()100%11100%
PruneExpiredAsync()100%11100%
EnsureCreatedAsync()83.33%6697.96%
UpdateLeaseAsync()75%4495%
OpenConnectionAsync()50%2271.43%
AddMilliseconds(...)100%11100%
get_Table()100%11100%
get_IndexName()100%11100%
Quote(...)100%11100%

File(s)

/home/runner/work/AsyncResponse/AsyncResponse/src/DurableFlows/AsyncResponse.DurableFlows.SqlServer/SqlServerDurableFlows.cs

#LineLine coverage
 1using AsyncResponse;
 2using AsyncResponse.DurableFlows.Internal;
 3using AsyncResponse.DurableFlows.SqlServer;
 4using Microsoft.Data.SqlClient;
 5using Microsoft.Extensions.DependencyInjection.Extensions;
 6using Microsoft.Extensions.Options;
 7
 8namespace Microsoft.Extensions.DependencyInjection
 9{
 10    /// <summary>DI registration for the SQL Server durable-flow state store.</summary>
 11    public static class SqlServerDurableFlowServiceCollectionExtensions
 12    {
 13        /// <summary>Stores durable-flow state in SQL Server.</summary>
 14        public static AsyncResponseRegistrationBuilder WithSqlServerDurableFlows(
 15            this AsyncResponseRegistrationBuilder builder,
 16            Action<SqlServerDurableFlowOptions>? configure = null)
 17        {
 18            // Singleton on purpose: schema provisioning is cached per store instance, and the
 19            // executor resolves the store from a fresh scope per flow execution — a scoped store
 20            // would re-run EnsureCreated's DDL round-trip on every run.
 21            builder.Services.TryAddSingleton<SqlServerFlowStateStore>();
 22            return builder.WithDurableFlows<SqlServerFlowStateStore, SqlServerDurableFlowOptions>(configure);
 23        }
 24    }
 25}
 26
 27namespace AsyncResponse.DurableFlows.SqlServer
 28{
 29/// <summary>Options for the SQL Server durable-flow state store.</summary>
 30public sealed class SqlServerDurableFlowOptions : DurableFlowOptions
 31{
 32    /// <summary>SQL Server connection string. Required.</summary>
 33    public string? ConnectionString { get; set; }
 34
 35    /// <summary>Database schema that contains the durable-flow table. Default: <c>dbo</c>.</summary>
 36    public string SchemaName { get; set; } = "dbo";
 37
 38    /// <summary>Table storing one durable-flow ledger row per flow id.</summary>
 39    public string TableName { get; set; } = "asyncresponse_flow_state";
 40
 41    /// <summary>Creates the schema, table, and expiry index on first use.</summary>
 42    public bool AutoCreateSchema { get; set; } = true;
 43
 44    /// <summary>
 45    /// How often <see cref="SqlServerFlowStateStore.TryCreateAsync"/> opportunistically deletes one
 46    /// bounded batch (1000 rows) of expired rows (loads already treat expired state as absent;
 47    /// pruning bounds table growth). Zero or negative prunes on every save. Default: 5 minutes.
 48    /// </summary>
 49    public TimeSpan PruneInterval { get; set; } = TimeSpan.FromMinutes(5);
 50
 51    /// <summary>
 52    /// Maximum serialized flow-state size in bytes accepted by writes; oversized ledgers fail fast
 53    /// with an actionable error instead of an opaque provider error. Default: <c>null</c>
 54    /// (unlimited — <c>nvarchar(max)</c> is effectively unbounded), settable as an operator budget.
 55    /// </summary>
 56    public long? MaxStateBytes { get; set; }
 57
 58    /// <summary>Validates option values and throws on misconfiguration.</summary>
 59    public void Validate()
 60    {
 61        if (string.IsNullOrWhiteSpace(ConnectionString))
 62            throw new InvalidOperationException($"{nameof(SqlServerDurableFlowOptions)}.{nameof(ConnectionString)} must 
 63
 64        DurableFlowStoreShared.ValidateIdentifier(SchemaName, $"{nameof(SqlServerDurableFlowOptions)}.{nameof(SchemaName
 65        DurableFlowStoreShared.ValidateIdentifier(TableName, $"{nameof(SqlServerDurableFlowOptions)}.{nameof(TableName)}
 66        if (MaxStateBytes is <= 0)
 67            throw new InvalidOperationException($"{nameof(SqlServerDurableFlowOptions)}.{nameof(MaxStateBytes)} must be 
 68    }
 69}
 70
 71/// <summary>SQL Server implementation of <see cref="IFlowStateStore"/>.</summary>
 72public sealed class SqlServerFlowStateStore : IFlowStateStore
 73{
 74    private const int PruneBatchSize = 1000;
 75
 76    private readonly SqlServerDurableFlowOptions _options;
 377    private readonly SemaphoreSlim _ensureGate = new(1, 1);
 78    private long _lastPruneTicks;
 79    private bool _created;
 80
 381    public SqlServerFlowStateStore(IOptions<SqlServerDurableFlowOptions> options)
 82    {
 383        _options = options.Value;
 384        _options.Validate();
 385    }
 86
 87    public async Task<FlowState?> LoadAsync(string flowId, CancellationToken cancellationToken = default)
 88    {
 389        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 390        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 91
 92        // All expiry/lease time math in this store runs on the database clock (SYSUTCDATETIME()),
 93        // never an app-computed timestamp: with multiple workers, app clock skew beyond the lease
 94        // window would let two nodes both consider a lease expired and double-run a flow.
 395        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 196        await using var command = connection.CreateCommand();
 197        command.CommandText = $"SELECT state_json, revision FROM {Table} WHERE flow_id = @flow_id AND expires_at_utc > S
 198        command.Parameters.AddWithValue("@flow_id", flowId);
 99
 1100        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 1101        if (!await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 1102            return null;
 103
 1104        return DurableFlowStoreShared.ReadState(flowId, reader.GetString(0), reader.GetInt64(1));
 1105    }
 106
 107    public async Task<bool> TryCreateAsync(string flowId, FlowState state, TimeSpan ttl, CancellationToken cancellationT
 108    {
 1109        DurableFlowStoreShared.ValidateCreate(flowId, state, ttl);
 1110        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "SQL Server");
 1111        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 1112        if (DurableFlowStoreShared.ShouldPrune(ref _lastPruneTicks, _options.PruneInterval))
 1113            await PruneExpiredAsync(cancellationToken).ConfigureAwait(false);
 114
 1115        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1116        await using var command = connection.CreateCommand();
 1117        command.CommandText =
 1118            $"""
 1119            MERGE {Table} WITH (HOLDLOCK) AS target
 1120            USING (SELECT @flow_id AS flow_id) AS source ON target.flow_id = source.flow_id
 1121            WHEN MATCHED AND target.expires_at_utc <= SYSUTCDATETIME() THEN
 1122                UPDATE SET state_json = @state_json,
 1123                           expires_at_utc = {AddMilliseconds("@ttl_ms")},
 1124                           updated_at_utc = SYSUTCDATETIME(),
 1125                           revision = @revision,
 1126                           lease_id = NULL,
 1127                           lease_expires_at_utc = NULL
 1128            WHEN NOT MATCHED THEN
 1129                INSERT (flow_id, state_json, expires_at_utc, updated_at_utc, revision)
 1130                VALUES (@flow_id, @state_json, {AddMilliseconds("@ttl_ms")}, SYSUTCDATETIME(), @revision);
 1131            """;
 1132        command.Parameters.AddWithValue("@flow_id", flowId);
 1133        command.Parameters.AddWithValue("@state_json", stateJson);
 1134        command.Parameters.AddWithValue("@ttl_ms", DurableFlowStoreShared.ServerClockTtlMilliseconds(ttl));
 1135        command.Parameters.AddWithValue("@revision", state.Revision);
 1136        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 1137    }
 138
 139    public async Task<bool> TryUpdateAsync(
 140        string flowId,
 141        FlowState state,
 142        long expectedRevision,
 143        TimeSpan ttl,
 144        string? leaseId = null,
 145        CancellationToken cancellationToken = default)
 146    {
 1147        DurableFlowStoreShared.ValidateUpdate(flowId, state, expectedRevision, ttl);
 1148        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "SQL Server");
 1149        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 150
 1151        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1152        await using var command = connection.CreateCommand();
 1153        command.CommandText =
 1154            $"""
 1155            UPDATE {Table}
 1156            SET state_json = @state_json,
 1157                expires_at_utc = {AddMilliseconds("@ttl_ms")},
 1158                updated_at_utc = SYSUTCDATETIME(),
 1159                revision = @new_revision
 1160            WHERE flow_id = @flow_id
 1161              AND revision = @expected_revision
 1162              AND expires_at_utc > SYSUTCDATETIME()
 1163              AND (@lease_id IS NULL OR (lease_id = @lease_id AND lease_expires_at_utc > SYSUTCDATETIME()));
 1164            """;
 1165        command.Parameters.AddWithValue("@flow_id", flowId);
 1166        command.Parameters.AddWithValue("@state_json", stateJson);
 1167        command.Parameters.AddWithValue("@ttl_ms", DurableFlowStoreShared.ServerClockTtlMilliseconds(ttl));
 1168        command.Parameters.AddWithValue("@expected_revision", expectedRevision);
 1169        command.Parameters.AddWithValue("@new_revision", state.Revision);
 1170        command.Parameters.AddWithValue("@lease_id", (object?)leaseId ?? DBNull.Value);
 1171        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 1172    }
 173
 174    public Task<bool> TryAcquireLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken canc
 3175        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: true, cancellationToken);
 176
 177    public Task<bool> TryRenewLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken cancel
 1178        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: false, cancellationToken);
 179
 180    public async Task ReleaseLeaseAsync(string flowId, string leaseId, CancellationToken cancellationToken = default)
 181    {
 1182        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 1183        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1184        await using var command = connection.CreateCommand();
 1185        command.CommandText = $"UPDATE {Table} SET lease_id = NULL, lease_expires_at_utc = NULL WHERE flow_id = @flow_id
 1186        command.Parameters.AddWithValue("@flow_id", flowId);
 1187        command.Parameters.AddWithValue("@lease_id", leaseId);
 1188        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 1189    }
 190
 191    public async Task<bool> TryDeleteAsync(string flowId, CancellationToken cancellationToken = default)
 192    {
 1193        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 1194        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 195
 1196        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1197        await using var command = connection.CreateCommand();
 1198        command.CommandText = $"DELETE FROM {Table} WHERE flow_id = @flow_id;";
 1199        command.Parameters.AddWithValue("@flow_id", flowId);
 1200        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 1201    }
 202
 203    private async Task PruneExpiredAsync(CancellationToken cancellationToken)
 204    {
 205        // One bounded batch per prune interval (policy shared by all relational stores): an
 206        // unbatched DELETE over a large expired backlog holds row locks and bloats one
 207        // transaction for the unlucky create that triggered the prune. Loads already filter on
 208        // expiry, so any backlog beyond the batch just waits for the next interval.
 1209        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1210        await using var command = connection.CreateCommand();
 1211        command.CommandText = $"DELETE TOP ({PruneBatchSize}) FROM {Table} WHERE expires_at_utc <= SYSUTCDATETIME();";
 1212        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 1213    }
 214
 215    private async Task EnsureCreatedAsync(CancellationToken cancellationToken)
 216    {
 3217        if (_created || !_options.AutoCreateSchema)
 3218            return;
 219
 1220        await _ensureGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 221        try
 222        {
 1223            if (_created)
 0224                return;
 225
 1226            await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1227            await using var transaction = (SqlTransaction)await connection.BeginTransactionAsync(cancellationToken).Conf
 228
 229            // Serialize schema creation across processes. The IF-NOT-EXISTS guards are not atomic
 230            // against a concurrent create of the same object (catalog errors 2714/2627). The
 231            // transaction-scoped application lock (keyed by schema, shared with the channel/transport
 232            // packages) lets one instance build the schema while the rest wait and then find it
 233            // already present.
 1234            await using (var lockCommand = connection.CreateCommand())
 235            {
 1236                lockCommand.Transaction = transaction;
 1237                lockCommand.CommandText =
 1238                    """
 1239                    DECLARE @lock_result int;
 1240                    EXEC @lock_result = sp_getapplock
 1241                        @Resource = @lock_resource,
 1242                        @LockMode = 'Exclusive',
 1243                        @LockOwner = 'Transaction',
 1244                        @LockTimeout = 60000;
 1245                    IF @lock_result < 0
 1246                        THROW 51000, N'Failed to acquire the AsyncResponse DDL application lock.', 1;
 1247                    """;
 1248                lockCommand.Parameters.AddWithValue("@lock_resource", DurableFlowStoreShared.SchemaLockResource(_options
 1249                await lockCommand.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 250            }
 251
 1252            await using var command = connection.CreateCommand();
 1253            command.Transaction = transaction;
 1254            command.CommandText =
 1255                $"""
 1256                IF SCHEMA_ID(N'{_options.SchemaName}') IS NULL
 1257                    EXEC(N'CREATE SCHEMA {Quote(_options.SchemaName)}');
 1258
 1259                IF OBJECT_ID(N'{_options.SchemaName}.{_options.TableName}', N'U') IS NULL
 1260                CREATE TABLE {Table} (
 1261                    flow_id nvarchar(400) NOT NULL PRIMARY KEY,
 1262                    state_json nvarchar(max) NOT NULL,
 1263                    expires_at_utc datetime2 NOT NULL,
 1264                    updated_at_utc datetime2 NOT NULL,
 1265                    revision bigint NOT NULL CONSTRAINT {Quote($"DF_{_options.TableName}_revision")} DEFAULT 0,
 1266                    lease_id nvarchar(64) NULL,
 1267                    lease_expires_at_utc datetime2 NULL
 1268                );
 1269
 1270                IF NOT EXISTS (SELECT 1 FROM sys.indexes WHERE name = N'{IndexName}' AND object_id = OBJECT_ID(N'{_optio
 1271                    CREATE INDEX {Quote(IndexName)} ON {Table} (expires_at_utc);
 1272                """;
 1273            await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 1274            await transaction.CommitAsync(cancellationToken).ConfigureAwait(false);
 1275            _created = true;
 1276        }
 277        finally
 278        {
 1279            _ensureGate.Release();
 280        }
 3281    }
 282
 283    private async Task<bool> UpdateLeaseAsync(
 284        string flowId,
 285        string leaseId,
 286        TimeSpan leaseDuration,
 287        bool acquire,
 288        CancellationToken cancellationToken)
 289    {
 3290        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 3291        ArgumentException.ThrowIfNullOrWhiteSpace(leaseId);
 3292        if (leaseDuration <= TimeSpan.Zero)
 2293            throw new ArgumentOutOfRangeException(nameof(leaseDuration));
 294
 1295        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 1296        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1297        await using var command = connection.CreateCommand();
 298        // Lease fencing runs entirely on the database clock: acquire steals only leases the
 299        // database considers expired, and renew/extend stays relative to SYSUTCDATETIME(), so
 300        // worker clock skew can never make two nodes hold the same lease.
 1301        command.CommandText =
 1302            $"""
 1303            UPDATE {Table}
 1304            SET lease_id = @lease_id, lease_expires_at_utc = {AddMilliseconds("@lease_ms")}
 1305            WHERE flow_id = @flow_id
 1306              AND expires_at_utc > SYSUTCDATETIME()
 1307              AND {(acquire ? "(lease_id IS NULL OR lease_expires_at_utc <= SYSUTCDATETIME() OR lease_id = @lease_id)" :
 1308            """;
 1309        command.Parameters.AddWithValue("@flow_id", flowId);
 1310        command.Parameters.AddWithValue("@lease_id", leaseId);
 1311        command.Parameters.AddWithValue("@lease_ms", DurableFlowStoreShared.ServerClockTtlMilliseconds(leaseDuration));
 1312        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 1313    }
 314
 315    private async Task<SqlConnection> OpenConnectionAsync(CancellationToken cancellationToken)
 316    {
 3317        var connection = new SqlConnection(_options.ConnectionString);
 318        try
 319        {
 3320            await connection.OpenAsync(cancellationToken).ConfigureAwait(false);
 1321            return connection;
 322        }
 2323        catch
 324        {
 2325            await connection.DisposeAsync().ConfigureAwait(false);
 2326            throw;
 327        }
 1328    }
 329
 330    /// <summary>
 331    /// SQL expression adding a millisecond bigint parameter to the database clock (the same
 332    /// pattern as the SQL Server channel package). DATEADD only takes int arguments, so the value
 333    /// is split into whole seconds and a sub-second remainder — TTLs and lease durations stay on
 334    /// the database clock, immune to app-side clock skew, without overflowing on multi-day spans
 335    /// such as the 7-day default state expiry.
 336    /// </summary>
 337    private static string AddMilliseconds(string parameterName)
 1338        => $"DATEADD(SECOND, CAST({parameterName} / 1000 AS int), DATEADD(MILLISECOND, CAST({parameterName} % 1000 AS in
 339
 1340    private string Table => $"{Quote(_options.SchemaName)}.{Quote(_options.TableName)}";
 1341    private string IndexName => $"{_options.TableName}_expires_idx";
 1342    private static string Quote(string identifier) => "[" + identifier + "]";
 343}
 344}
Generated by: ReportGenerator 5.5.10.0
08/03/2026 - 08:59:44
GitHub | reportgenerator.io

Methods/Properties

.ctor(Microsoft.Extensions.Options.IOptions<AsyncResponse.DurableFlows.SqlServer.SqlServerDurableFlowOptions>)
LoadAsync()
TryCreateAsync()
TryUpdateAsync()
TryAcquireLeaseAsync(string, string, System.TimeSpan, System.Threading.CancellationToken)
TryRenewLeaseAsync(string, string, System.TimeSpan, System.Threading.CancellationToken)
ReleaseLeaseAsync()
TryDeleteAsync()
PruneExpiredAsync()
EnsureCreatedAsync()
UpdateLeaseAsync()
OpenConnectionAsync()
AddMilliseconds(string)
get_Table()
get_IndexName()
Quote(string)