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

Information
Class: Microsoft.Extensions.DependencyInjection.SqlServerDurableFlowServiceCollectionExtensions
Assembly: AsyncResponse.DurableFlows.SqlServer
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/DurableFlows/AsyncResponse.DurableFlows.SqlServer/SqlServerDurableFlows.cs
Line coverage
100%
Covered lines: 2
Uncovered lines: 0
Coverable lines: 2
Total lines: 344
Line coverage: 100%
Branch coverage
N/A
Covered branches: 0
Total branches: 0
Branch coverage: N/A
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
WithSqlServerDurableFlows(...)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.
 221            builder.Services.TryAddSingleton<SqlServerFlowStateStore>();
 222            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;
 77    private readonly SemaphoreSlim _ensureGate = new(1, 1);
 78    private long _lastPruneTicks;
 79    private bool _created;
 80
 81    public SqlServerFlowStateStore(IOptions<SqlServerDurableFlowOptions> options)
 82    {
 83        _options = options.Value;
 84        _options.Validate();
 85    }
 86
 87    public async Task<FlowState?> LoadAsync(string flowId, CancellationToken cancellationToken = default)
 88    {
 89        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 90        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.
 95        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 96        await using var command = connection.CreateCommand();
 97        command.CommandText = $"SELECT state_json, revision FROM {Table} WHERE flow_id = @flow_id AND expires_at_utc > S
 98        command.Parameters.AddWithValue("@flow_id", flowId);
 99
 100        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 101        if (!await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 102            return null;
 103
 104        return DurableFlowStoreShared.ReadState(flowId, reader.GetString(0), reader.GetInt64(1));
 105    }
 106
 107    public async Task<bool> TryCreateAsync(string flowId, FlowState state, TimeSpan ttl, CancellationToken cancellationT
 108    {
 109        DurableFlowStoreShared.ValidateCreate(flowId, state, ttl);
 110        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "SQL Server");
 111        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 112        if (DurableFlowStoreShared.ShouldPrune(ref _lastPruneTicks, _options.PruneInterval))
 113            await PruneExpiredAsync(cancellationToken).ConfigureAwait(false);
 114
 115        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 116        await using var command = connection.CreateCommand();
 117        command.CommandText =
 118            $"""
 119            MERGE {Table} WITH (HOLDLOCK) AS target
 120            USING (SELECT @flow_id AS flow_id) AS source ON target.flow_id = source.flow_id
 121            WHEN MATCHED AND target.expires_at_utc <= SYSUTCDATETIME() THEN
 122                UPDATE SET state_json = @state_json,
 123                           expires_at_utc = {AddMilliseconds("@ttl_ms")},
 124                           updated_at_utc = SYSUTCDATETIME(),
 125                           revision = @revision,
 126                           lease_id = NULL,
 127                           lease_expires_at_utc = NULL
 128            WHEN NOT MATCHED THEN
 129                INSERT (flow_id, state_json, expires_at_utc, updated_at_utc, revision)
 130                VALUES (@flow_id, @state_json, {AddMilliseconds("@ttl_ms")}, SYSUTCDATETIME(), @revision);
 131            """;
 132        command.Parameters.AddWithValue("@flow_id", flowId);
 133        command.Parameters.AddWithValue("@state_json", stateJson);
 134        command.Parameters.AddWithValue("@ttl_ms", DurableFlowStoreShared.ServerClockTtlMilliseconds(ttl));
 135        command.Parameters.AddWithValue("@revision", state.Revision);
 136        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 137    }
 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    {
 147        DurableFlowStoreShared.ValidateUpdate(flowId, state, expectedRevision, ttl);
 148        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "SQL Server");
 149        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 150
 151        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 152        await using var command = connection.CreateCommand();
 153        command.CommandText =
 154            $"""
 155            UPDATE {Table}
 156            SET state_json = @state_json,
 157                expires_at_utc = {AddMilliseconds("@ttl_ms")},
 158                updated_at_utc = SYSUTCDATETIME(),
 159                revision = @new_revision
 160            WHERE flow_id = @flow_id
 161              AND revision = @expected_revision
 162              AND expires_at_utc > SYSUTCDATETIME()
 163              AND (@lease_id IS NULL OR (lease_id = @lease_id AND lease_expires_at_utc > SYSUTCDATETIME()));
 164            """;
 165        command.Parameters.AddWithValue("@flow_id", flowId);
 166        command.Parameters.AddWithValue("@state_json", stateJson);
 167        command.Parameters.AddWithValue("@ttl_ms", DurableFlowStoreShared.ServerClockTtlMilliseconds(ttl));
 168        command.Parameters.AddWithValue("@expected_revision", expectedRevision);
 169        command.Parameters.AddWithValue("@new_revision", state.Revision);
 170        command.Parameters.AddWithValue("@lease_id", (object?)leaseId ?? DBNull.Value);
 171        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 172    }
 173
 174    public Task<bool> TryAcquireLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken canc
 175        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: true, cancellationToken);
 176
 177    public Task<bool> TryRenewLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken cancel
 178        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: false, cancellationToken);
 179
 180    public async Task ReleaseLeaseAsync(string flowId, string leaseId, CancellationToken cancellationToken = default)
 181    {
 182        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 183        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 184        await using var command = connection.CreateCommand();
 185        command.CommandText = $"UPDATE {Table} SET lease_id = NULL, lease_expires_at_utc = NULL WHERE flow_id = @flow_id
 186        command.Parameters.AddWithValue("@flow_id", flowId);
 187        command.Parameters.AddWithValue("@lease_id", leaseId);
 188        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 189    }
 190
 191    public async Task<bool> TryDeleteAsync(string flowId, CancellationToken cancellationToken = default)
 192    {
 193        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 194        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 195
 196        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 197        await using var command = connection.CreateCommand();
 198        command.CommandText = $"DELETE FROM {Table} WHERE flow_id = @flow_id;";
 199        command.Parameters.AddWithValue("@flow_id", flowId);
 200        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 201    }
 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.
 209        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 210        await using var command = connection.CreateCommand();
 211        command.CommandText = $"DELETE TOP ({PruneBatchSize}) FROM {Table} WHERE expires_at_utc <= SYSUTCDATETIME();";
 212        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 213    }
 214
 215    private async Task EnsureCreatedAsync(CancellationToken cancellationToken)
 216    {
 217        if (_created || !_options.AutoCreateSchema)
 218            return;
 219
 220        await _ensureGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 221        try
 222        {
 223            if (_created)
 224                return;
 225
 226            await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 227            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.
 234            await using (var lockCommand = connection.CreateCommand())
 235            {
 236                lockCommand.Transaction = transaction;
 237                lockCommand.CommandText =
 238                    """
 239                    DECLARE @lock_result int;
 240                    EXEC @lock_result = sp_getapplock
 241                        @Resource = @lock_resource,
 242                        @LockMode = 'Exclusive',
 243                        @LockOwner = 'Transaction',
 244                        @LockTimeout = 60000;
 245                    IF @lock_result < 0
 246                        THROW 51000, N'Failed to acquire the AsyncResponse DDL application lock.', 1;
 247                    """;
 248                lockCommand.Parameters.AddWithValue("@lock_resource", DurableFlowStoreShared.SchemaLockResource(_options
 249                await lockCommand.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 250            }
 251
 252            await using var command = connection.CreateCommand();
 253            command.Transaction = transaction;
 254            command.CommandText =
 255                $"""
 256                IF SCHEMA_ID(N'{_options.SchemaName}') IS NULL
 257                    EXEC(N'CREATE SCHEMA {Quote(_options.SchemaName)}');
 258
 259                IF OBJECT_ID(N'{_options.SchemaName}.{_options.TableName}', N'U') IS NULL
 260                CREATE TABLE {Table} (
 261                    flow_id nvarchar(400) NOT NULL PRIMARY KEY,
 262                    state_json nvarchar(max) NOT NULL,
 263                    expires_at_utc datetime2 NOT NULL,
 264                    updated_at_utc datetime2 NOT NULL,
 265                    revision bigint NOT NULL CONSTRAINT {Quote($"DF_{_options.TableName}_revision")} DEFAULT 0,
 266                    lease_id nvarchar(64) NULL,
 267                    lease_expires_at_utc datetime2 NULL
 268                );
 269
 270                IF NOT EXISTS (SELECT 1 FROM sys.indexes WHERE name = N'{IndexName}' AND object_id = OBJECT_ID(N'{_optio
 271                    CREATE INDEX {Quote(IndexName)} ON {Table} (expires_at_utc);
 272                """;
 273            await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 274            await transaction.CommitAsync(cancellationToken).ConfigureAwait(false);
 275            _created = true;
 276        }
 277        finally
 278        {
 279            _ensureGate.Release();
 280        }
 281    }
 282
 283    private async Task<bool> UpdateLeaseAsync(
 284        string flowId,
 285        string leaseId,
 286        TimeSpan leaseDuration,
 287        bool acquire,
 288        CancellationToken cancellationToken)
 289    {
 290        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 291        ArgumentException.ThrowIfNullOrWhiteSpace(leaseId);
 292        if (leaseDuration <= TimeSpan.Zero)
 293            throw new ArgumentOutOfRangeException(nameof(leaseDuration));
 294
 295        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 296        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 297        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.
 301        command.CommandText =
 302            $"""
 303            UPDATE {Table}
 304            SET lease_id = @lease_id, lease_expires_at_utc = {AddMilliseconds("@lease_ms")}
 305            WHERE flow_id = @flow_id
 306              AND expires_at_utc > SYSUTCDATETIME()
 307              AND {(acquire ? "(lease_id IS NULL OR lease_expires_at_utc <= SYSUTCDATETIME() OR lease_id = @lease_id)" :
 308            """;
 309        command.Parameters.AddWithValue("@flow_id", flowId);
 310        command.Parameters.AddWithValue("@lease_id", leaseId);
 311        command.Parameters.AddWithValue("@lease_ms", DurableFlowStoreShared.ServerClockTtlMilliseconds(leaseDuration));
 312        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 313    }
 314
 315    private async Task<SqlConnection> OpenConnectionAsync(CancellationToken cancellationToken)
 316    {
 317        var connection = new SqlConnection(_options.ConnectionString);
 318        try
 319        {
 320            await connection.OpenAsync(cancellationToken).ConfigureAwait(false);
 321            return connection;
 322        }
 323        catch
 324        {
 325            await connection.DisposeAsync().ConfigureAwait(false);
 326            throw;
 327        }
 328    }
 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)
 338        => $"DATEADD(SECOND, CAST({parameterName} / 1000 AS int), DATEADD(MILLISECOND, CAST({parameterName} % 1000 AS in
 339
 340    private string Table => $"{Quote(_options.SchemaName)}.{Quote(_options.TableName)}";
 341    private string IndexName => $"{_options.TableName}_expires_idx";
 342    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

WithSqlServerDurableFlows(Microsoft.Extensions.DependencyInjection.AsyncResponseRegistrationBuilder, System.Action<AsyncResponse.DurableFlows.SqlServer.SqlServerDurableFlowOptions>)