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

Information
Class: AsyncResponse.DurableFlows.Sqlite.SqliteDurableFlowOptions
Assembly: AsyncResponse.DurableFlows.Sqlite
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/DurableFlows/AsyncResponse.DurableFlows.Sqlite/SqliteDurableFlows.cs
Line coverage
100%
Covered lines: 10
Uncovered lines: 0
Coverable lines: 10
Total lines: 345
Line coverage: 100%
Branch coverage
100%
Covered branches: 6
Total branches: 6
Branch coverage: 100%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
.ctor()100%11100%
Validate()100%66100%

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>
 333    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>
 336    public string TableName { get; set; } = "asyncresponse_flow_state";
 37
 38    /// <summary>Creates the table and expiry index on first use.</summary>
 339    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>
 346    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    {
 358        if (string.IsNullOrWhiteSpace(ConnectionString))
 359            throw new InvalidOperationException($"{nameof(SqliteDurableFlowOptions)}.{nameof(ConnectionString)} must be 
 60
 361        DurableFlowStoreShared.ValidateIdentifier(TableName, $"{nameof(SqliteDurableFlowOptions)}.{nameof(TableName)}", 
 362        if (MaxStateBytes is <= 0)
 263            throw new InvalidOperationException($"{nameof(SqliteDurableFlowOptions)}.{nameof(MaxStateBytes)} must be pos
 264    }
 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;
 77    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).
 86    private readonly SemaphoreSlim _writeGate = new(1, 1);
 87    private long _lastPruneTicks;
 88    private bool _created;
 89
 90    public SqliteFlowStateStore(IOptions<SqliteDurableFlowOptions> options)
 91    {
 92        _options = options.Value;
 93        _options.Validate();
 94    }
 95
 96    public async Task<FlowState?> LoadAsync(string flowId, CancellationToken cancellationToken = default)
 97    {
 98        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 99        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 100
 101        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 102        await using var command = connection.CreateCommand();
 103        command.CommandText =
 104            $"""
 105            SELECT state_json, revision
 106            FROM {Table}
 107            WHERE flow_id = $flow_id AND expires_at_utc > $now_utc;
 108            """;
 109        command.Parameters.AddWithValue("$flow_id", flowId);
 110        command.Parameters.AddWithValue("$now_utc", DateTime.UtcNow);
 111
 112        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 113        if (!await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 114            return null;
 115
 116        return DurableFlowStoreShared.ReadState(flowId, reader.GetString(0), reader.GetInt64(1));
 117    }
 118
 119    public async Task<bool> TryCreateAsync(
 120        string flowId,
 121        FlowState state,
 122        TimeSpan ttl,
 123        CancellationToken cancellationToken = default)
 124    {
 125        DurableFlowStoreShared.ValidateCreate(flowId, state, ttl);
 126        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "SQLite");
 127        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 128        if (DurableFlowStoreShared.ShouldPrune(ref _lastPruneTicks, _options.PruneInterval))
 129            await PruneExpiredAsync(cancellationToken).ConfigureAwait(false);
 130        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 131        await using var command = connection.CreateCommand();
 132        command.CommandText =
 133            $"""
 134            INSERT INTO {Table} (flow_id, state_json, expires_at_utc, updated_at_utc, revision)
 135            VALUES ($flow_id, $state_json, $expires_at_utc, $now_utc, $revision)
 136            ON CONFLICT(flow_id) DO UPDATE SET
 137                state_json = excluded.state_json,
 138                expires_at_utc = excluded.expires_at_utc,
 139                updated_at_utc = excluded.updated_at_utc,
 140                revision = excluded.revision,
 141                lease_id = NULL,
 142                lease_expires_at_utc = NULL
 143            WHERE {Table}.expires_at_utc <= $now_utc;
 144            """;
 145        var now = DateTime.UtcNow;
 146        command.Parameters.AddWithValue("$flow_id", flowId);
 147        command.Parameters.AddWithValue("$state_json", stateJson);
 148        command.Parameters.AddWithValue("$expires_at_utc", DurableFlowStoreShared.AddSaturating(now, ttl));
 149        command.Parameters.AddWithValue("$now_utc", now);
 150        command.Parameters.AddWithValue("$revision", state.Revision);
 151        return await ExecuteWriteAsync(command, cancellationToken).ConfigureAwait(false) > 0;
 152    }
 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    {
 162        DurableFlowStoreShared.ValidateUpdate(flowId, state, expectedRevision, ttl);
 163        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "SQLite");
 164        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 165        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 166        await using var command = connection.CreateCommand();
 167        var now = DateTime.UtcNow;
 168        command.CommandText =
 169            $"""
 170            UPDATE {Table}
 171            SET state_json = $state_json,
 172                expires_at_utc = $expires_at_utc,
 173                updated_at_utc = $updated_at_utc,
 174                revision = $new_revision
 175            WHERE flow_id = $flow_id
 176              AND revision = $expected_revision
 177              AND expires_at_utc > $now_utc
 178              AND ($lease_id IS NULL OR (lease_id = $lease_id AND lease_expires_at_utc > $now_utc));
 179            """;
 180        command.Parameters.AddWithValue("$flow_id", flowId);
 181        command.Parameters.AddWithValue("$state_json", stateJson);
 182        command.Parameters.AddWithValue("$expires_at_utc", DurableFlowStoreShared.AddSaturating(now, ttl));
 183        command.Parameters.AddWithValue("$updated_at_utc", now);
 184        command.Parameters.AddWithValue("$new_revision", state.Revision);
 185        command.Parameters.AddWithValue("$expected_revision", expectedRevision);
 186        command.Parameters.AddWithValue("$now_utc", now);
 187        command.Parameters.AddWithValue("$lease_id", (object?)leaseId ?? DBNull.Value);
 188        return await ExecuteWriteAsync(command, cancellationToken).ConfigureAwait(false) > 0;
 189    }
 190
 191    public Task<bool> TryAcquireLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken canc
 192        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, renew: false, cancellationToken);
 193
 194    public Task<bool> TryRenewLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken cancel
 195        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, renew: true, cancellationToken);
 196
 197    public async Task ReleaseLeaseAsync(string flowId, string leaseId, CancellationToken cancellationToken = default)
 198    {
 199        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 200        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 201        await using var command = connection.CreateCommand();
 202        command.CommandText = $"UPDATE {Table} SET lease_id = NULL, lease_expires_at_utc = NULL WHERE flow_id = $flow_id
 203        command.Parameters.AddWithValue("$flow_id", flowId);
 204        command.Parameters.AddWithValue("$lease_id", leaseId);
 205        await ExecuteWriteAsync(command, cancellationToken).ConfigureAwait(false);
 206    }
 207
 208    public async Task<bool> TryDeleteAsync(string flowId, CancellationToken cancellationToken = default)
 209    {
 210        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 211        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 212
 213        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 214        await using var command = connection.CreateCommand();
 215        command.CommandText = $"DELETE FROM {Table} WHERE flow_id = $flow_id;";
 216        command.Parameters.AddWithValue("$flow_id", flowId);
 217        return await ExecuteWriteAsync(command, cancellationToken).ConfigureAwait(false) > 0;
 218    }
 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.
 228        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 229        await using var command = connection.CreateCommand();
 230        command.CommandText =
 231            $"""
 232            DELETE FROM {Table}
 233            WHERE flow_id IN (SELECT flow_id FROM {Table} WHERE expires_at_utc <= $now_utc LIMIT {PruneBatchSize});
 234            """;
 235        command.Parameters.AddWithValue("$now_utc", DateTime.UtcNow);
 236        await ExecuteWriteAsync(command, cancellationToken).ConfigureAwait(false);
 237    }
 238
 239    private async Task EnsureCreatedAsync(CancellationToken cancellationToken)
 240    {
 241        if (_created || !_options.AutoCreateSchema)
 242            return;
 243
 244        await _ensureGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 245        try
 246        {
 247            if (_created)
 248                return;
 249
 250            await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 251            await using var command = connection.CreateCommand();
 252            command.CommandText =
 253                $"""
 254                -- WAL is the right journal mode for this store's use case (concurrent flow
 255                -- executors on one node): readers never block behind a writer, which rollback
 256                -- journal mode does not guarantee — concurrent load/save storms on slow disks
 257                -- surface as SQLITE_BUSY 'database is locked' there. The mode is persistent in
 258                -- the database file, so setting it alongside the schema costs nothing per
 259                -- operation. Manually-provisioned databases (AutoCreateSchema=false) should set
 260                -- it themselves — see docs/durable-flow-state-stores.md.
 261                PRAGMA journal_mode=WAL;
 262                CREATE TABLE IF NOT EXISTS {Table} (
 263                    flow_id TEXT NOT NULL PRIMARY KEY,
 264                    state_json TEXT NOT NULL,
 265                    expires_at_utc TEXT NOT NULL,
 266                    updated_at_utc TEXT NOT NULL,
 267                    revision INTEGER NOT NULL DEFAULT 0,
 268                    lease_id TEXT NULL,
 269                    lease_expires_at_utc TEXT NULL
 270                );
 271                CREATE INDEX IF NOT EXISTS {IndexName} ON {Table} (expires_at_utc);
 272                """;
 273            await ExecuteWriteAsync(command, cancellationToken).ConfigureAwait(false);
 274            _created = true;
 275        }
 276        finally
 277        {
 278            _ensureGate.Release();
 279        }
 280    }
 281
 282    private async Task<bool> UpdateLeaseAsync(
 283        string flowId,
 284        string leaseId,
 285        TimeSpan leaseDuration,
 286        bool renew,
 287        CancellationToken cancellationToken)
 288    {
 289        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 290        ArgumentException.ThrowIfNullOrWhiteSpace(leaseId);
 291        if (leaseDuration <= TimeSpan.Zero)
 292            throw new ArgumentOutOfRangeException(nameof(leaseDuration));
 293
 294        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 295        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 296        await using var command = connection.CreateCommand();
 297        var now = DateTime.UtcNow;
 298        command.CommandText =
 299            $"""
 300            UPDATE {Table}
 301            SET lease_id = $lease_id, lease_expires_at_utc = $lease_expires_at_utc
 302            WHERE flow_id = $flow_id
 303              AND expires_at_utc > $now_utc
 304              AND {(renew ? "lease_id = $lease_id AND lease_expires_at_utc > $now_utc" : "(lease_id IS NULL OR lease_exp
 305            """;
 306        command.Parameters.AddWithValue("$flow_id", flowId);
 307        command.Parameters.AddWithValue("$lease_id", leaseId);
 308        command.Parameters.AddWithValue("$lease_expires_at_utc", DurableFlowStoreShared.AddSaturating(now, leaseDuration
 309        command.Parameters.AddWithValue("$now_utc", now);
 310        return await ExecuteWriteAsync(command, cancellationToken).ConfigureAwait(false) > 0;
 311    }
 312
 313    private async Task<int> ExecuteWriteAsync(SqliteCommand command, CancellationToken cancellationToken)
 314    {
 315        await _writeGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 316        try
 317        {
 318            return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 319        }
 320        finally
 321        {
 322            _writeGate.Release();
 323        }
 324    }
 325
 326    private async Task<SqliteConnection> OpenConnectionAsync(CancellationToken cancellationToken)
 327    {
 328        var connection = new SqliteConnection(_options.ConnectionString);
 329        try
 330        {
 331            await connection.OpenAsync(cancellationToken).ConfigureAwait(false);
 332            return connection;
 333        }
 334        catch
 335        {
 336            await connection.DisposeAsync().ConfigureAwait(false);
 337            throw;
 338        }
 339    }
 340
 341    private string Table => Quote(_options.TableName);
 342    private string IndexName => Quote($"{_options.TableName}_expires_idx");
 343    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()
Validate()