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

Information
Class: Microsoft.Extensions.DependencyInjection.SqlServerDurableFlowServiceCollectionExtensions
Assembly: AsyncResponse.DurableFlows.SqlServer
File(s): /_/src/DurableFlows/AsyncResponse.DurableFlows.SqlServer/SqlServerDurableFlows.cs
Line coverage
100%
Covered lines: 2
Uncovered lines: 0
Coverable lines: 2
Total lines: 486
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)

/_/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.
 20423            builder.Services.TryAddSingleton<SqlServerFlowStateStore>();
 20424            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;
 89    private readonly SemaphoreSlim _ensureGate = new(1, 1);
 90    private long _lastPruneTicks;
 91    private volatile bool _created;
 92
 93    public SqlServerFlowStateStore(IOptions<SqlServerDurableFlowOptions> options, ILogger<SqlServerFlowStateStore>? logg
 94    {
 95        _options = options.Value;
 96        _options.Validate();
 97        _logger = logger;
 98    }
 99
 100    public async Task<FlowState?> LoadAsync(string flowId, CancellationToken cancellationToken = default)
 101    {
 102        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 103        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.
 108        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 109        await using var command = connection.CreateCommand();
 110        command.CommandText = $"SELECT state_json, revision FROM {Table} WHERE flow_id = @flow_id AND expires_at_utc > S
 111        command.Parameters.AddWithValue("@flow_id", flowId);
 112
 113        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 114        if (!await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 115            return null;
 116
 117        return DurableFlowStoreShared.ReadState(flowId, reader.GetString(0), reader.GetInt64(1));
 118    }
 119
 120    /// <inheritdoc />
 121    public void ValidateCreate(string flowId, FlowState state, TimeSpan ttl)
 122    {
 123        DurableFlowStoreShared.ValidateCreate(flowId, state, ttl);
 124        if (_options.MaxStateBytes is not null)
 125            _ = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "SQL Server");
 126    }
 127
 128    public async Task<bool> TryCreateAsync(string flowId, FlowState state, TimeSpan ttl, CancellationToken cancellationT
 129    {
 130        DurableFlowStoreShared.ValidateCreate(flowId, state, ttl);
 131        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "SQL Server");
 132        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 133        if (DurableFlowStoreShared.ShouldPrune(ref _lastPruneTicks, _options.PruneInterval))
 134            await DurableFlowStoreShared.PruneQuietlyAsync(() => PruneExpiredAsync(cancellationToken), _options.PruneBud
 135
 136        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 137        await using var command = connection.CreateCommand();
 138        command.CommandText =
 139            $"""
 140            MERGE {Table} WITH (HOLDLOCK) AS target
 141            USING (SELECT @flow_id AS flow_id) AS source ON target.flow_id = source.flow_id
 142            WHEN MATCHED AND target.expires_at_utc <= SYSUTCDATETIME() THEN
 143                UPDATE SET state_json = @state_json,
 144                           expires_at_utc = {AddMilliseconds("@ttl_ms")},
 145                           updated_at_utc = SYSUTCDATETIME(),
 146                           revision = @revision,
 147                           lease_id = NULL,
 148                           lease_expires_at_utc = NULL
 149            WHEN NOT MATCHED THEN
 150                INSERT (flow_id, state_json, expires_at_utc, updated_at_utc, revision)
 151                VALUES (@flow_id, @state_json, {AddMilliseconds("@ttl_ms")}, SYSUTCDATETIME(), @revision);
 152            """;
 153        command.Parameters.AddWithValue("@flow_id", flowId);
 154        command.Parameters.AddWithValue("@state_json", stateJson);
 155        command.Parameters.AddWithValue("@ttl_ms", DurableFlowStoreShared.ServerClockTtlMilliseconds(ttl));
 156        command.Parameters.AddWithValue("@revision", state.Revision);
 157        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 158    }
 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    {
 168        DurableFlowStoreShared.ValidateUpdate(flowId, state, expectedRevision, ttl);
 169        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "SQL Server");
 170        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 171
 172        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 173        await using var command = connection.CreateCommand();
 174        command.CommandText =
 175            $"""
 176            UPDATE {Table}
 177            SET state_json = @state_json,
 178                expires_at_utc = {AddMilliseconds("@ttl_ms")},
 179                updated_at_utc = SYSUTCDATETIME(),
 180                revision = @new_revision
 181            WHERE flow_id = @flow_id
 182              AND revision = @expected_revision
 183              AND expires_at_utc > SYSUTCDATETIME()
 184              AND (@lease_id IS NULL OR (lease_id = @lease_id AND lease_expires_at_utc > SYSUTCDATETIME()));
 185            """;
 186        command.Parameters.AddWithValue("@flow_id", flowId);
 187        command.Parameters.AddWithValue("@state_json", stateJson);
 188        command.Parameters.AddWithValue("@ttl_ms", DurableFlowStoreShared.ServerClockTtlMilliseconds(ttl));
 189        command.Parameters.AddWithValue("@expected_revision", expectedRevision);
 190        command.Parameters.AddWithValue("@new_revision", state.Revision);
 191        command.Parameters.AddWithValue("@lease_id", (object?)leaseId ?? DBNull.Value);
 192        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 193    }
 194
 195    public Task<bool> TryAcquireLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken canc
 196        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: true, cancellationToken);
 197
 198    public Task<bool> TryRenewLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken cancel
 199        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: false, cancellationToken);
 200
 201    public async Task ReleaseLeaseAsync(string flowId, string leaseId, CancellationToken cancellationToken = default)
 202    {
 203        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 204        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 205        await using var command = connection.CreateCommand();
 206        command.CommandText = $"UPDATE {Table} SET lease_id = NULL, lease_expires_at_utc = NULL WHERE flow_id = @flow_id
 207        command.Parameters.AddWithValue("@flow_id", flowId);
 208        command.Parameters.AddWithValue("@lease_id", leaseId);
 209        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 210    }
 211
 212    /// <inheritdoc />
 213    public async Task<FlowLeaseObservation?> ObserveLeaseAsync(string flowId, CancellationToken cancellationToken = defa
 214    {
 215        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 216        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.
 225        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 226        await using var command = connection.CreateCommand();
 227        command.CommandText = $"SELECT lease_id, lease_expires_at_utc FROM {Table} WHERE flow_id = @flow_id;";
 228        command.Parameters.AddWithValue("@flow_id", flowId);
 229
 230        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 231        if (!await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 232            return FlowLeaseObservation.Unheld;
 233
 234        return DurableFlowStoreShared.LeaseObservation(
 235            reader.IsDBNull(0) ? null : reader.GetString(0),
 236            reader.IsDBNull(1) ? null : reader.GetDateTime(1));
 237    }
 238
 239    public async Task<bool> TryDeleteAsync(string flowId, CancellationToken cancellationToken = default)
 240    {
 241        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 242        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 243
 244        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 245        await using var command = connection.CreateCommand();
 246        command.CommandText = $"DELETE FROM {Table} WHERE flow_id = @flow_id;";
 247        command.Parameters.AddWithValue("@flow_id", flowId);
 248        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 249    }
 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.
 258        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 259        await using var command = connection.CreateCommand();
 260        command.CommandText = $"DELETE TOP ({DurableFlowStoreShared.PruneBatchSize}) FROM {Table} WHERE expires_at_utc <
 261        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 262    }
 263
 264    private async Task EnsureCreatedAsync(CancellationToken cancellationToken)
 265    {
 266        if (_created)
 267            return;
 268
 269        await _ensureGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 270        try
 271        {
 272            if (_created)
 273                return;
 274
 275            await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 276
 277            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.
 286                if (!await ObjectExistsAsync(connection, cancellationToken).ConfigureAwait(false))
 287                    return;
 288
 289                await VerifyRelationsAsync(connection, transaction: null, cancellationToken).ConfigureAwait(false);
 290                _created = true;
 291                return;
 292            }
 293
 294            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.
 301            await using (var lockCommand = connection.CreateCommand())
 302            {
 303                lockCommand.Transaction = transaction;
 304                lockCommand.CommandText =
 305                    """
 306                    DECLARE @lock_result int;
 307                    EXEC @lock_result = sp_getapplock
 308                        @Resource = @lock_resource,
 309                        @LockMode = 'Exclusive',
 310                        @LockOwner = 'Transaction',
 311                        @LockTimeout = 60000;
 312                    IF @lock_result < 0
 313                        THROW 51000, N'Failed to acquire the AsyncResponse DDL application lock.', 1;
 314                    """;
 315                lockCommand.Parameters.AddWithValue("@lock_resource", DurableFlowStoreShared.SchemaLockResource(_options
 316                await lockCommand.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 317            }
 318
 319            await using var command = connection.CreateCommand();
 320            command.Transaction = transaction;
 321            command.CommandText =
 322                $"""
 323                IF SCHEMA_ID(N'{_options.SchemaName}') IS NULL
 324                    EXEC(N'CREATE SCHEMA {Quote(_options.SchemaName)}');
 325
 326                IF OBJECT_ID(N'{_options.SchemaName}.{_options.TableName}', N'U') IS NULL
 327                CREATE TABLE {Table} (
 328                    flow_id nvarchar(400) COLLATE Latin1_General_100_BIN2 NOT NULL PRIMARY KEY,
 329                    state_json nvarchar(max) NOT NULL,
 330                    expires_at_utc datetime2 NOT NULL,
 331                    updated_at_utc datetime2 NOT NULL,
 332                    -- Unnamed DEFAULT deliberately: the previous derived name prefixed "DF_" and
 333                    -- suffixed "_revision" onto the full table name, so a table name this store
 334                    -- otherwise accepts (117 characters) produced a 129-character constraint name
 335                    -- and the CREATE failed with error 103 — SQL Server's identifier cap is 128.
 336                    -- Nothing reads the constraint by name; the store never drops or alters it.
 337                    revision bigint NOT NULL DEFAULT 0,
 338                    lease_id nvarchar(64) NULL,
 339                    lease_expires_at_utc datetime2 NULL
 340                );
 341
 342                IF NOT EXISTS (SELECT 1 FROM sys.indexes WHERE name = N'{IndexName}' AND object_id = OBJECT_ID(N'{_optio
 343                    CREATE INDEX {Quote(IndexName)} ON {Table} (expires_at_utc);
 344                """;
 345            try
 346            {
 347                await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 348            }
 349            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.
 356                await SqlServerRelationVerifier.ThrowDiagnosedCollisionAsync(
 357                    OpenConnectionAsync,
 358                    ex,
 359                    _options.SchemaName,
 360                    "durable-flow",
 361                    ExpectedObjects(),
 362                    cancellationToken).ConfigureAwait(false);
 363                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.
 370            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.
 379            await VerifyRelationsAsync(connection, transaction: null, cancellationToken).ConfigureAwait(false);
 380            _created = true;
 381        }
 382        finally
 383        {
 384            _ensureGate.Release();
 385        }
 386    }
 387
 388    private Task VerifyRelationsAsync(SqlConnection connection, SqlTransaction? transaction, CancellationToken cancellat
 389        => SqlServerRelationVerifier.VerifyAsync(
 390            connection,
 391            transaction,
 392            _options.SchemaName,
 393            "durable-flow",
 394            ExpectedObjects(),
 395            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    {
 405        await using var command = connection.CreateCommand();
 406        command.CommandText =
 407            """
 408            SELECT CASE WHEN EXISTS (
 409                SELECT 1
 410                FROM sys.objects o
 411                JOIN sys.schemas s ON s.schema_id = o.schema_id
 412                WHERE s.name = @schema AND o.name = @table) THEN 1 ELSE 0 END;
 413            """;
 414        command.Parameters.AddWithValue("@schema", _options.SchemaName);
 415        command.Parameters.AddWithValue("@table", _options.TableName);
 416        return (int)(await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false))! == 1;
 417    }
 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() =>
 425            [
 426                new(_options.TableName, SqlServerObjectKind.Table,
 427                [
 428                    new("flow_id", "nvarchar(400)", Nullable: false, RequiresBinaryCollation: true),
 429                    new("state_json", "nvarchar(max)", Nullable: false),
 430                    new("expires_at_utc", "datetime2(7)", Nullable: false),
 431                    new("updated_at_utc", "datetime2(7)", Nullable: false),
 432                    new("revision", "bigint", Nullable: false),
 433                    new("lease_id", "nvarchar(64)", Nullable: true),
 434                    new("lease_expires_at_utc", "datetime2(7)", Nullable: true)
 435                ],
 436                PrimaryKey: ["flow_id"])
 437            ];
 438
 439
 440    private async Task<bool> UpdateLeaseAsync(
 441        string flowId,
 442        string leaseId,
 443        TimeSpan leaseDuration,
 444        bool acquire,
 445        CancellationToken cancellationToken)
 446    {
 447        DurableFlowStoreShared.ValidateLeaseArgs(flowId, leaseId, leaseDuration);
 448
 449        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 450        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 451        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.
 455        command.CommandText =
 456            $"""
 457            UPDATE {Table}
 458            SET lease_id = @lease_id, lease_expires_at_utc = {AddMilliseconds("@lease_ms")}
 459            WHERE flow_id = @flow_id
 460              AND expires_at_utc > SYSUTCDATETIME()
 461              AND {(acquire ? "(lease_id IS NULL OR lease_expires_at_utc <= SYSUTCDATETIME() OR lease_id = @lease_id)" :
 462            """;
 463        command.Parameters.AddWithValue("@flow_id", flowId);
 464        command.Parameters.AddWithValue("@lease_id", leaseId);
 465        command.Parameters.AddWithValue("@lease_ms", DurableFlowStoreShared.ServerClockTtlMilliseconds(leaseDuration));
 466        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 467    }
 468
 469    private Task<SqlConnection> OpenConnectionAsync(CancellationToken cancellationToken)
 470        => 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)
 480        => $"DATEADD(SECOND, CAST({parameterName} / 1000 AS int), DATEADD(MILLISECOND, CAST({parameterName} % 1000 AS in
 481
 482    private string Table => $"{Quote(_options.SchemaName)}.{Quote(_options.TableName)}";
 483    private string IndexName => DurableFlowStoreShared.DerivedName(_options.TableName, "_expires_idx", 128);
 484    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

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