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

Information
Class: AsyncResponse.DurableFlows.SqlServer.SqlServerFlowStateStore
Assembly: AsyncResponse.DurableFlows.SqlServer
File(s): /_/src/DurableFlows/AsyncResponse.DurableFlows.SqlServer/SqlServerDurableFlows.cs
Line coverage
88%
Covered lines: 209
Uncovered lines: 26
Coverable lines: 235
Total lines: 486
Line coverage: 88.9%
Branch coverage
86%
Covered branches: 26
Total branches: 30
Branch coverage: 86.6%
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%44100%
ValidateCreate(...)50%22100%
TryCreateAsync()100%22100%
TryUpdateAsync()100%22100%
TryAcquireLeaseAsync(...)100%11100%
TryRenewLeaseAsync(...)100%11100%
ReleaseLeaseAsync()100%11100%
ObserveLeaseAsync()100%88100%
TryDeleteAsync()100%11100%
PruneExpiredAsync()100%11100%
EnsureCreatedAsync()70%111081.42%
VerifyRelationsAsync(...)100%11100%
ObjectExistsAsync()100%210%
ExpectedObjects()100%11100%
UpdateLeaseAsync()100%22100%
OpenConnectionAsync(...)100%11100%
AddMilliseconds(...)100%11100%
get_Table()100%11100%
get_IndexName()100%11100%
Quote(...)100%11100%

File(s)

/_/src/DurableFlows/AsyncResponse.DurableFlows.SqlServer/SqlServerDurableFlows.cs

