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

Information
Class: Microsoft.Extensions.DependencyInjection.SqliteDurableFlowServiceCollectionExtensions
Assembly: AsyncResponse.DurableFlows.Sqlite
File(s): /_/src/DurableFlows/AsyncResponse.DurableFlows.Sqlite/SqliteDurableFlows.cs
Line coverage
100%
Covered lines: 2
Uncovered lines: 0
Coverable lines: 2
Total lines: 633
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
WithSqliteDurableFlows(...)100%11100%

File(s)

/_/src/DurableFlows/AsyncResponse.DurableFlows.Sqlite/SqliteDurableFlows.cs

#LineLine coverage
 1using AsyncResponse;
 2using AsyncResponse.DurableFlows.Internal;
 3using AsyncResponse.DurableFlows.Sqlite;
 4using Microsoft.Data.Sqlite;
 5using Microsoft.Extensions.DependencyInjection.Extensions;
 6using Microsoft.Extensions.Logging;
 7using Microsoft.Extensions.Options;
 8
 9namespace Microsoft.Extensions.DependencyInjection
 10{
 11    /// <summary>DI registration for the SQLite durable-flow state store.</summary>
 12    public static class SqliteDurableFlowServiceCollectionExtensions
 13    {
 14        /// <summary>Stores durable-flow state in SQLite.</summary>
 15        public static AsyncResponseRegistrationBuilder WithSqliteDurableFlows(
 16            this AsyncResponseRegistrationBuilder builder,
 17            Action<SqliteDurableFlowOptions>? configure = null)
 18        {
 19            // Singleton on purpose: schema provisioning is cached per store instance, and the
 20            // executor resolves the store from a fresh scope per flow execution — a scoped store
 21            // would re-run EnsureCreated's DDL round-trip on every run.
 21422            builder.Services.TryAddSingleton<SqliteFlowStateStore>();
 21423            return builder.WithDurableFlows<SqliteFlowStateStore, SqliteDurableFlowOptions>(configure);
 24        }
 25    }
 26}
 27
 28namespace AsyncResponse.DurableFlows.Sqlite
 29{
 30/// <summary>Options for the SQLite durable-flow state store.</summary>
 31public sealed class SqliteDurableFlowOptions : DurableFlowOptions
 32{
 33    /// <summary>SQLite connection string. Default: <c>Data Source=asyncresponse-flow-state.db</c>.</summary>
 34    public string ConnectionString { get; set; } = "Data Source=asyncresponse-flow-state.db";
 35
 36    /// <summary>Table storing one durable-flow ledger row per flow id.</summary>
 37    public string TableName { get; set; } = "asyncresponse_flow_state";
 38
 39    /// <summary>Creates the table and expiry index on first use.</summary>
 40    public bool AutoCreateSchema { get; set; } = true;
 41
 42    /// <summary>
 43    /// How often <see cref="SqliteFlowStateStore.TryCreateAsync"/> opportunistically deletes one bounded
 44    /// batch (1000 rows) of expired rows (loads already treat expired state as absent; pruning
 45    /// bounds table growth). Zero or negative prunes on every save. Default: 5 minutes.
 46    /// </summary>
 47    public TimeSpan PruneInterval { get; set; } = TimeSpan.FromMinutes(5);
 48
 49    /// <summary>
 50    /// Wall-clock budget one opportunistic prune may spend draining expired rows in batches of
 51    /// 1000 after its first batch (the first always runs). A single batch per interval capped
 52    /// cleanup at ~3 rows/second, which any busier instance outgrew forever; the prune now drains
 53    /// batches until one comes back short or this budget lapses, and reports the outcome on the
 54    /// <c>AsyncResponse</c> meter (<c>asyncresponse.flow_state.pruned_rows</c>,
 55    /// <c>prune_failures</c>, <c>prune_budget_exhausted</c>) and the store's logger. The create
 56    /// that triggers the prune waits for it, so this bounds that create's added latency. Zero
 57    /// keeps the historical single batch. Default: 2 seconds.
 58    /// </summary>
 59    public TimeSpan PruneBudget { get; set; } = DurableFlowStoreShared.DefaultPruneBudget;
 60
 61    /// <summary>
 62    /// Maximum serialized flow-state size in bytes accepted by writes; oversized ledgers fail fast
 63    /// with an actionable error instead of an opaque provider error. Default: <c>null</c>
 64    /// (unlimited — SQLite <c>TEXT</c> holds up to ~1 GB), settable as an operator budget.
 65    /// </summary>
 66    public long? MaxStateBytes { get; set; }
 67
 68    /// <summary>Validates option values and throws on misconfiguration.</summary>
 69    public void Validate()
 70    {
 71        DurableFlowStoreShared.ValidateConnectionString(ConnectionString, nameof(SqliteDurableFlowOptions));
 72        DurableFlowStoreShared.ValidateIdentifier(TableName, $"{nameof(SqliteDurableFlowOptions)}.{nameof(TableName)}", 
 73        DurableFlowStoreShared.ValidateMaxStateBytes(MaxStateBytes, nameof(SqliteDurableFlowOptions));
 74        DurableFlowStoreShared.ValidatePruneBudget(PruneBudget, nameof(SqliteDurableFlowOptions));
 75    }
 76}
 77
 78/// <summary>SQLite implementation of <see cref="IFlowStateStore"/>.</summary>
 79public sealed class SqliteFlowStateStore : IFlowStateStore
 80{
 81    private readonly ILogger<SqliteFlowStateStore>? _logger;
 82
 83    // Time authority: this store deliberately keeps the app clock (DateTime.UtcNow) for expiry
 84    // and lease comparisons. A SQLite database file lives on a single machine, and every writer
 85    // is a process on that machine sharing the same clock — the multi-node clock-skew hazard the
 86    // server-clock stores guard against cannot occur, and SQLite has no server clock to ask.
 87    private readonly SqliteDurableFlowOptions _options;
 88    private readonly SemaphoreSlim _ensureGate = new(1, 1);
 89
 90    // SQLite allows exactly one writer at a time, and its cross-connection busy handler is a
 91    // poll loop, not a queue: under heavy concurrency on a slow machine an unlucky writer can
 92    // lose every poll until the busy timeout expires ('database is locked' storms on 2-core CI
 93    // runners). Serializing this process's writers through a real FIFO gate costs no throughput
 94    // (they would serialize inside SQLite anyway) and makes in-process contention
 95    // starvation-free; the busy timeout then only covers cross-process writers. Reads stay
 96    // concurrent (WAL).
 97    private readonly SemaphoreSlim _writeGate = new(1, 1);
 98    private long _lastPruneTicks;
 99    private volatile bool _created;
 100
 101    public SqliteFlowStateStore(IOptions<SqliteDurableFlowOptions> options, ILogger<SqliteFlowStateStore>? logger = null
 102    {
 103        _logger = logger;
 104        _options = options.Value;
 105        _options.Validate();
 106    }
 107
 108    public async Task<FlowState?> LoadAsync(string flowId, CancellationToken cancellationToken = default)
 109    {
 110        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 111        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 112
 113        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 114        await using var command = connection.CreateCommand();
 115        command.CommandText =
 116            $"""
 117            SELECT state_json, revision
 118            FROM {Table}
 119            WHERE flow_id = $flow_id AND expires_at_utc > $now_utc;
 120            """;
 121        command.Parameters.AddWithValue("$flow_id", flowId);
 122        command.Parameters.AddWithValue("$now_utc", DateTime.UtcNow);
 123
 124        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 125        if (!await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 126            return null;
 127
 128        return DurableFlowStoreShared.ReadState(flowId, reader.GetString(0), reader.GetInt64(1));
 129    }
 130
 131    /// <inheritdoc />
 132    public void ValidateCreate(string flowId, FlowState state, TimeSpan ttl)
 133    {
 134        DurableFlowStoreShared.ValidateCreate(flowId, state, ttl);
 135        if (_options.MaxStateBytes is not null)
 136            _ = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "SQLite");
 137    }
 138
 139    public async Task<bool> TryCreateAsync(
 140        string flowId,
 141        FlowState state,
 142        TimeSpan ttl,
 143        CancellationToken cancellationToken = default)
 144    {
 145        DurableFlowStoreShared.ValidateCreate(flowId, state, ttl);
 146        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "SQLite");
 147        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 148        if (DurableFlowStoreShared.ShouldPrune(ref _lastPruneTicks, _options.PruneInterval))
 149            await DurableFlowStoreShared.PruneQuietlyAsync(() => PruneExpiredAsync(cancellationToken), _options.PruneBud
 150        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 151        await using var command = connection.CreateCommand();
 152        command.CommandText =
 153            $"""
 154            INSERT INTO {Table} (flow_id, state_json, expires_at_utc, updated_at_utc, revision)
 155            VALUES ($flow_id, $state_json, $expires_at_utc, $now_utc, $revision)
 156            ON CONFLICT(flow_id) DO UPDATE SET
 157                state_json = excluded.state_json,
 158                expires_at_utc = excluded.expires_at_utc,
 159                updated_at_utc = excluded.updated_at_utc,
 160                revision = excluded.revision,
 161                lease_id = NULL,
 162                lease_expires_at_utc = NULL
 163            WHERE {Table}.expires_at_utc <= $now_utc;
 164            """;
 165        var now = DateTime.UtcNow;
 166        command.Parameters.AddWithValue("$flow_id", flowId);
 167        command.Parameters.AddWithValue("$state_json", stateJson);
 168        command.Parameters.AddWithValue("$expires_at_utc", DurableFlowStoreShared.AddSaturating(now, ttl));
 169        command.Parameters.AddWithValue("$now_utc", now);
 170        command.Parameters.AddWithValue("$revision", state.Revision);
 171        return await ExecuteWriteAsync(command, cancellationToken).ConfigureAwait(false) > 0;
 172    }
 173
 174    public async Task<bool> TryUpdateAsync(
 175        string flowId,
 176        FlowState state,
 177        long expectedRevision,
 178        TimeSpan ttl,
 179        string? leaseId = null,
 180        CancellationToken cancellationToken = default)
 181    {
 182        DurableFlowStoreShared.ValidateUpdate(flowId, state, expectedRevision, ttl);
 183        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "SQLite");
 184        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 185        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 186        await using var command = connection.CreateCommand();
 187        var now = DateTime.UtcNow;
 188        command.CommandText =
 189            $"""
 190            UPDATE {Table}
 191            SET state_json = $state_json,
 192                expires_at_utc = $expires_at_utc,
 193                updated_at_utc = $updated_at_utc,
 194                revision = $new_revision
 195            WHERE flow_id = $flow_id
 196              AND revision = $expected_revision
 197              AND expires_at_utc > $now_utc
 198              AND ($lease_id IS NULL OR (lease_id = $lease_id AND lease_expires_at_utc > $now_utc));
 199            """;
 200        command.Parameters.AddWithValue("$flow_id", flowId);
 201        command.Parameters.AddWithValue("$state_json", stateJson);
 202        command.Parameters.AddWithValue("$expires_at_utc", DurableFlowStoreShared.AddSaturating(now, ttl));
 203        command.Parameters.AddWithValue("$updated_at_utc", now);
 204        command.Parameters.AddWithValue("$new_revision", state.Revision);
 205        command.Parameters.AddWithValue("$expected_revision", expectedRevision);
 206        command.Parameters.AddWithValue("$now_utc", now);
 207        command.Parameters.AddWithValue("$lease_id", (object?)leaseId ?? DBNull.Value);
 208        return await ExecuteWriteAsync(command, cancellationToken).ConfigureAwait(false) > 0;
 209    }
 210
 211    public Task<bool> TryAcquireLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken canc
 212        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: true, cancellationToken);
 213
 214    public Task<bool> TryRenewLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken cancel
 215        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: false, cancellationToken);
 216
 217    public async Task ReleaseLeaseAsync(string flowId, string leaseId, CancellationToken cancellationToken = default)
 218    {
 219        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 220        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 221        await using var command = connection.CreateCommand();
 222        command.CommandText = $"UPDATE {Table} SET lease_id = NULL, lease_expires_at_utc = NULL WHERE flow_id = $flow_id
 223        command.Parameters.AddWithValue("$flow_id", flowId);
 224        command.Parameters.AddWithValue("$lease_id", leaseId);
 225        await ExecuteWriteAsync(command, cancellationToken).ConfigureAwait(false);
 226    }
 227
 228    /// <inheritdoc />
 229    public async Task<FlowLeaseObservation?> ObserveLeaseAsync(string flowId, CancellationToken cancellationToken = defa
 230    {
 231        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 232        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 233
 234        // The two lease columns exactly as stored — deliberately no $now_utc predicate, unlike
 235        // every other statement in this store: an expired lease nobody has taken over must keep
 236        // reading as the same lease, because the engine's proof of a live holder is that two
 237        // observations DIFFER. Whether it has lapsed stays UpdateLeaseAsync's call. A read, so it
 238        // stays outside the write gate (WAL readers never block on the writer).
 239        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 240        await using var command = connection.CreateCommand();
 241        command.CommandText = $"SELECT lease_id, lease_expires_at_utc FROM {Table} WHERE flow_id = $flow_id;";
 242        command.Parameters.AddWithValue("$flow_id", flowId);
 243
 244        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 245        if (!await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 246            return FlowLeaseObservation.Unheld;
 247
 248        // UpdateLeaseAsync binds a UTC DateTime, which Microsoft.Data.Sqlite stores as zone-less
 249        // ISO-8601 TEXT with all seven fractional digits; GetDateTime parses it back tick for tick
 250        // as Unspecified, and the shared shaper stamps it UTC.
 251        return DurableFlowStoreShared.LeaseObservation(
 252            reader.IsDBNull(0) ? null : reader.GetString(0),
 253            reader.IsDBNull(1) ? null : reader.GetDateTime(1));
 254    }
 255
 256    public async Task<bool> TryDeleteAsync(string flowId, CancellationToken cancellationToken = default)
 257    {
 258        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 259        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 260
 261        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 262        await using var command = connection.CreateCommand();
 263        command.CommandText = $"DELETE FROM {Table} WHERE flow_id = $flow_id;";
 264        command.Parameters.AddWithValue("$flow_id", flowId);
 265        return await ExecuteWriteAsync(command, cancellationToken).ConfigureAwait(false) > 0;
 266    }
 267
 268    private async Task<int> PruneExpiredAsync(CancellationToken cancellationToken)
 269    {
 270        // Timestamps are stored as ISO-8601 TEXT, which compares correctly lexicographically.
 271        // One bounded batch per call; DurableFlowStoreShared.PruneQuietlyAsync repeats it under
 272        // the PruneBudget while batches come back full (policy shared by all relational stores): an
 273        // unbatched DELETE over a large expired backlog holds the single SQLite write lock for
 274        // the whole sweep. Loads already filter on expiry, so any backlog beyond the batch just
 275        // waits for the next interval. Id-subquery form because DELETE ... LIMIT needs a
 276        // non-default SQLite compile flag.
 277        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 278        await using var command = connection.CreateCommand();
 279        command.CommandText =
 280            $"""
 281            DELETE FROM {Table}
 282            WHERE flow_id IN (SELECT flow_id FROM {Table} WHERE expires_at_utc <= $now_utc LIMIT {DurableFlowStoreShared
 283            """;
 284        command.Parameters.AddWithValue("$now_utc", DateTime.UtcNow);
 285        return await ExecuteWriteAsync(command, cancellationToken).ConfigureAwait(false);
 286    }
 287
 288    private async Task EnsureCreatedAsync(CancellationToken cancellationToken)
 289    {
 290        if (_created)
 291            return;
 292
 293        await _ensureGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 294        try
 295        {
 296            if (_created)
 297                return;
 298
 299            await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 300            if (_options.AutoCreateSchema)
 301            {
 302                await using var command = connection.CreateCommand();
 303                command.CommandText =
 304                    $"""
 305                    -- WAL is the right journal mode for this store's use case (concurrent flow
 306                    -- executors on one node): readers never block behind a writer, which rollback
 307                    -- journal mode does not guarantee — concurrent load/save storms on slow disks
 308                    -- surface as SQLITE_BUSY 'database is locked' there. The mode is persistent in
 309                    -- the database file, so setting it alongside the schema costs nothing per
 310                    -- operation. Manually-provisioned databases (AutoCreateSchema=false) should set
 311                    -- it themselves — see docs/durable-flow-state-stores.md.
 312                    PRAGMA journal_mode=WAL;
 313                    CREATE TABLE IF NOT EXISTS {Table} (
 314                        flow_id TEXT NOT NULL PRIMARY KEY,
 315                        state_json TEXT NOT NULL,
 316                        expires_at_utc TEXT NOT NULL,
 317                        updated_at_utc TEXT NOT NULL,
 318                        revision INTEGER NOT NULL DEFAULT 0,
 319                        lease_id TEXT NULL,
 320                        lease_expires_at_utc TEXT NULL
 321                    );
 322                    CREATE INDEX IF NOT EXISTS {IndexName} ON {Table} (expires_at_utc);
 323                    """;
 324                await ExecuteWriteAsync(command, cancellationToken).ConfigureAwait(false);
 325
 326                // Fall THROUGH to verification instead of latching here. CREATE TABLE IF NOT EXISTS
 327                // is a no-op against a table an earlier build or a hand-run migration left behind,
 328                // so latching on the DDL trusted its shape for the process lifetime — none of the
 329                // load-bearing checks below (a single-column NOT NULL primary key, ISO-8601 text
 330                // affinity on the expiry and lease columns, no extra NOT NULL column, a BINARY
 331                // flow_id collation) ever ran on the DEFAULT path. MySQL, PostgreSQL and SQL Server
 332                // all verify after their DDL for exactly this reason.
 333                _created = await VerifyFlowTableAsync(connection, cancellationToken).ConfigureAwait(false);
 334                return;
 335            }
 336
 337            // Operator-provisioned schema: nothing on this path issues DDL (not even the WAL
 338            // pragma — see the note above), but the table's shape is still verified before the
 339            // store trusts it. Latch only when the table was actually verified (MySQL/Oracle
 340            // parity): an absent table must keep re-verifying, or a migration that lands AFTER
 341            // the first operation would never have its shape checked for the process lifetime.
 342            _created = await VerifyFlowTableAsync(connection, cancellationToken).ConfigureAwait(false);
 343        }
 344        finally
 345        {
 346            _ensureGate.Release();
 347        }
 348    }
 349
 350    /// <summary>
 351    /// Checks the operator-provisioned table against the shape this store reads and writes.
 352    /// Declared types are compared by SQLite AFFINITY, not spelling, so any declaration that
 353    /// behaves like the documented DDL passes. Two properties are load-bearing and misfire
 354    /// SILENTLY or at the first flow — the wrong end of the deployment — when absent:
 355    /// <list type="bullet">
 356    /// <item><description>
 357    /// A single-column PRIMARY KEY on flow_id. <see cref="TryCreateAsync"/>'s upsert targets
 358    /// <c>ON CONFLICT(flow_id)</c>, which requires a uniqueness constraint on exactly that
 359    /// column; and SQLite's historical quirk admits NULL keys when the PRIMARY KEY column is
 360    /// not also declared NOT NULL.
 361    /// </description></item>
 362    /// <item><description>
 363    /// TEXT affinity on the timestamp columns. Expiry and lease fencing compare ISO-8601
 364    /// strings lexicographically; a numeric affinity coerces digit-only values and breaks that
 365    /// ordering.
 366    /// </description></item>
 367    /// </list>
 368    /// </summary>
 369    private async Task<bool> VerifyFlowTableAsync(SqliteConnection connection, CancellationToken cancellationToken)
 370    {
 371        var columns = new Dictionary<string, ActualColumn>(StringComparer.OrdinalIgnoreCase);
 372        await using (var command = connection.CreateCommand())
 373        {
 374            // Read-only: table_info reports name, declared type, NOT NULL, default, and the
 375            // primary-key ordinal. Generated columns are not listed — exactly right for the
 376            // extra-column check below, because they fill themselves in on insert.
 377            command.CommandText = $"PRAGMA table_info({Table});";
 378            await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 379            while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 380            {
 381                columns[reader.GetString(1)] = new ActualColumn(
 382                    DeclaredType: reader.GetString(2),
 383                    NotNull: reader.GetInt64(3) != 0,
 384                    HasDefault: !reader.IsDBNull(4),
 385                    PrimaryKeyOrdinal: reader.GetInt64(5));
 386            }
 387        }
 388
 389        if (columns.Count == 0)
 390        {
 391            // The table does not exist: AutoCreateSchema = false and the migration has not run
 392            // yet. That surfaces at the first query with a clear SQLite error, and failing here
 393            // would break the documented "create it yourself, later" workflow. Returning false
 394            // keeps _created unlatched so the next operation re-verifies once the migration has
 395            // run.
 396            return false;
 397        }
 398
 399        foreach (var expected in ExpectedColumns)
 400        {
 401            if (!columns.TryGetValue(expected.Name, out var actual))
 402            {
 403                throw new InvalidOperationException(
 404                    $"The SQLite durable-flow table '{_options.TableName}' has no '{expected.Name}' column. It was creat
 405                    "earlier build or by hand and does not match the shape this store reads and writes " +
 406                    $"({string.Join(", ", ExpectedColumns.Select(column => $"{column.Name} {column.Declaration}"))}). Re
 407                    "with the DDL in docs/durable-flow-state-stores.md (tables this build creates get that shape automat
 408            }
 409
 410            if (expected.Mismatch(actual) is { } mismatch)
 411            {
 412                throw new InvalidOperationException(
 413                    $"The SQLite durable-flow table '{_options.TableName}' declares {expected.Name} as " +
 414                    $"'{(actual.DeclaredType.Length == 0 ? "(no type)" : actual.DeclaredType)}{(actual.NotNull ? " NOT N
 415                    $"which {mismatch}. This store needs {expected.Name} {expected.Declaration}. SQLite cannot alter a c
 416                    "place — re-create the table with the DDL in docs/durable-flow-state-stores.md (tables this build cr
 417                    "that shape automatically).");
 418            }
 419        }
 420
 421        // Exactly PRIMARY KEY (flow_id): the create upsert's ON CONFLICT(flow_id) needs a
 422        // uniqueness constraint on that column alone — a composite key constrains a different
 423        // tuple, so SQLite rejects the upsert at the first flow rather than at startup.
 424        if (columns["flow_id"].PrimaryKeyOrdinal != 1 || columns.Values.Count(column => column.PrimaryKeyOrdinal != 0) !
 425        {
 426            throw new InvalidOperationException(
 427                $"The SQLite durable-flow table '{_options.TableName}' does not declare PRIMARY KEY (flow_id) on that co
 428                "Starting a flow is an insert-if-absent targeting ON CONFLICT(flow_id), which requires a uniqueness cons
 429                "exactly flow_id — without it every flow creation fails, and a composite key admits duplicate ids. Re-cr
 430                "table with the DDL in docs/durable-flow-state-stores.md (tables this build creates declare it automatic
 431        }
 432
 433        // Columns this store never names in an INSERT. One that the database cannot fill in for
 434        // itself makes EVERY create fail — the shape is otherwise perfect, so the failure arrives
 435        // at the first flow rather than at startup. Generated columns never reach this loop; so
 436        // is anything nullable or defaulted.
 437        foreach (var (name, actual) in columns)
 438        {
 439            if (ExpectedColumns.Any(expected => string.Equals(expected.Name, name, StringComparison.OrdinalIgnoreCase))
 440                || !actual.NotNull
 441                || actual.HasDefault)
 442            {
 443                continue;
 444            }
 445
 446            throw new InvalidOperationException(
 447                $"The SQLite durable-flow table '{_options.TableName}' has an extra column '{name}' ({actual.DeclaredTyp
 448                "with no default. This store writes only its own columns, so every flow creation would fail on that colu
 449                "it a default, make it nullable or generated, or move it to a table of your own.");
 450        }
 451
 452        await VerifyFlowIdCollationAsync(connection, cancellationToken).ConfigureAwait(false);
 453
 454        return true;
 455    }
 456
 457    /// <summary>
 458    /// flow_id must compare ordinally. PRAGMA table_info does not report a column's collation, so
 459    /// the declaration is read from sqlite_master instead — the one property of this table that a
 460    /// shape check cannot see and that fails SILENTLY: under COLLATE NOCASE the primary key and
 461    /// every <c>WHERE flow_id = $flow_id</c> fold case, so "Order-A1" and "order-a1" become one
 462    /// key. The second flow's insert-if-absent then reports "already running" and a load for one
 463    /// id returns the other run's ledger. SQLite is the only one of the six relational stores that
 464    /// was not checking this; MySQL, PostgreSQL, SQL Server, Oracle and EF Core all do.
 465    /// <para>
 466    /// The lookup matches the stored table name case-insensitively — <c>sqlite_master.name</c>
 467    /// compares BINARY while SQLite resolves identifiers case-insensitively everywhere else, so a
 468    /// case-variant table silently no-opped this whole check. And EVERY identifier-boundary
 469    /// occurrence of <c>flow_id</c> is inspected, not the first substring hit: an earlier column
 470    /// ending in flow_id (<c>parent_flow_id</c>) captured the match and hid the real column's
 471    /// collation, and a table-level <c>PRIMARY KEY (flow_id COLLATE NOCASE)</c> — which the docs
 472    /// promise is caught — was never reached.
 473    /// </para>
 474    /// </summary>
 475    private async Task VerifyFlowIdCollationAsync(SqliteConnection connection, CancellationToken cancellationToken)
 476    {
 477        await using var command = connection.CreateCommand();
 478        command.CommandText = "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = $name COLLATE NOCASE;";
 479        command.Parameters.AddWithValue("$name", _options.TableName);
 480
 481        var ddl = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false) as string;
 482        if (string.IsNullOrEmpty(ddl))
 483            return;
 484
 485        // Each flow_id occurrence's clause, up to the next column separator: the column
 486        // declaration carries a column-level COLLATE, the table-level PRIMARY KEY clause a
 487        // key-level one. Occurrences without a COLLATE (an FK column list, a plain PK clause)
 488        // are skipped.
 489        for (var start = IndexOfFlowIdIdentifier(ddl, 0); start >= 0; start = IndexOfFlowIdIdentifier(ddl, start + 1))
 490        {
 491            var end = ddl.IndexOf(',', start);
 492            var declaration = end < 0 ? ddl[start..] : ddl[start..end];
 493
 494            var collate = declaration.IndexOf("COLLATE", StringComparison.OrdinalIgnoreCase);
 495            if (collate < 0)
 496                continue;
 497
 498            var tail = declaration[(collate + "COLLATE".Length)..].TrimStart();
 499            var tokenEnd = 0;
 500            while (tokenEnd < tail.Length && !char.IsWhiteSpace(tail[tokenEnd]) && tail[tokenEnd] is not (')' or ','))
 501                tokenEnd++;
 502
 503            var collation = tail[..tokenEnd].Trim('"');
 504            if (collation.Length == 0 || string.Equals(collation, "BINARY", StringComparison.OrdinalIgnoreCase))
 505                continue;
 506
 507            throw new InvalidOperationException(
 508                $"The SQLite durable-flow table '{_options.TableName}' declares flow_id with COLLATE {collation}. Flow i
 509                "ordinally by this library, so a case- or accent-insensitive collation makes ids differing only in case 
 510                "primary key: the second flow fails to start and a load returns the other run's state. Re-create the tab
 511                "using the default BINARY collation (the DDL in docs/durable-flow-state-stores.md, which tables this bui
 512        }
 513    }
 514
 515    /// <summary>
 516    /// Finds the next occurrence of <c>flow_id</c> that is a whole identifier — not the tail of
 517    /// <c>parent_flow_id</c> or the head of <c>flow_id_shadow</c>. Quoted forms ("flow_id",
 518    /// [flow_id], `flow_id`) satisfy the boundary test through their quote characters.
 519    /// </summary>
 520    private static int IndexOfFlowIdIdentifier(string ddl, int startIndex)
 521    {
 522        for (var index = ddl.IndexOf("flow_id", startIndex, StringComparison.OrdinalIgnoreCase);
 523             index >= 0;
 524             index = index + 1 < ddl.Length ? ddl.IndexOf("flow_id", index + 1, StringComparison.OrdinalIgnoreCase) : -1
 525        {
 526            var beforeIsIdentifierChar = index > 0 && (char.IsAsciiLetterOrDigit(ddl[index - 1]) || ddl[index - 1] == '_
 527            var afterIndex = index + "flow_id".Length;
 528            var afterIsIdentifierChar = afterIndex < ddl.Length && (char.IsAsciiLetterOrDigit(ddl[afterIndex]) || ddl[af
 529            if (!beforeIsIdentifierChar && !afterIsIdentifierChar)
 530                return index;
 531        }
 532
 533        return -1;
 534    }
 535
 536    /// <summary>SQLite column-affinity rules (in the documented precedence order).</summary>
 537    private static string Affinity(string declaredType)
 538    {
 539        if (declaredType.Contains("INT", StringComparison.OrdinalIgnoreCase))
 540            return "INTEGER";
 541        if (declaredType.Contains("CHAR", StringComparison.OrdinalIgnoreCase)
 542            || declaredType.Contains("CLOB", StringComparison.OrdinalIgnoreCase)
 543            || declaredType.Contains("TEXT", StringComparison.OrdinalIgnoreCase))
 544        {
 545            return "TEXT";
 546        }
 547
 548        if (declaredType.Length == 0 || declaredType.Contains("BLOB", StringComparison.OrdinalIgnoreCase))
 549            return "BLOB";
 550        if (declaredType.Contains("REAL", StringComparison.OrdinalIgnoreCase)
 551            || declaredType.Contains("FLOA", StringComparison.OrdinalIgnoreCase)
 552            || declaredType.Contains("DOUB", StringComparison.OrdinalIgnoreCase))
 553        {
 554            return "REAL";
 555        }
 556
 557        return "NUMERIC";
 558    }
 559
 560    private sealed record ActualColumn(string DeclaredType, bool NotNull, bool HasDefault, long PrimaryKeyOrdinal);
 561
 562    private sealed record ExpectedColumn(string Name, string Declaration, string RequiredAffinity, bool NotNull)
 563    {
 564        public string? Mismatch(ActualColumn actual)
 565        {
 566            if (!string.Equals(Affinity(actual.DeclaredType), RequiredAffinity, StringComparison.Ordinal))
 567                return $"resolves to {Affinity(actual.DeclaredType)} affinity where {RequiredAffinity} is required";
 568            if (actual.NotNull != NotNull)
 569                return NotNull ? "must be NOT NULL" : "must be nullable (this store writes and clears NULL there)";
 570            return null;
 571        }
 572    }
 573
 574    private static readonly ExpectedColumn[] ExpectedColumns =
 575    [
 576        new("flow_id", "TEXT NOT NULL PRIMARY KEY", "TEXT", NotNull: true),
 577        new("state_json", "TEXT NOT NULL", "TEXT", NotNull: true),
 578        new("expires_at_utc", "TEXT NOT NULL", "TEXT", NotNull: true),
 579        new("updated_at_utc", "TEXT NOT NULL", "TEXT", NotNull: true),
 580        new("revision", "INTEGER NOT NULL DEFAULT 0", "INTEGER", NotNull: true),
 581        new("lease_id", "TEXT NULL", "TEXT", NotNull: false),
 582        new("lease_expires_at_utc", "TEXT NULL", "TEXT", NotNull: false)
 583    ];
 584
 585    private async Task<bool> UpdateLeaseAsync(
 586        string flowId,
 587        string leaseId,
 588        TimeSpan leaseDuration,
 589        bool acquire,
 590        CancellationToken cancellationToken)
 591    {
 592        DurableFlowStoreShared.ValidateLeaseArgs(flowId, leaseId, leaseDuration);
 593
 594        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 595        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 596        await using var command = connection.CreateCommand();
 597        var now = DateTime.UtcNow;
 598        command.CommandText =
 599            $"""
 600            UPDATE {Table}
 601            SET lease_id = $lease_id, lease_expires_at_utc = $lease_expires_at_utc
 602            WHERE flow_id = $flow_id
 603              AND expires_at_utc > $now_utc
 604              AND {(acquire ? "(lease_id IS NULL OR lease_expires_at_utc <= $now_utc OR lease_id = $lease_id)" : "lease_
 605            """;
 606        command.Parameters.AddWithValue("$flow_id", flowId);
 607        command.Parameters.AddWithValue("$lease_id", leaseId);
 608        command.Parameters.AddWithValue("$lease_expires_at_utc", DurableFlowStoreShared.AddSaturating(now, leaseDuration
 609        command.Parameters.AddWithValue("$now_utc", now);
 610        return await ExecuteWriteAsync(command, cancellationToken).ConfigureAwait(false) > 0;
 611    }
 612
 613    private async Task<int> ExecuteWriteAsync(SqliteCommand command, CancellationToken cancellationToken)
 614    {
 615        await _writeGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 616        try
 617        {
 618            return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 619        }
 620        finally
 621        {
 622            _writeGate.Release();
 623        }
 624    }
 625
 626    private Task<SqliteConnection> OpenConnectionAsync(CancellationToken cancellationToken)
 627        => DurableFlowStoreShared.OpenConnectionAsync<SqliteConnection>(_options.ConnectionString, cancellationToken);
 628
 629    private string Table => Quote(_options.TableName);
 630    private string IndexName => Quote($"{_options.TableName}_expires_idx");
 631    private static string Quote(string identifier) => "\"" + identifier.Replace("\"", "\"\"", StringComparison.Ordinal) 
 632}
 633}
Generated by: ReportGenerator 5.5.11.0
09/22/2026 - 09:41:04
GitHub | reportgenerator.io

Methods/Properties

WithSqliteDurableFlows(Microsoft.Extensions.DependencyInjection.AsyncResponseRegistrationBuilder,System.Action`1<AsyncResponse.DurableFlows.Sqlite.SqliteDurableFlowOptions>)