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

Information
Class: AsyncResponse.DurableFlows.Sqlite.SqliteFlowStateStore
Assembly: AsyncResponse.DurableFlows.Sqlite
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/DurableFlows/AsyncResponse.DurableFlows.Sqlite/SqliteDurableFlows.cs
Line coverage
100%
Covered lines: 176
Uncovered lines: 0
Coverable lines: 176
Total lines: 345
Line coverage: 100%
Branch coverage
93%
Covered branches: 15
Total branches: 16
Branch coverage: 93.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%11100%
TryCreateAsync()100%22100%
TryUpdateAsync()100%22100%
TryAcquireLeaseAsync(...)100%11100%
TryRenewLeaseAsync(...)100%11100%
ReleaseLeaseAsync()100%11100%
TryDeleteAsync()100%11100%
PruneExpiredAsync()100%11100%
EnsureCreatedAsync()100%66100%
UpdateLeaseAsync()100%44100%
ExecuteWriteAsync()100%11100%
OpenConnectionAsync()50%22100%
get_Table()100%11100%
get_IndexName()100%11100%
Quote(...)100%11100%

File(s)

/home/runner/work/AsyncResponse/AsyncResponse/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.Options;
 7
 8namespace Microsoft.Extensions.DependencyInjection
 9{
 10    /// <summary>DI registration for the SQLite durable-flow state store.</summary>
 11    public static class SqliteDurableFlowServiceCollectionExtensions
 12    {
 13        /// <summary>Stores durable-flow state in SQLite.</summary>
 14        public static AsyncResponseRegistrationBuilder WithSqliteDurableFlows(
 15            this AsyncResponseRegistrationBuilder builder,
 16            Action<SqliteDurableFlowOptions>? configure = null)
 17        {
 18            // Singleton on purpose: schema provisioning is cached per store instance, and the
 19            // executor resolves the store from a fresh scope per flow execution — a scoped store
 20            // would re-run EnsureCreated's DDL round-trip on every run.
 21            builder.Services.TryAddSingleton<SqliteFlowStateStore>();
 22            return builder.WithDurableFlows<SqliteFlowStateStore, SqliteDurableFlowOptions>(configure);
 23        }
 24    }
 25}
 26
 27namespace AsyncResponse.DurableFlows.Sqlite
 28{
 29/// <summary>Options for the SQLite durable-flow state store.</summary>
 30public sealed class SqliteDurableFlowOptions : DurableFlowOptions
 31{
 32    /// <summary>SQLite connection string. Default: <c>Data Source=asyncresponse-flow-state.db</c>.</summary>
 33    public string ConnectionString { get; set; } = "Data Source=asyncresponse-flow-state.db";
 34
 35    /// <summary>Table storing one durable-flow ledger row per flow id.</summary>
 36    public string TableName { get; set; } = "asyncresponse_flow_state";
 37
 38    /// <summary>Creates the table and expiry index on first use.</summary>
 39    public bool AutoCreateSchema { get; set; } = true;
 40
 41    /// <summary>
 42    /// How often <see cref="SqliteFlowStateStore.TryCreateAsync"/> opportunistically deletes one bounded
 43    /// batch (1000 rows) of expired rows (loads already treat expired state as absent; pruning
 44    /// bounds table growth). Zero or negative prunes on every save. Default: 5 minutes.
 45    /// </summary>
 46    public TimeSpan PruneInterval { get; set; } = TimeSpan.FromMinutes(5);
 47
 48    /// <summary>
 49    /// Maximum serialized flow-state size in bytes accepted by writes; oversized ledgers fail fast
 50    /// with an actionable error instead of an opaque provider error. Default: <c>null</c>
 51    /// (unlimited — SQLite <c>TEXT</c> holds up to ~1 GB), settable as an operator budget.
 52    /// </summary>
 53    public long? MaxStateBytes { get; set; }
 54
 55    /// <summary>Validates option values and throws on misconfiguration.</summary>
 56    public void Validate()
 57    {
 58        if (string.IsNullOrWhiteSpace(ConnectionString))
 59            throw new InvalidOperationException($"{nameof(SqliteDurableFlowOptions)}.{nameof(ConnectionString)} must be 
 60
 61        DurableFlowStoreShared.ValidateIdentifier(TableName, $"{nameof(SqliteDurableFlowOptions)}.{nameof(TableName)}", 
 62        if (MaxStateBytes is <= 0)
 63            throw new InvalidOperationException($"{nameof(SqliteDurableFlowOptions)}.{nameof(MaxStateBytes)} must be pos
 64    }
 65}
 66
 67/// <summary>SQLite implementation of <see cref="IFlowStateStore"/>.</summary>
 68public sealed class SqliteFlowStateStore : IFlowStateStore
 69{
 70    private const int PruneBatchSize = 1000;
 71
 72    // Time authority: this store deliberately keeps the app clock (DateTime.UtcNow) for expiry
 73    // and lease comparisons. A SQLite database file lives on a single machine, and every writer
 74    // is a process on that machine sharing the same clock — the multi-node clock-skew hazard the
 75    // server-clock stores guard against cannot occur, and SQLite has no server clock to ask.
 76    private readonly SqliteDurableFlowOptions _options;
 377    private readonly SemaphoreSlim _ensureGate = new(1, 1);
 78
 79    // SQLite allows exactly one writer at a time, and its cross-connection busy handler is a
 80    // poll loop, not a queue: under heavy concurrency on a slow machine an unlucky writer can
 81    // lose every poll until the busy timeout expires ('database is locked' storms on 2-core CI
 82    // runners). Serializing this process's writers through a real FIFO gate costs no throughput
 83    // (they would serialize inside SQLite anyway) and makes in-process contention
 84    // starvation-free; the busy timeout then only covers cross-process writers. Reads stay
 85    // concurrent (WAL).
 386    private readonly SemaphoreSlim _writeGate = new(1, 1);
 87    private long _lastPruneTicks;
 88    private bool _created;
 89
 390    public SqliteFlowStateStore(IOptions<SqliteDurableFlowOptions> options)
 91    {
 392        _options = options.Value;
 393        _options.Validate();
 394    }
 95
 96    public async Task<FlowState?> LoadAsync(string flowId, CancellationToken cancellationToken = default)
 97    {
 398        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 399        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 100
 3101        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 3102        await using var command = connection.CreateCommand();
 3103        command.CommandText =
 3104            $"""
 3105            SELECT state_json, revision
 3106            FROM {Table}
 3107            WHERE flow_id = $flow_id AND expires_at_utc > $now_utc;
 3108            """;
 3109        command.Parameters.AddWithValue("$flow_id", flowId);
 3110        command.Parameters.AddWithValue("$now_utc", DateTime.UtcNow);
 111
 3112        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 3113        if (!await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 3114            return null;
 115
 3116        return DurableFlowStoreShared.ReadState(flowId, reader.GetString(0), reader.GetInt64(1));
 2117    }
 118
 119    public async Task<bool> TryCreateAsync(
 120        string flowId,
 121        FlowState state,
 122        TimeSpan ttl,
 123        CancellationToken cancellationToken = default)
 124    {
 3125        DurableFlowStoreShared.ValidateCreate(flowId, state, ttl);
 3126        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "SQLite");
 3127        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 3128        if (DurableFlowStoreShared.ShouldPrune(ref _lastPruneTicks, _options.PruneInterval))
 3129            await PruneExpiredAsync(cancellationToken).ConfigureAwait(false);
 3130        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 3131        await using var command = connection.CreateCommand();
 3132        command.CommandText =
 3133            $"""
 3134            INSERT INTO {Table} (flow_id, state_json, expires_at_utc, updated_at_utc, revision)
 3135            VALUES ($flow_id, $state_json, $expires_at_utc, $now_utc, $revision)
 3136            ON CONFLICT(flow_id) DO UPDATE SET
 3137                state_json = excluded.state_json,
 3138                expires_at_utc = excluded.expires_at_utc,
 3139                updated_at_utc = excluded.updated_at_utc,
 3140                revision = excluded.revision,
 3141                lease_id = NULL,
 3142                lease_expires_at_utc = NULL
 3143            WHERE {Table}.expires_at_utc <= $now_utc;
 3144            """;
 3145        var now = DateTime.UtcNow;
 3146        command.Parameters.AddWithValue("$flow_id", flowId);
 3147        command.Parameters.AddWithValue("$state_json", stateJson);
 3148        command.Parameters.AddWithValue("$expires_at_utc", DurableFlowStoreShared.AddSaturating(now, ttl));
 3149        command.Parameters.AddWithValue("$now_utc", now);
 3150        command.Parameters.AddWithValue("$revision", state.Revision);
 3151        return await ExecuteWriteAsync(command, cancellationToken).ConfigureAwait(false) > 0;
 2152    }
 153
 154    public async Task<bool> TryUpdateAsync(
 155        string flowId,
 156        FlowState state,
 157        long expectedRevision,
 158        TimeSpan ttl,
 159        string? leaseId = null,
 160        CancellationToken cancellationToken = default)
 161    {
 3162        DurableFlowStoreShared.ValidateUpdate(flowId, state, expectedRevision, ttl);
 3163        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "SQLite");
 3164        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 3165        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 3166        await using var command = connection.CreateCommand();
 3167        var now = DateTime.UtcNow;
 3168        command.CommandText =
 3169            $"""
 3170            UPDATE {Table}
 3171            SET state_json = $state_json,
 3172                expires_at_utc = $expires_at_utc,
 3173                updated_at_utc = $updated_at_utc,
 3174                revision = $new_revision
 3175            WHERE flow_id = $flow_id
 3176              AND revision = $expected_revision
 3177              AND expires_at_utc > $now_utc
 3178              AND ($lease_id IS NULL OR (lease_id = $lease_id AND lease_expires_at_utc > $now_utc));
 3179            """;
 3180        command.Parameters.AddWithValue("$flow_id", flowId);
 3181        command.Parameters.AddWithValue("$state_json", stateJson);
 3182        command.Parameters.AddWithValue("$expires_at_utc", DurableFlowStoreShared.AddSaturating(now, ttl));
 3183        command.Parameters.AddWithValue("$updated_at_utc", now);
 3184        command.Parameters.AddWithValue("$new_revision", state.Revision);
 3185        command.Parameters.AddWithValue("$expected_revision", expectedRevision);
 3186        command.Parameters.AddWithValue("$now_utc", now);
 3187        command.Parameters.AddWithValue("$lease_id", (object?)leaseId ?? DBNull.Value);
 3188        return await ExecuteWriteAsync(command, cancellationToken).ConfigureAwait(false) > 0;
 2189    }
 190
 191    public Task<bool> TryAcquireLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken canc
 3192        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, renew: false, cancellationToken);
 193
 194    public Task<bool> TryRenewLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken cancel
 3195        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, renew: true, cancellationToken);
 196
 197    public async Task ReleaseLeaseAsync(string flowId, string leaseId, CancellationToken cancellationToken = default)
 198    {
 3199        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 3200        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 3201        await using var command = connection.CreateCommand();
 3202        command.CommandText = $"UPDATE {Table} SET lease_id = NULL, lease_expires_at_utc = NULL WHERE flow_id = $flow_id
 3203        command.Parameters.AddWithValue("$flow_id", flowId);
 3204        command.Parameters.AddWithValue("$lease_id", leaseId);
 3205        await ExecuteWriteAsync(command, cancellationToken).ConfigureAwait(false);
 2206    }
 207
 208    public async Task<bool> TryDeleteAsync(string flowId, CancellationToken cancellationToken = default)
 209    {
 3210        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 3211        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 212
 3213        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 3214        await using var command = connection.CreateCommand();
 3215        command.CommandText = $"DELETE FROM {Table} WHERE flow_id = $flow_id;";
 3216        command.Parameters.AddWithValue("$flow_id", flowId);
 3217        return await ExecuteWriteAsync(command, cancellationToken).ConfigureAwait(false) > 0;
 2218    }
 219
 220    private async Task PruneExpiredAsync(CancellationToken cancellationToken)
 221    {
 222        // Timestamps are stored as ISO-8601 TEXT, which compares correctly lexicographically.
 223        // One bounded batch per prune interval (policy shared by all relational stores): an
 224        // unbatched DELETE over a large expired backlog holds the single SQLite write lock for
 225        // the whole sweep. Loads already filter on expiry, so any backlog beyond the batch just
 226        // waits for the next interval. Id-subquery form because DELETE ... LIMIT needs a
 227        // non-default SQLite compile flag.
 3228        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 3229        await using var command = connection.CreateCommand();
 3230        command.CommandText =
 3231            $"""
 3232            DELETE FROM {Table}
 3233            WHERE flow_id IN (SELECT flow_id FROM {Table} WHERE expires_at_utc <= $now_utc LIMIT {PruneBatchSize});
 3234            """;
 3235        command.Parameters.AddWithValue("$now_utc", DateTime.UtcNow);
 3236        await ExecuteWriteAsync(command, cancellationToken).ConfigureAwait(false);
 2237    }
 238
 239    private async Task EnsureCreatedAsync(CancellationToken cancellationToken)
 240    {
 3241        if (_created || !_options.AutoCreateSchema)
 3242            return;
 243
 3244        await _ensureGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 245        try
 246        {
 3247            if (_created)
 3248                return;
 249
 3250            await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 3251            await using var command = connection.CreateCommand();
 3252            command.CommandText =
 3253                $"""
 3254                -- WAL is the right journal mode for this store's use case (concurrent flow
 3255                -- executors on one node): readers never block behind a writer, which rollback
 3256                -- journal mode does not guarantee — concurrent load/save storms on slow disks
 3257                -- surface as SQLITE_BUSY 'database is locked' there. The mode is persistent in
 3258                -- the database file, so setting it alongside the schema costs nothing per
 3259                -- operation. Manually-provisioned databases (AutoCreateSchema=false) should set
 3260                -- it themselves — see docs/durable-flow-state-stores.md.
 3261                PRAGMA journal_mode=WAL;
 3262                CREATE TABLE IF NOT EXISTS {Table} (
 3263                    flow_id TEXT NOT NULL PRIMARY KEY,
 3264                    state_json TEXT NOT NULL,
 3265                    expires_at_utc TEXT NOT NULL,
 3266                    updated_at_utc TEXT NOT NULL,
 3267                    revision INTEGER NOT NULL DEFAULT 0,
 3268                    lease_id TEXT NULL,
 3269                    lease_expires_at_utc TEXT NULL
 3270                );
 3271                CREATE INDEX IF NOT EXISTS {IndexName} ON {Table} (expires_at_utc);
 3272                """;
 3273            await ExecuteWriteAsync(command, cancellationToken).ConfigureAwait(false);
 3274            _created = true;
 2275        }
 276        finally
 277        {
 3278            _ensureGate.Release();
 279        }
 2280    }
 281
 282    private async Task<bool> UpdateLeaseAsync(
 283        string flowId,
 284        string leaseId,
 285        TimeSpan leaseDuration,
 286        bool renew,
 287        CancellationToken cancellationToken)
 288    {
 3289        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 3290        ArgumentException.ThrowIfNullOrWhiteSpace(leaseId);
 3291        if (leaseDuration <= TimeSpan.Zero)
 3292            throw new ArgumentOutOfRangeException(nameof(leaseDuration));
 293
 3294        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 3295        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 3296        await using var command = connection.CreateCommand();
 3297        var now = DateTime.UtcNow;
 3298        command.CommandText =
 3299            $"""
 3300            UPDATE {Table}
 3301            SET lease_id = $lease_id, lease_expires_at_utc = $lease_expires_at_utc
 3302            WHERE flow_id = $flow_id
 3303              AND expires_at_utc > $now_utc
 3304              AND {(renew ? "lease_id = $lease_id AND lease_expires_at_utc > $now_utc" : "(lease_id IS NULL OR lease_exp
 3305            """;
 3306        command.Parameters.AddWithValue("$flow_id", flowId);
 3307        command.Parameters.AddWithValue("$lease_id", leaseId);
 3308        command.Parameters.AddWithValue("$lease_expires_at_utc", DurableFlowStoreShared.AddSaturating(now, leaseDuration
 3309        command.Parameters.AddWithValue("$now_utc", now);
 3310        return await ExecuteWriteAsync(command, cancellationToken).ConfigureAwait(false) > 0;
 2311    }
 312
 313    private async Task<int> ExecuteWriteAsync(SqliteCommand command, CancellationToken cancellationToken)
 314    {
 3315        await _writeGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 316        try
 317        {
 3318            return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 319        }
 320        finally
 321        {
 3322            _writeGate.Release();
 323        }
 2324    }
 325
 326    private async Task<SqliteConnection> OpenConnectionAsync(CancellationToken cancellationToken)
 327    {
 3328        var connection = new SqliteConnection(_options.ConnectionString);
 329        try
 330        {
 3331            await connection.OpenAsync(cancellationToken).ConfigureAwait(false);
 3332            return connection;
 333        }
 3334        catch
 335        {
 2336            await connection.DisposeAsync().ConfigureAwait(false);
 2337            throw;
 338        }
 2339    }
 340
 3341    private string Table => Quote(_options.TableName);
 3342    private string IndexName => Quote($"{_options.TableName}_expires_idx");
 3343    private static string Quote(string identifier) => "\"" + identifier.Replace("\"", "\"\"", StringComparison.Ordinal) 
 344}
 345}
Generated by: ReportGenerator 5.5.10.0
08/03/2026 - 08:59:44
GitHub | reportgenerator.io

Methods/Properties

.ctor(Microsoft.Extensions.Options.IOptions<AsyncResponse.DurableFlows.Sqlite.SqliteDurableFlowOptions>)
LoadAsync()
TryCreateAsync()
TryUpdateAsync()
TryAcquireLeaseAsync(string, string, System.TimeSpan, System.Threading.CancellationToken)
TryRenewLeaseAsync(string, string, System.TimeSpan, System.Threading.CancellationToken)
ReleaseLeaseAsync()
TryDeleteAsync()
PruneExpiredAsync()
EnsureCreatedAsync()
UpdateLeaseAsync()
ExecuteWriteAsync()
OpenConnectionAsync()
get_Table()
get_IndexName()
Quote(string)