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

Information
Class: AsyncResponse.DurableFlows.Sqlite.SqliteFlowStateStore
Assembly: AsyncResponse.DurableFlows.Sqlite
File(s): /_/src/DurableFlows/AsyncResponse.DurableFlows.Sqlite/SqliteDurableFlows.cs
Line coverage
98%
Covered lines: 291
Uncovered lines: 3
Coverable lines: 294
Total lines: 633
Line coverage: 98.9%
Branch coverage
87%
Covered branches: 93
Total branches: 106
Branch coverage: 87.7%
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(...)100%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()100%8897.29%
VerifyFlowTableAsync()92.3%2626100%
VerifyFlowIdCollationAsync()100%2222100%
IndexOfFlowIdIdentifier(...)68.75%1616100%
Affinity(...)72.22%191884.61%
get_DeclaredType()100%11100%
get_Name()100%11100%
Mismatch(...)83.33%66100%
.cctor()100%11100%
UpdateLeaseAsync()100%22100%
ExecuteWriteAsync()100%11100%
OpenConnectionAsync(...)100%11100%
get_Table()100%11100%
get_IndexName()100%11100%
Quote(...)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.
 22            builder.Services.TryAddSingleton<SqliteFlowStateStore>();
 23            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;
 27788    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).
 27797    private readonly SemaphoreSlim _writeGate = new(1, 1);
 98    private long _lastPruneTicks;
 99    private volatile bool _created;
 100
 277101    public SqliteFlowStateStore(IOptions<SqliteDurableFlowOptions> options, ILogger<SqliteFlowStateStore>? logger = null
 102    {
 277103        _logger = logger;
 277104        _options = options.Value;
 277105        _options.Validate();
 277106    }
 107
 108    public async Task<FlowState?> LoadAsync(string flowId, CancellationToken cancellationToken = default)
 109    {
 1543110        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 1543111        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 112
 1539113        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1539114        await using var command = connection.CreateCommand();
 1539115        command.CommandText =
 1539116            $"""
 1539117            SELECT state_json, revision
 1539118            FROM {Table}
 1539119            WHERE flow_id = $flow_id AND expires_at_utc > $now_utc;
 1539120            """;
 1539121        command.Parameters.AddWithValue("$flow_id", flowId);
 1539122        command.Parameters.AddWithValue("$now_utc", DateTime.UtcNow);
 123
 1539124        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 1539125        if (!await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 410126            return null;
 127
 1129128        return DurableFlowStoreShared.ReadState(flowId, reader.GetString(0), reader.GetInt64(1));
 1537129    }
 130
 131    /// <inheritdoc />
 132    public void ValidateCreate(string flowId, FlowState state, TimeSpan ttl)
 133    {
 148134        DurableFlowStoreShared.ValidateCreate(flowId, state, ttl);
 148135        if (_options.MaxStateBytes is not null)
 8136            _ = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "SQLite");
 144137    }
 138
 139    public async Task<bool> TryCreateAsync(
 140        string flowId,
 141        FlowState state,
 142        TimeSpan ttl,
 143        CancellationToken cancellationToken = default)
 144    {
 5231145        DurableFlowStoreShared.ValidateCreate(flowId, state, ttl);
 5230146        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "SQLite");
 5230147        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 5210148        if (DurableFlowStoreShared.ShouldPrune(ref _lastPruneTicks, _options.PruneInterval))
 370149            await DurableFlowStoreShared.PruneQuietlyAsync(() => PruneExpiredAsync(cancellationToken), _options.PruneBud
 5210150        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 5210151        await using var command = connection.CreateCommand();
 5210152        command.CommandText =
 5210153            $"""
 5210154            INSERT INTO {Table} (flow_id, state_json, expires_at_utc, updated_at_utc, revision)
 5210155            VALUES ($flow_id, $state_json, $expires_at_utc, $now_utc, $revision)
 5210156            ON CONFLICT(flow_id) DO UPDATE SET
 5210157                state_json = excluded.state_json,
 5210158                expires_at_utc = excluded.expires_at_utc,
 5210159                updated_at_utc = excluded.updated_at_utc,
 5210160                revision = excluded.revision,
 5210161                lease_id = NULL,
 5210162                lease_expires_at_utc = NULL
 5210163            WHERE {Table}.expires_at_utc <= $now_utc;
 5210164            """;
 5210165        var now = DateTime.UtcNow;
 5210166        command.Parameters.AddWithValue("$flow_id", flowId);
 5210167        command.Parameters.AddWithValue("$state_json", stateJson);
 5210168        command.Parameters.AddWithValue("$expires_at_utc", DurableFlowStoreShared.AddSaturating(now, ttl));
 5210169        command.Parameters.AddWithValue("$now_utc", now);
 5210170        command.Parameters.AddWithValue("$revision", state.Revision);
 5210171        return await ExecuteWriteAsync(command, cancellationToken).ConfigureAwait(false) > 0;
 5206172    }
 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    {
 921182        DurableFlowStoreShared.ValidateUpdate(flowId, state, expectedRevision, ttl);
 921183        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "SQLite");
 919184        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 919185        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 919186        await using var command = connection.CreateCommand();
 919187        var now = DateTime.UtcNow;
 919188        command.CommandText =
 919189            $"""
 919190            UPDATE {Table}
 919191            SET state_json = $state_json,
 919192                expires_at_utc = $expires_at_utc,
 919193                updated_at_utc = $updated_at_utc,
 919194                revision = $new_revision
 919195            WHERE flow_id = $flow_id
 919196              AND revision = $expected_revision
 919197              AND expires_at_utc > $now_utc
 919198              AND ($lease_id IS NULL OR (lease_id = $lease_id AND lease_expires_at_utc > $now_utc));
 919199            """;
 919200        command.Parameters.AddWithValue("$flow_id", flowId);
 919201        command.Parameters.AddWithValue("$state_json", stateJson);
 919202        command.Parameters.AddWithValue("$expires_at_utc", DurableFlowStoreShared.AddSaturating(now, ttl));
 919203        command.Parameters.AddWithValue("$updated_at_utc", now);
 919204        command.Parameters.AddWithValue("$new_revision", state.Revision);
 919205        command.Parameters.AddWithValue("$expected_revision", expectedRevision);
 919206        command.Parameters.AddWithValue("$now_utc", now);
 919207        command.Parameters.AddWithValue("$lease_id", (object?)leaseId ?? DBNull.Value);
 919208        return await ExecuteWriteAsync(command, cancellationToken).ConfigureAwait(false) > 0;
 919209    }
 210
 211    public Task<bool> TryAcquireLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken canc
 180212        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: true, cancellationToken);
 213
 214    public Task<bool> TryRenewLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken cancel
 19215        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: false, cancellationToken);
 216
 217    public async Task ReleaseLeaseAsync(string flowId, string leaseId, CancellationToken cancellationToken = default)
 218    {
 154219        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 154220        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 154221        await using var command = connection.CreateCommand();
 154222        command.CommandText = $"UPDATE {Table} SET lease_id = NULL, lease_expires_at_utc = NULL WHERE flow_id = $flow_id
 154223        command.Parameters.AddWithValue("$flow_id", flowId);
 154224        command.Parameters.AddWithValue("$lease_id", leaseId);
 154225        await ExecuteWriteAsync(command, cancellationToken).ConfigureAwait(false);
 154226    }
 227
 228    /// <inheritdoc />
 229    public async Task<FlowLeaseObservation?> ObserveLeaseAsync(string flowId, CancellationToken cancellationToken = defa
 230    {
 52231        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 46232        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).
 44239        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 42240        await using var command = connection.CreateCommand();
 42241        command.CommandText = $"SELECT lease_id, lease_expires_at_utc FROM {Table} WHERE flow_id = $flow_id;";
 42242        command.Parameters.AddWithValue("$flow_id", flowId);
 243
 42244        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 42245        if (!await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 8246            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.
 34251        return DurableFlowStoreShared.LeaseObservation(
 34252            reader.IsDBNull(0) ? null : reader.GetString(0),
 34253            reader.IsDBNull(1) ? null : reader.GetDateTime(1));
 42254    }
 255
 256    public async Task<bool> TryDeleteAsync(string flowId, CancellationToken cancellationToken = default)
 257    {
 416258        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 416259        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 260
 416261        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 416262        await using var command = connection.CreateCommand();
 416263        command.CommandText = $"DELETE FROM {Table} WHERE flow_id = $flow_id;";
 416264        command.Parameters.AddWithValue("$flow_id", flowId);
 416265        return await ExecuteWriteAsync(command, cancellationToken).ConfigureAwait(false) > 0;
 416266    }
 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.
 187277        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 187278        await using var command = connection.CreateCommand();
 187279        command.CommandText =
 187280            $"""
 187281            DELETE FROM {Table}
 187282            WHERE flow_id IN (SELECT flow_id FROM {Table} WHERE expires_at_utc <= $now_utc LIMIT {DurableFlowStoreShared
 187283            """;
 187284        command.Parameters.AddWithValue("$now_utc", DateTime.UtcNow);
 187285        return await ExecuteWriteAsync(command, cancellationToken).ConfigureAwait(false);
 181286    }
 287
 288    private async Task EnsureCreatedAsync(CancellationToken cancellationToken)
 289    {
 8505290        if (_created)
 8181291            return;
 292
 324293        await _ensureGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 294        try
 295        {
 322296            if (_created)
 117297                return;
 298
 205299            await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 203300            if (_options.AutoCreateSchema)
 301            {
 177302                await using var command = connection.CreateCommand();
 177303                command.CommandText =
 177304                    $"""
 177305                    -- WAL is the right journal mode for this store's use case (concurrent flow
 177306                    -- executors on one node): readers never block behind a writer, which rollback
 177307                    -- journal mode does not guarantee — concurrent load/save storms on slow disks
 177308                    -- surface as SQLITE_BUSY 'database is locked' there. The mode is persistent in
 177309                    -- the database file, so setting it alongside the schema costs nothing per
 177310                    -- operation. Manually-provisioned databases (AutoCreateSchema=false) should set
 177311                    -- it themselves — see docs/durable-flow-state-stores.md.
 177312                    PRAGMA journal_mode=WAL;
 177313                    CREATE TABLE IF NOT EXISTS {Table} (
 177314                        flow_id TEXT NOT NULL PRIMARY KEY,
 177315                        state_json TEXT NOT NULL,
 177316                        expires_at_utc TEXT NOT NULL,
 177317                        updated_at_utc TEXT NOT NULL,
 177318                        revision INTEGER NOT NULL DEFAULT 0,
 177319                        lease_id TEXT NULL,
 177320                        lease_expires_at_utc TEXT NULL
 177321                    );
 177322                    CREATE INDEX IF NOT EXISTS {IndexName} ON {Table} (expires_at_utc);
 177323                    """;
 177324                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.
 177333                _created = await VerifyFlowTableAsync(connection, cancellationToken).ConfigureAwait(false);
 173334                return;
 0335            }
 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.
 26342            _created = await VerifyFlowTableAsync(connection, cancellationToken).ConfigureAwait(false);
 8343        }
 344        finally
 345        {
 322346            _ensureGate.Release();
 347        }
 8479348    }
 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    {
 203371        var columns = new Dictionary<string, ActualColumn>(StringComparer.OrdinalIgnoreCase);
 203372        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.
 203377            command.CommandText = $"PRAGMA table_info({Table});";
 203378            await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 1592379            while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 380            {
 1389381                columns[reader.GetString(1)] = new ActualColumn(
 1389382                    DeclaredType: reader.GetString(2),
 1389383                    NotNull: reader.GetInt64(3) != 0,
 1389384                    HasDefault: !reader.IsDBNull(4),
 1389385                    PrimaryKeyOrdinal: reader.GetInt64(5));
 386            }
 203387        }
 388
 203389        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.
 4396            return false;
 397        }
 398
 3140399        foreach (var expected in ExpectedColumns)
 400        {
 1375401            if (!columns.TryGetValue(expected.Name, out var actual))
 402            {
 4403                throw new InvalidOperationException(
 4404                    $"The SQLite durable-flow table '{_options.TableName}' has no '{expected.Name}' column. It was creat
 4405                    "earlier build or by hand and does not match the shape this store reads and writes " +
 28406                    $"({string.Join(", ", ExpectedColumns.Select(column => $"{column.Name} {column.Declaration}"))}). Re
 4407                    "with the DDL in docs/durable-flow-state-stores.md (tables this build creates get that shape automat
 408            }
 409
 1371410            if (expected.Mismatch(actual) is { } mismatch)
 411            {
 4412                throw new InvalidOperationException(
 4413                    $"The SQLite durable-flow table '{_options.TableName}' declares {expected.Name} as " +
 4414                    $"'{(actual.DeclaredType.Length == 0 ? "(no type)" : actual.DeclaredType)}{(actual.NotNull ? " NOT N
 4415                    $"which {mismatch}. This store needs {expected.Name} {expected.Declaration}. SQLite cannot alter a c
 4416                    "place — re-create the table with the DDL in docs/durable-flow-state-stores.md (tables this build cr
 4417                    "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.
 1508424        if (columns["flow_id"].PrimaryKeyOrdinal != 1 || columns.Values.Count(column => column.PrimaryKeyOrdinal != 0) !
 425        {
 4426            throw new InvalidOperationException(
 4427                $"The SQLite durable-flow table '{_options.TableName}' does not declare PRIMARY KEY (flow_id) on that co
 4428                "Starting a flow is an insert-if-absent targeting ON CONFLICT(flow_id), which requires a uniqueness cons
 4429                "exactly flow_id — without it every flow creation fails, and a composite key admits duplicate ids. Re-cr
 4430                "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.
 3006437        foreach (var (name, actual) in columns)
 438        {
 6609439            if (ExpectedColumns.Any(expected => string.Equals(expected.Name, name, StringComparison.OrdinalIgnoreCase))
 1317440                || !actual.NotNull
 1317441                || actual.HasDefault)
 442            {
 443                continue;
 444            }
 445
 2446            throw new InvalidOperationException(
 2447                $"The SQLite durable-flow table '{_options.TableName}' has an extra column '{name}' ({actual.DeclaredTyp
 2448                "with no default. This store writes only its own columns, so every flow creation would fail on that colu
 2449                "it a default, make it nullable or generated, or move it to a table of your own.");
 450        }
 451
 185452        await VerifyFlowIdCollationAsync(connection, cancellationToken).ConfigureAwait(false);
 453
 177454        return true;
 181455    }
 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    {
 185477        await using var command = connection.CreateCommand();
 185478        command.CommandText = "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = $name COLLATE NOCASE;";
 185479        command.Parameters.AddWithValue("$name", _options.TableName);
 480
 185481        var ddl = await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false) as string;
 185482        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.
 732489        for (var start = IndexOfFlowIdIdentifier(ddl, 0); start >= 0; start = IndexOfFlowIdIdentifier(ddl, start + 1))
 490        {
 189491            var end = ddl.IndexOf(',', start);
 189492            var declaration = end < 0 ? ddl[start..] : ddl[start..end];
 493
 189494            var collate = declaration.IndexOf("COLLATE", StringComparison.OrdinalIgnoreCase);
 189495            if (collate < 0)
 496                continue;
 497
 8498            var tail = declaration[(collate + "COLLATE".Length)..].TrimStart();
 8499            var tokenEnd = 0;
 56500            while (tokenEnd < tail.Length && !char.IsWhiteSpace(tail[tokenEnd]) && tail[tokenEnd] is not (')' or ','))
 48501                tokenEnd++;
 502
 8503            var collation = tail[..tokenEnd].Trim('"');
 8504            if (collation.Length == 0 || string.Equals(collation, "BINARY", StringComparison.OrdinalIgnoreCase))
 505                continue;
 506
 8507            throw new InvalidOperationException(
 8508                $"The SQLite durable-flow table '{_options.TableName}' declares flow_id with COLLATE {collation}. Flow i
 8509                "ordinally by this library, so a case- or accent-insensitive collation makes ids differing only in case 
 8510                "primary key: the second flow fails to start and a load returns the other run's state. Re-create the tab
 8511                "using the default BINARY collation (the DDL in docs/durable-flow-state-stores.md, which tables this bui
 512        }
 177513    }
 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    {
 366522        for (var index = ddl.IndexOf("flow_id", startIndex, StringComparison.OrdinalIgnoreCase);
 368523             index >= 0;
 2524             index = index + 1 < ddl.Length ? ddl.IndexOf("flow_id", index + 1, StringComparison.OrdinalIgnoreCase) : -1
 525        {
 191526            var beforeIsIdentifierChar = index > 0 && (char.IsAsciiLetterOrDigit(ddl[index - 1]) || ddl[index - 1] == '_
 191527            var afterIndex = index + "flow_id".Length;
 191528            var afterIsIdentifierChar = afterIndex < ddl.Length && (char.IsAsciiLetterOrDigit(ddl[afterIndex]) || ddl[af
 191529            if (!beforeIsIdentifierChar && !afterIsIdentifierChar)
 189530                return index;
 531        }
 532
 177533        return -1;
 534    }
 535
 536    /// <summary>SQLite column-affinity rules (in the documented precedence order).</summary>
 537    private static string Affinity(string declaredType)
 538    {
 1373539        if (declaredType.Contains("INT", StringComparison.OrdinalIgnoreCase))
 193540            return "INTEGER";
 1180541        if (declaredType.Contains("CHAR", StringComparison.OrdinalIgnoreCase)
 1180542            || declaredType.Contains("CLOB", StringComparison.OrdinalIgnoreCase)
 1180543            || declaredType.Contains("TEXT", StringComparison.OrdinalIgnoreCase))
 544        {
 1176545            return "TEXT";
 546        }
 547
 4548        if (declaredType.Length == 0 || declaredType.Contains("BLOB", StringComparison.OrdinalIgnoreCase))
 0549            return "BLOB";
 4550        if (declaredType.Contains("REAL", StringComparison.OrdinalIgnoreCase)
 4551            || declaredType.Contains("FLOA", StringComparison.OrdinalIgnoreCase)
 4552            || declaredType.Contains("DOUB", StringComparison.OrdinalIgnoreCase))
 553        {
 0554            return "REAL";
 555        }
 556
 4557        return "NUMERIC";
 558    }
 559
 5665560    private sealed record ActualColumn(string DeclaredType, bool NotNull, bool HasDefault, long PrimaryKeyOrdinal);
 561
 9525562    private sealed record ExpectedColumn(string Name, string Declaration, string RequiredAffinity, bool NotNull)
 563    {
 564        public string? Mismatch(ActualColumn actual)
 565        {
 1371566            if (!string.Equals(Affinity(actual.DeclaredType), RequiredAffinity, StringComparison.Ordinal))
 2567                return $"resolves to {Affinity(actual.DeclaredType)} affinity where {RequiredAffinity} is required";
 1369568            if (actual.NotNull != NotNull)
 2569                return NotNull ? "must be NOT NULL" : "must be nullable (this store writes and clears NULL there)";
 1367570            return null;
 571        }
 572    }
 573
 6574    private static readonly ExpectedColumn[] ExpectedColumns =
 6575    [
 6576        new("flow_id", "TEXT NOT NULL PRIMARY KEY", "TEXT", NotNull: true),
 6577        new("state_json", "TEXT NOT NULL", "TEXT", NotNull: true),
 6578        new("expires_at_utc", "TEXT NOT NULL", "TEXT", NotNull: true),
 6579        new("updated_at_utc", "TEXT NOT NULL", "TEXT", NotNull: true),
 6580        new("revision", "INTEGER NOT NULL DEFAULT 0", "INTEGER", NotNull: true),
 6581        new("lease_id", "TEXT NULL", "TEXT", NotNull: false),
 6582        new("lease_expires_at_utc", "TEXT NULL", "TEXT", NotNull: false)
 6583    ];
 584
 585    private async Task<bool> UpdateLeaseAsync(
 586        string flowId,
 587        string leaseId,
 588        TimeSpan leaseDuration,
 589        bool acquire,
 590        CancellationToken cancellationToken)
 591    {
 199592        DurableFlowStoreShared.ValidateLeaseArgs(flowId, leaseId, leaseDuration);
 593
 197594        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 197595        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 197596        await using var command = connection.CreateCommand();
 197597        var now = DateTime.UtcNow;
 197598        command.CommandText =
 197599            $"""
 197600            UPDATE {Table}
 197601            SET lease_id = $lease_id, lease_expires_at_utc = $lease_expires_at_utc
 197602            WHERE flow_id = $flow_id
 197603              AND expires_at_utc > $now_utc
 197604              AND {(acquire ? "(lease_id IS NULL OR lease_expires_at_utc <= $now_utc OR lease_id = $lease_id)" : "lease_
 197605            """;
 197606        command.Parameters.AddWithValue("$flow_id", flowId);
 197607        command.Parameters.AddWithValue("$lease_id", leaseId);
 197608        command.Parameters.AddWithValue("$lease_expires_at_utc", DurableFlowStoreShared.AddSaturating(now, leaseDuration
 197609        command.Parameters.AddWithValue("$now_utc", now);
 197610        return await ExecuteWriteAsync(command, cancellationToken).ConfigureAwait(false) > 0;
 197611    }
 612
 613    private async Task<int> ExecuteWriteAsync(SqliteCommand command, CancellationToken cancellationToken)
 614    {
 7260615        await _writeGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 616        try
 617        {
 7260618            return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 619        }
 620        finally
 621        {
 7260622            _writeGate.Release();
 623        }
 7250624    }
 625
 626    private Task<SqliteConnection> OpenConnectionAsync(CancellationToken cancellationToken)
 8871627        => DurableFlowStoreShared.OpenConnectionAsync<SqliteConnection>(_options.ConnectionString, cancellationToken);
 628
 14618629    private string Table => Quote(_options.TableName);
 177630    private string IndexName => Quote($"{_options.TableName}_expires_idx");
 14795631    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

.ctor(Microsoft.Extensions.Options.IOptions`1<AsyncResponse.DurableFlows.Sqlite.SqliteDurableFlowOptions>,Microsoft.Extensions.Logging.ILogger`1<AsyncResponse.DurableFlows.Sqlite.SqliteFlowStateStore>)
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()
VerifyFlowTableAsync()
VerifyFlowIdCollationAsync()
IndexOfFlowIdIdentifier(System.String,System.Int32)
Affinity(System.String)
get_DeclaredType()
get_Name()
Mismatch(AsyncResponse.DurableFlows.Sqlite.SqliteFlowStateStore/ActualColumn)
.cctor()
UpdateLeaseAsync()
ExecuteWriteAsync()
OpenConnectionAsync(System.Threading.CancellationToken)
get_Table()
get_IndexName()
Quote(System.String)