#LineLine coverage
 1using AsyncResponse;
 2using AsyncResponse.DurableFlows.Internal;
 3using AsyncResponse.DurableFlows.SqlServer;
 4using AsyncResponse.Internal;
 5using Microsoft.Data.SqlClient;
 6using Microsoft.Extensions.DependencyInjection.Extensions;
 7using Microsoft.Extensions.Logging;
 8using Microsoft.Extensions.Options;
 9
 10namespace Microsoft.Extensions.DependencyInjection
 11{
 12    /// <summary>DI registration for the SQL Server durable-flow state store.</summary>
 13    public static class SqlServerDurableFlowServiceCollectionExtensions
 14    {
 15        /// <summary>Stores durable-flow state in SQL Server.</summary>
 16        public static AsyncResponseRegistrationBuilder WithSqlServerDurableFlows(
 17            this AsyncResponseRegistrationBuilder builder,
 18            Action<SqlServerDurableFlowOptions>? configure = null)
 19        {
 20            // Singleton on purpose: schema provisioning is cached per store instance, and the
 21            // executor resolves the store from a fresh scope per flow execution — a scoped store
 22            // would re-run EnsureCreated's DDL round-trip on every run.
 23            builder.Services.TryAddSingleton<SqlServerFlowStateStore>();
 24            return builder.WithDurableFlows<SqlServerFlowStateStore, SqlServerDurableFlowOptions>(configure);
 25        }
 26    }
 27}
 28
 29namespace AsyncResponse.DurableFlows.SqlServer
 30{
 31/// <summary>Options for the SQL Server durable-flow state store.</summary>
 32public sealed class SqlServerDurableFlowOptions : DurableFlowOptions
 33{
 34    /// <summary>SQL Server connection string. Required.</summary>
 35    public string? ConnectionString { get; set; }
 36
 37    /// <summary>Database schema that contains the durable-flow table. Default: <c>dbo</c>.</summary>
 38    public string SchemaName { get; set; } = "dbo";
 39
 40    /// <summary>Table storing one durable-flow ledger row per flow id.</summary>
 41    public string TableName { get; set; } = "asyncresponse_flow_state";
 42
 43    /// <summary>Creates the schema, table, and expiry index on first use.</summary>
 44    public bool AutoCreateSchema { get; set; } = true;
 45
 46    /// <summary>
 47    /// How often <see cref="SqlServerFlowStateStore.TryCreateAsync"/> opportunistically deletes one
 48    /// bounded batch (1000 rows) of expired rows (loads already treat expired state as absent;
 49    /// pruning bounds table growth). Zero or negative prunes on every save. Default: 5 minutes.
 50    /// </summary>
 51    public TimeSpan PruneInterval { get; set; } = TimeSpan.FromMinutes(5);
 52
 53    /// <summary>
 54    /// Wall-clock budget one opportunistic prune may spend draining expired rows in batches of
 55    /// 1000 after its first batch (the first always runs). A single batch per interval capped
 56    /// cleanup at ~3 rows/second, which any busier instance outgrew forever; the prune now drains
 57    /// batches until one comes back short or this budget lapses, and reports the outcome on the
 58    /// <c>AsyncResponse</c> meter (<c>asyncresponse.flow_state.pruned_rows</c>,
 59    /// <c>prune_failures</c>, <c>prune_budget_exhausted</c>) and the store's logger. The create
 60    /// that triggers the prune waits for it, so this bounds that create's added latency. Zero
 61    /// keeps the historical single batch. Default: 2 seconds.
 62    /// </summary>
 63    public TimeSpan PruneBudget { get; set; } = DurableFlowStoreShared.DefaultPruneBudget;
 64
 65    /// <summary>
 66    /// Maximum serialized flow-state size in bytes accepted by writes; oversized ledgers fail fast
 67    /// with an actionable error instead of an opaque provider error. Default: <c>null</c>
 68    /// (unlimited — <c>nvarchar(max)</c> is effectively unbounded), settable as an operator budget.
 69    /// </summary>
 70    public long? MaxStateBytes { get; set; }
 71
 72    /// <summary>Validates option values and throws on misconfiguration.</summary>
 73    public void Validate()
 74    {
 75        DurableFlowStoreShared.ValidateConnectionString(ConnectionString, nameof(SqlServerDurableFlowOptions));
 76        DurableFlowStoreShared.ValidateIdentifier(SchemaName, $"{nameof(SqlServerDurableFlowOptions)}.{nameof(SchemaName
 77        DurableFlowStoreShared.ValidateIdentifier(TableName, $"{nameof(SqlServerDurableFlowOptions)}.{nameof(TableName)}
 78        DurableFlowStoreShared.ValidateMaxStateBytes(MaxStateBytes, nameof(SqlServerDurableFlowOptions));
 79        DurableFlowStoreShared.ValidatePruneBudget(PruneBudget, nameof(SqlServerDurableFlowOptions));
 80    }
 81}
 82
 83/// <summary>SQL Server implementation of <see cref="IFlowStateStore"/>.</summary>
 84public sealed class SqlServerFlowStateStore : IFlowStateStore
 85{
 86    private readonly ILogger<SqlServerFlowStateStore>? _logger;
 87
 88    private readonly SqlServerDurableFlowOptions _options;
 21589    private readonly SemaphoreSlim _ensureGate = new(1, 1);
 90    private long _lastPruneTicks;
 91    private volatile bool _created;
 92
 21593    public SqlServerFlowStateStore(IOptions<SqlServerDurableFlowOptions> options, ILogger<SqlServerFlowStateStore>? logg
 94    {
 21595        _options = options.Value;
 21596        _options.Validate();
 21397        _logger = logger;
 21398    }
 99
 100    public async Task<FlowState?> LoadAsync(string flowId, CancellationToken cancellationToken = default)
 101    {
 681102        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 681103        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 104
 105        // All expiry/lease time math in this store runs on the database clock (SYSUTCDATETIME()),
 106        // never an app-computed timestamp: with multiple workers, app clock skew beyond the lease
 107        // window would let two nodes both consider a lease expired and double-run a flow.
 679108        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 679109        await using var command = connection.CreateCommand();
 679110        command.CommandText = $"SELECT state_json, revision FROM {Table} WHERE flow_id = @flow_id AND expires_at_utc > S
 679111        command.Parameters.AddWithValue("@flow_id", flowId);
 112
 679113        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 679114        if (!await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 4115            return null;
 116
 675117        return DurableFlowStoreShared.ReadState(flowId, reader.GetString(0), reader.GetInt64(1));
 677118    }
 119
 120    /// <inheritdoc />
 121    public void ValidateCreate(string flowId, FlowState state, TimeSpan ttl)
 122    {
 136123        DurableFlowStoreShared.ValidateCreate(flowId, state, ttl);
 136124        if (_options.MaxStateBytes is not null)
 4125            _ = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "SQL Server");
 134126    }
 127
 128    public async Task<bool> TryCreateAsync(string flowId, FlowState state, TimeSpan ttl, CancellationToken cancellationT
 129    {
 293130        DurableFlowStoreShared.ValidateCreate(flowId, state, ttl);
 292131        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "SQL Server");
 292132        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 292133        if (DurableFlowStoreShared.ShouldPrune(ref _lastPruneTicks, _options.PruneInterval))
 270134            await DurableFlowStoreShared.PruneQuietlyAsync(() => PruneExpiredAsync(cancellationToken), _options.PruneBud
 135
 292136        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 292137        await using var command = connection.CreateCommand();
 292138        command.CommandText =
 292139            $"""
 292140            MERGE {Table} WITH (HOLDLOCK) AS target
 292141            USING (SELECT @flow_id AS flow_id) AS source ON target.flow_id = source.flow_id
 292142            WHEN MATCHED AND target.expires_at_utc <= SYSUTCDATETIME() THEN
 292143                UPDATE SET state_json = @state_json,
 292144                           expires_at_utc = {AddMilliseconds("@ttl_ms")},
 292145                           updated_at_utc = SYSUTCDATETIME(),
 292146                           revision = @revision,
 292147                           lease_id = NULL,
 292148                           lease_expires_at_utc = NULL
 292149            WHEN NOT MATCHED THEN
 292150                INSERT (flow_id, state_json, expires_at_utc, updated_at_utc, revision)
 292151                VALUES (@flow_id, @state_json, {AddMilliseconds("@ttl_ms")}, SYSUTCDATETIME(), @revision);
 292152            """;
 292153        command.Parameters.AddWithValue("@flow_id", flowId);
 292154        command.Parameters.AddWithValue("@state_json", stateJson);
 292155        command.Parameters.AddWithValue("@ttl_ms", DurableFlowStoreShared.ServerClockTtlMilliseconds(ttl));
 292156        command.Parameters.AddWithValue("@revision", state.Revision);
 292157        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 292158    }
 159
 160    public async Task<bool> TryUpdateAsync(
 161        string flowId,
 162        FlowState state,
 163        long expectedRevision,
 164        TimeSpan ttl,
 165        string? leaseId = null,
 166        CancellationToken cancellationToken = default)
 167    {
 863168        DurableFlowStoreShared.ValidateUpdate(flowId, state, expectedRevision, ttl);
 863169        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "SQL Server");
 863170        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 171
 863172        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 863173        await using var command = connection.CreateCommand();
 863174        command.CommandText =
 863175            $"""
 863176            UPDATE {Table}
 863177            SET state_json = @state_json,
 863178                expires_at_utc = {AddMilliseconds("@ttl_ms")},
 863179                updated_at_utc = SYSUTCDATETIME(),
 863180                revision = @new_revision
 863181            WHERE flow_id = @flow_id
 863182              AND revision = @expected_revision
 863183              AND expires_at_utc > SYSUTCDATETIME()
 863184              AND (@lease_id IS NULL OR (lease_id = @lease_id AND lease_expires_at_utc > SYSUTCDATETIME()));
 863185            """;
 863186        command.Parameters.AddWithValue("@flow_id", flowId);
 863187        command.Parameters.AddWithValue("@state_json", stateJson);
 863188        command.Parameters.AddWithValue("@ttl_ms", DurableFlowStoreShared.ServerClockTtlMilliseconds(ttl));
 863189        command.Parameters.AddWithValue("@expected_revision", expectedRevision);
 863190        command.Parameters.AddWithValue("@new_revision", state.Revision);
 863191        command.Parameters.AddWithValue("@lease_id", (object?)leaseId ?? DBNull.Value);
 863192        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 863193    }
 194
 195    public Task<bool> TryAcquireLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken canc
 150196        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: true, cancellationToken);
 197
 198    public Task<bool> TryRenewLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken cancel
 9199        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: false, cancellationToken);
 200
 201    public async Task ReleaseLeaseAsync(string flowId, string leaseId, CancellationToken cancellationToken = default)
 202    {
 142203        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 142204        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 142205        await using var command = connection.CreateCommand();
 142206        command.CommandText = $"UPDATE {Table} SET lease_id = NULL, lease_expires_at_utc = NULL WHERE flow_id = @flow_id
 142207        command.Parameters.AddWithValue("@flow_id", flowId);
 142208        command.Parameters.AddWithValue("@lease_id", leaseId);
 142209        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 142210    }
 211
 212    /// <inheritdoc />
 213    public async Task<FlowLeaseObservation?> ObserveLeaseAsync(string flowId, CancellationToken cancellationToken = defa
 214    {
 18215        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 12216        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 217
 218        // The two lease columns exactly as stored — deliberately no SYSUTCDATETIME() predicate,
 219        // unlike every other statement in this store: an expired lease nobody has taken over must
 220        // keep reading as the same lease, because the engine's proof of a live holder is that two
 221        // observations DIFFER. Whether it has lapsed stays UpdateLeaseAsync's call, on the
 222        // database clock. datetime2 carries no zone and reads back Unspecified; the value is
 223        // SYSUTCDATETIME() arithmetic, and the shared shaper stamps it UTC at full datetime2(7)
 224        // precision.
 12225        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 12226        await using var command = connection.CreateCommand();
 12227        command.CommandText = $"SELECT lease_id, lease_expires_at_utc FROM {Table} WHERE flow_id = @flow_id;";
 12228        command.Parameters.AddWithValue("@flow_id", flowId);
 229
 12230        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 12231        if (!await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 2232            return FlowLeaseObservation.Unheld;
 233
 10234        return DurableFlowStoreShared.LeaseObservation(
 10235            reader.IsDBNull(0) ? null : reader.GetString(0),
 10236            reader.IsDBNull(1) ? null : reader.GetDateTime(1));
 12237    }
 238
 239    public async Task<bool> TryDeleteAsync(string flowId, CancellationToken cancellationToken = default)
 240    {
 10241        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 10242        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 243
 10244        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 10245        await using var command = connection.CreateCommand();
 10246        command.CommandText = $"DELETE FROM {Table} WHERE flow_id = @flow_id;";
 10247        command.Parameters.AddWithValue("@flow_id", flowId);
 10248        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 10249    }
 250
 251    private async Task<int> PruneExpiredAsync(CancellationToken cancellationToken)
 252    {
 253        // One bounded batch per call; DurableFlowStoreShared.PruneQuietlyAsync repeats it under
 254        // the PruneBudget while batches come back full (policy shared by all relational stores): an
 255        // unbatched DELETE over a large expired backlog holds row locks and bloats one
 256        // transaction for the unlucky create that triggered the prune. Loads already filter on
 257        // expiry, so any backlog beyond the batch just waits for the next interval.
 135258        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 135259        await using var command = connection.CreateCommand();
 135260        command.CommandText = $"DELETE TOP ({DurableFlowStoreShared.PruneBatchSize}) FROM {Table} WHERE expires_at_utc <
 135261        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 135262    }
 263
 264    private async Task EnsureCreatedAsync(CancellationToken cancellationToken)
 265    {
 2157266        if (_created)
 1909267            return;
 268
 248269        await _ensureGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 270        try
 271        {
 248272            if (_created)
 111273                return;
 274
 137275            await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 276
 135277            if (!_options.AutoCreateSchema)
 278            {
 279                // Operator-managed schema: no DDL and no DDL lock, but the SAME catalog
 280                // verification the DDL path runs — an operator-provisioned table with the wrong
 281                // shape (a case-insensitive flow_id collation above all) fails silently at
 282                // runtime, which is exactly what verification exists to catch. An absent object is
 283                // fine: the migration has not run yet, the first query surfaces a clear SQL Server
 284                // error (the documented "create it yourself, later" workflow), and _created stays
 285                // unlatched so a later operation re-verifies once the migration lands.
 0286                if (!await ObjectExistsAsync(connection, cancellationToken).ConfigureAwait(false))
 287                    return;
 288
 0289                await VerifyRelationsAsync(connection, transaction: null, cancellationToken).ConfigureAwait(false);
 0290                _created = true;
 0291                return;
 292            }
 293
 135294            await using var transaction = (SqlTransaction)await connection.BeginTransactionAsync(cancellationToken).Conf
 295
 296            // Serialize schema creation across processes. The IF-NOT-EXISTS guards are not atomic
 297            // against a concurrent create of the same object (catalog errors 2714/2627). The
 298            // transaction-scoped application lock (keyed by schema, shared with the channel/transport
 299            // packages) lets one instance build the schema while the rest wait and then find it
 300            // already present.
 135301            await using (var lockCommand = connection.CreateCommand())
 302            {
 135303                lockCommand.Transaction = transaction;
 135304                lockCommand.CommandText =
 135305                    """
 135306                    DECLARE @lock_result int;
 135307                    EXEC @lock_result = sp_getapplock
 135308                        @Resource = @lock_resource,
 135309                        @LockMode = 'Exclusive',
 135310                        @LockOwner = 'Transaction',
 135311                        @LockTimeout = 60000;
 135312                    IF @lock_result < 0
 135313                        THROW 51000, N'Failed to acquire the AsyncResponse DDL application lock.', 1;
 135314                    """;
 135315                lockCommand.Parameters.AddWithValue("@lock_resource", DurableFlowStoreShared.SchemaLockResource(_options
 135316                await lockCommand.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 317            }
 318
 135319            await using var command = connection.CreateCommand();
 135320            command.Transaction = transaction;
 135321            command.CommandText =
 135322                $"""
 135323                IF SCHEMA_ID(N'{_options.SchemaName}') IS NULL
 135324                    EXEC(N'CREATE SCHEMA {Quote(_options.SchemaName)}');
 135325
 135326                IF OBJECT_ID(N'{_options.SchemaName}.{_options.TableName}', N'U') IS NULL
 135327                CREATE TABLE {Table} (
 135328                    flow_id nvarchar(400) COLLATE Latin1_General_100_BIN2 NOT NULL PRIMARY KEY,
 135329                    state_json nvarchar(max) NOT NULL,
 135330                    expires_at_utc datetime2 NOT NULL,
 135331                    updated_at_utc datetime2 NOT NULL,
 135332                    -- Unnamed DEFAULT deliberately: the previous derived name prefixed "DF_" and
 135333                    -- suffixed "_revision" onto the full table name, so a table name this store
 135334                    -- otherwise accepts (117 characters) produced a 129-character constraint name
 135335                    -- and the CREATE failed with error 103 — SQL Server's identifier cap is 128.
 135336                    -- Nothing reads the constraint by name; the store never drops or alters it.
 135337                    revision bigint NOT NULL DEFAULT 0,
 135338                    lease_id nvarchar(64) NULL,
 135339                    lease_expires_at_utc datetime2 NULL
 135340                );
 135341
 135342                IF NOT EXISTS (SELECT 1 FROM sys.indexes WHERE name = N'{IndexName}' AND object_id = OBJECT_ID(N'{_optio
 135343                    CREATE INDEX {Quote(IndexName)} ON {Table} (expires_at_utc);
 135344                """;
 345            try
 346            {
 135347                await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 135348            }
 0349            catch (SqlException ex)
 350            {
 351                // The batch can break BEFORE the verification below runs: a name held by another
 352                // component's table suppresses the guarded CREATE and the index that follows hits
 353                // the wrong table, and a name held by a view fails outright with error 2714. Run
 354                // the same catalog checks now, on a fresh connection (the objects in question are
 355                // somebody else's and already committed), so the operator gets the precise reason.
 0356                await SqlServerRelationVerifier.ThrowDiagnosedCollisionAsync(
 0357                    OpenConnectionAsync,
 0358                    ex,
 0359                    _options.SchemaName,
 0360                    "durable-flow",
 0361                    ExpectedObjects(),
 0362                    cancellationToken).ConfigureAwait(false);
 0363                throw;
 364            }
 365
 366            // Post-DDL catalog verification inside the DDL transaction (and therefore under the
 367            // shared application lock): the existence guard above only asks "is there a user table
 368            // with this name", so another component's table silently suppresses creation and a
 369            // view or synonym makes the CREATE fail with raw error 2714.
 135370            await transaction.CommitAsync(cancellationToken).ConfigureAwait(false);
 371
 372            // Verified AFTER the commit, on the same connection but outside the transaction. The
 373            // checks read the catalog, and a transaction that has just run DDL still holds
 374            // schema-modification locks — catalog reads under those deadlock (error 1205) against
 375            // this store's own live traffic, which is already polling by the time a later
 376            // EnsureCreated re-runs. Correctness does not need the transaction: the application
 377            // lock serialized the DDL, and what these checks look for is somebody ELSE'S committed
 378            // object occupying a name, never our own uncommitted work.
 135379            await VerifyRelationsAsync(connection, transaction: null, cancellationToken).ConfigureAwait(false);
 135380            _created = true;
 135381        }
 382        finally
 383        {
 248384            _ensureGate.Release();
 385        }
 2155386    }
 387
 388    private Task VerifyRelationsAsync(SqlConnection connection, SqlTransaction? transaction, CancellationToken cancellat
 135389        => SqlServerRelationVerifier.VerifyAsync(
 135390            connection,
 135391            transaction,
 135392            _options.SchemaName,
 135393            "durable-flow",
 135394            ExpectedObjects(),
 135395            cancellationToken);
 396
 397    /// <summary>
 398    /// Reports whether ANY object occupies the configured name (any kind: a view or foreign
 399    /// component's object must reach verification, which names the precise wrong-kind reason
 400    /// instead of skipping the checks). The catalog's own collation decides case matching, exactly
 401    /// as the server resolves the runtime identifier.
 402    /// </summary>
 403    private async Task<bool> ObjectExistsAsync(SqlConnection connection, CancellationToken cancellationToken)
 404    {
 0405        await using var command = connection.CreateCommand();
 0406        command.CommandText =
 0407            """
 0408            SELECT CASE WHEN EXISTS (
 0409                SELECT 1
 0410                FROM sys.objects o
 0411                JOIN sys.schemas s ON s.schema_id = o.schema_id
 0412                WHERE s.name = @schema AND o.name = @table) THEN 1 ELSE 0 END;
 0413            """;
 0414        command.Parameters.AddWithValue("@schema", _options.SchemaName);
 0415        command.Parameters.AddWithValue("@table", _options.TableName);
 0416        return (int)(await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false))! == 1;
 0417    }
 418
 419    /// <summary>The catalog shape this store's DDL intends — the single source for both the
 420    /// post-DDL verification and the failed-batch diagnosis.</summary>
 421    /// <remarks>A bare <c>datetime2</c> declaration is <c>datetime2(7)</c>; the expected types
 422    /// state the scale, because a reduced-scale <c>lease_expires_at_utc</c> rounds the fence on
 423    /// store — two nodes can then both read the lease as expired and run one flow at once.</remarks>
 424    private SqlServerRelationVerifier.ExpectedObject[] ExpectedObjects() =>
 137425            [
 137426                new(_options.TableName, SqlServerObjectKind.Table,
 137427                [
 137428                    new("flow_id", "nvarchar(400)", Nullable: false, RequiresBinaryCollation: true),
 137429                    new("state_json", "nvarchar(max)", Nullable: false),
 137430                    new("expires_at_utc", "datetime2(7)", Nullable: false),
 137431                    new("updated_at_utc", "datetime2(7)", Nullable: false),
 137432                    new("revision", "bigint", Nullable: false),
 137433                    new("lease_id", "nvarchar(64)", Nullable: true),
 137434                    new("lease_expires_at_utc", "datetime2(7)", Nullable: true)
 137435                ],
 137436                PrimaryKey: ["flow_id"])
 137437            ];
 438
 439
 440    private async Task<bool> UpdateLeaseAsync(
 441        string flowId,
 442        string leaseId,
 443        TimeSpan leaseDuration,
 444        bool acquire,
 445        CancellationToken cancellationToken)
 446    {
 159447        DurableFlowStoreShared.ValidateLeaseArgs(flowId, leaseId, leaseDuration);
 448
 157449        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 157450        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 157451        await using var command = connection.CreateCommand();
 452        // Lease fencing runs entirely on the database clock: acquire steals only leases the
 453        // database considers expired, and renew/extend stays relative to SYSUTCDATETIME(), so
 454        // worker clock skew can never make two nodes hold the same lease.
 157455        command.CommandText =
 157456            $"""
 157457            UPDATE {Table}
 157458            SET lease_id = @lease_id, lease_expires_at_utc = {AddMilliseconds("@lease_ms")}
 157459            WHERE flow_id = @flow_id
 157460              AND expires_at_utc > SYSUTCDATETIME()
 157461              AND {(acquire ? "(lease_id IS NULL OR lease_expires_at_utc <= SYSUTCDATETIME() OR lease_id = @lease_id)" :
 157462            """;
 157463        command.Parameters.AddWithValue("@flow_id", flowId);
 157464        command.Parameters.AddWithValue("@lease_id", leaseId);
 157465        command.Parameters.AddWithValue("@lease_ms", DurableFlowStoreShared.ServerClockTtlMilliseconds(leaseDuration));
 157466        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 157467    }
 468
 469    private Task<SqlConnection> OpenConnectionAsync(CancellationToken cancellationToken)
 2427470        => DurableFlowStoreShared.OpenConnectionAsync<SqlConnection>(_options.ConnectionString, cancellationToken);
 471
 472    /// <summary>
 473    /// SQL expression adding a millisecond bigint parameter to the database clock (the same
 474    /// pattern as the SQL Server channel package). DATEADD only takes int arguments, so the value
 475    /// is split into whole seconds and a sub-second remainder — TTLs and lease durations stay on
 476    /// the database clock, immune to app-side clock skew, without overflowing on multi-day spans
 477    /// such as the 7-day default state expiry.
 478    /// </summary>
 479    private static string AddMilliseconds(string parameterName)
 1604480        => $"DATEADD(SECOND, CAST({parameterName} / 1000 AS int), DATEADD(MILLISECOND, CAST({parameterName} % 1000 AS in
 481
 2560482    private string Table => $"{Quote(_options.SchemaName)}.{Quote(_options.TableName)}";
 270483    private string IndexName => DurableFlowStoreShared.DerivedName(_options.TableName, "_expires_idx", 128);
 5390484    private static string Quote(string identifier) => "[" + identifier + "]";
 485}
 486}
Generated by: ReportGenerator 5.5.11.0
09/22/2026 - 09:41:04
GitHub | reportgenerator.io

Methods/Properties

.ctor(Microsoft.Extensions.Options.IOptions`1<AsyncResponse.DurableFlows.SqlServer.SqlServerDurableFlowOptions>,Microsoft.Extensions.Logging.ILogger`1<AsyncResponse.DurableFlows.SqlServer.SqlServerFlowStateStore>)
LoadAsync()
ValidateCreate(System.String,AsyncResponse.FlowState,System.TimeSpan)
TryCreateAsync()
TryUpdateAsync()
TryAcquireLeaseAsync(System.String,System.String,System.TimeSpan,System.Threading.CancellationToken)
TryRenewLeaseAsync(System.String,System.String,System.TimeSpan,System.Threading.CancellationToken)
ReleaseLeaseAsync()
ObserveLeaseAsync()
TryDeleteAsync()
PruneExpiredAsync()
EnsureCreatedAsync()
VerifyRelationsAsync(Microsoft.Data.SqlClient.SqlConnection,Microsoft.Data.SqlClient.SqlTransaction,System.Threading.CancellationToken)
ObjectExistsAsync()
ExpectedObjects()
UpdateLeaseAsync()
OpenConnectionAsync(System.Threading.CancellationToken)
AddMilliseconds(System.String)
get_Table()
get_IndexName()
Quote(System.String)