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

Information
Class: Microsoft.Extensions.DependencyInjection.MySqlDurableFlowServiceCollectionExtensions
Assembly: AsyncResponse.DurableFlows.MySql
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/DurableFlows/AsyncResponse.DurableFlows.MySql/MySqlDurableFlows.cs
Line coverage
100%
Covered lines: 2
Uncovered lines: 0
Coverable lines: 2
Total lines: 325
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
WithMySqlDurableFlows(...)100%11100%

File(s)

/home/runner/work/AsyncResponse/AsyncResponse/src/DurableFlows/AsyncResponse.DurableFlows.MySql/MySqlDurableFlows.cs

#LineLine coverage
 1using AsyncResponse;
 2using AsyncResponse.DurableFlows.Internal;
 3using AsyncResponse.DurableFlows.MySql;
 4using Microsoft.Extensions.DependencyInjection.Extensions;
 5using Microsoft.Extensions.Options;
 6using MySqlConnector;
 7
 8namespace Microsoft.Extensions.DependencyInjection
 9{
 10    /// <summary>DI registration for the MySQL/MariaDB durable-flow state store.</summary>
 11    public static class MySqlDurableFlowServiceCollectionExtensions
 12    {
 13        /// <summary>Stores durable-flow state in MySQL or MariaDB.</summary>
 14        public static AsyncResponseRegistrationBuilder WithMySqlDurableFlows(
 15            this AsyncResponseRegistrationBuilder builder,
 16            Action<MySqlDurableFlowOptions>? 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.
 221            builder.Services.TryAddSingleton<MySqlFlowStateStore>();
 222            return builder.WithDurableFlows<MySqlFlowStateStore, MySqlDurableFlowOptions>(configure);
 23        }
 24    }
 25}
 26
 27namespace AsyncResponse.DurableFlows.MySql
 28{
 29/// <summary>Options for the MySQL/MariaDB durable-flow state store.</summary>
 30public sealed class MySqlDurableFlowOptions : DurableFlowOptions
 31{
 32    /// <summary>MySQL or MariaDB connection string. Required.</summary>
 33    public string? ConnectionString { get; set; }
 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="MySqlFlowStateStore.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 — <c>longtext</c> holds up to 4 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(MySqlDurableFlowOptions)}.{nameof(ConnectionString)} must be c
 60
 61        DurableFlowStoreShared.ValidateIdentifier(TableName, $"{nameof(MySqlDurableFlowOptions)}.{nameof(TableName)}", "
 62        if (MaxStateBytes is <= 0)
 63            throw new InvalidOperationException($"{nameof(MySqlDurableFlowOptions)}.{nameof(MaxStateBytes)} must be posi
 64    }
 65}
 66
 67/// <summary>MySQL/MariaDB implementation of <see cref="IFlowStateStore"/>.</summary>
 68public sealed class MySqlFlowStateStore : IFlowStateStore
 69{
 70    private const int PruneBatchSize = 1000;
 71
 72    /// <summary>
 73    /// SQL expression adding a millisecond bigint parameter to the database clock. All expiry and
 74    /// lease math runs on <c>UTC_TIMESTAMP(6)</c> (statement-stable, like <c>NOW()</c>) so app
 75    /// clock skew can never fence a lease in or out; microsecond arithmetic keeps
 76    /// <c>datetime(6)</c> precision.
 77    /// </summary>
 78    private static string AddMilliseconds(string parameterName)
 79        => $"TIMESTAMPADD(MICROSECOND, {parameterName} * 1000, UTC_TIMESTAMP(6))";
 80
 81    private readonly MySqlDurableFlowOptions _options;
 82    private readonly SemaphoreSlim _ensureGate = new(1, 1);
 83    private long _lastPruneTicks;
 84    private bool _created;
 85
 86    public MySqlFlowStateStore(IOptions<MySqlDurableFlowOptions> options)
 87    {
 88        _options = options.Value;
 89        _options.Validate();
 90    }
 91
 92    public async Task<FlowState?> LoadAsync(string flowId, CancellationToken cancellationToken = default)
 93    {
 94        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 95        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 96
 97        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 98        await using var command = connection.CreateCommand();
 99        command.CommandText = $"SELECT state_json, revision FROM {Table} WHERE flow_id = @flow_id AND expires_at_utc > U
 100        command.Parameters.AddWithValue("@flow_id", flowId);
 101
 102        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 103        if (!await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 104            return null;
 105
 106        return DurableFlowStoreShared.ReadState(flowId, reader.GetString(0), reader.GetInt64(1));
 107    }
 108
 109    public async Task<bool> TryCreateAsync(string flowId, FlowState state, TimeSpan ttl, CancellationToken cancellationT
 110    {
 111        DurableFlowStoreShared.ValidateCreate(flowId, state, ttl);
 112        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "MySQL");
 113        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 114        if (DurableFlowStoreShared.ShouldPrune(ref _lastPruneTicks, _options.PruneInterval))
 115            await PruneExpiredAsync(cancellationToken).ConfigureAwait(false);
 116
 117        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 118        await using var command = connection.CreateCommand();
 119        command.CommandText =
 120            $"""
 121            INSERT INTO {Table} (flow_id, state_json, expires_at_utc, updated_at_utc, revision)
 122            VALUES (@flow_id, @state_json, {AddMilliseconds("@ttl_ms")}, UTC_TIMESTAMP(6), @revision);
 123            """;
 124        command.Parameters.AddWithValue("@flow_id", flowId);
 125        command.Parameters.AddWithValue("@state_json", stateJson);
 126        command.Parameters.AddWithValue("@ttl_ms", DurableFlowStoreShared.ServerClockTtlMilliseconds(ttl));
 127        command.Parameters.AddWithValue("@revision", state.Revision);
 128        try
 129        {
 130            await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 131            return true;
 132        }
 133        catch (MySqlException exception) when (exception.Number == 1062)
 134        {
 135            // The id already exists. Only an expired row may be replaced below; do not use
 136            // INSERT IGNORE here because it also suppresses truncation and other data errors.
 137        }
 138
 139        // Exactly one caller can replace an expired ledger: after its conditional update, every
 140        // competing caller sees the new future expiry and returns false. This avoids relying on
 141        // MySQL's configurable "changed rows" versus "matched rows" result semantics.
 142        command.CommandText =
 143            $"""
 144            UPDATE {Table}
 145            SET state_json = @state_json,
 146                revision = @revision,
 147                lease_id = NULL,
 148                lease_expires_at_utc = NULL,
 149                updated_at_utc = UTC_TIMESTAMP(6),
 150                expires_at_utc = {AddMilliseconds("@ttl_ms")}
 151            WHERE flow_id = @flow_id AND expires_at_utc <= UTC_TIMESTAMP(6);
 152            """;
 153        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 154    }
 155
 156    public async Task<bool> TryUpdateAsync(
 157        string flowId,
 158        FlowState state,
 159        long expectedRevision,
 160        TimeSpan ttl,
 161        string? leaseId = null,
 162        CancellationToken cancellationToken = default)
 163    {
 164        DurableFlowStoreShared.ValidateUpdate(flowId, state, expectedRevision, ttl);
 165        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "MySQL");
 166        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 167
 168        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 169        await using var command = connection.CreateCommand();
 170        command.CommandText =
 171            $"""
 172            UPDATE {Table}
 173            SET state_json = @state_json,
 174                expires_at_utc = {AddMilliseconds("@ttl_ms")},
 175                updated_at_utc = UTC_TIMESTAMP(6),
 176                revision = @new_revision
 177            WHERE flow_id = @flow_id
 178              AND revision = @expected_revision
 179              AND expires_at_utc > UTC_TIMESTAMP(6)
 180              AND (@lease_id IS NULL OR (lease_id = @lease_id AND lease_expires_at_utc > UTC_TIMESTAMP(6)));
 181            """;
 182        command.Parameters.AddWithValue("@flow_id", flowId);
 183        command.Parameters.AddWithValue("@state_json", stateJson);
 184        command.Parameters.AddWithValue("@ttl_ms", DurableFlowStoreShared.ServerClockTtlMilliseconds(ttl));
 185        command.Parameters.AddWithValue("@expected_revision", expectedRevision);
 186        command.Parameters.AddWithValue("@new_revision", state.Revision);
 187        command.Parameters.AddWithValue("@lease_id", (object?)leaseId ?? DBNull.Value);
 188        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 189    }
 190
 191    public Task<bool> TryAcquireLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken canc
 192        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: true, cancellationToken);
 193
 194    public Task<bool> TryRenewLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken cancel
 195        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: false, 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 command.ExecuteNonQueryAsync(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 command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 218    }
 219
 220    private async Task PruneExpiredAsync(CancellationToken cancellationToken)
 221    {
 222        // One bounded batch per prune interval (policy shared by all relational stores): an
 223        // unbatched DELETE over a large expired backlog holds row locks and bloats one
 224        // transaction for the unlucky create that triggered the prune. Loads already filter on
 225        // expiry, so any backlog beyond the batch just waits for the next interval.
 226        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 227        await using var command = connection.CreateCommand();
 228        command.CommandText = $"DELETE FROM {Table} WHERE expires_at_utc <= UTC_TIMESTAMP(6) LIMIT {PruneBatchSize};";
 229        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 230    }
 231
 232    private async Task EnsureCreatedAsync(CancellationToken cancellationToken)
 233    {
 234        if (_created || !_options.AutoCreateSchema)
 235            return;
 236
 237        await _ensureGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 238        try
 239        {
 240            if (_created)
 241                return;
 242
 243            await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 244            await using var command = connection.CreateCommand();
 245            command.CommandText =
 246                $"""
 247                CREATE TABLE IF NOT EXISTS {Table} (
 248                    flow_id varchar(400) NOT NULL PRIMARY KEY,
 249                    state_json longtext NOT NULL,
 250                    expires_at_utc datetime(6) NOT NULL,
 251                    updated_at_utc datetime(6) NOT NULL,
 252                    revision bigint NOT NULL DEFAULT 0,
 253                    lease_id varchar(64) NULL,
 254                    lease_expires_at_utc datetime(6) NULL,
 255                    INDEX {IndexName} (expires_at_utc)
 256                );
 257                """;
 258            await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 259
 260            _created = true;
 261        }
 262        finally
 263        {
 264            _ensureGate.Release();
 265        }
 266    }
 267
 268    private async Task<bool> UpdateLeaseAsync(
 269        string flowId,
 270        string leaseId,
 271        TimeSpan leaseDuration,
 272        bool acquire,
 273        CancellationToken cancellationToken)
 274    {
 275        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 276        ArgumentException.ThrowIfNullOrWhiteSpace(leaseId);
 277        if (leaseDuration <= TimeSpan.Zero)
 278            throw new ArgumentOutOfRangeException(nameof(leaseDuration));
 279
 280        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 281        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 282        await using var command = connection.CreateCommand();
 283        // Lease fencing runs entirely on the database clock: acquire steals only leases the
 284        // database considers expired, and renew/extend stays relative to UTC_TIMESTAMP(6), so
 285        // worker clock skew can never make two nodes hold the same lease.
 286        command.CommandText =
 287            $"""
 288            UPDATE {Table}
 289            SET lease_id = @lease_id, lease_expires_at_utc = {AddMilliseconds("@lease_ms")}
 290            WHERE flow_id = @flow_id
 291              AND expires_at_utc > UTC_TIMESTAMP(6)
 292              AND {(acquire ? "(lease_id IS NULL OR lease_expires_at_utc <= UTC_TIMESTAMP(6) OR lease_id = @lease_id)" :
 293            """;
 294        command.Parameters.AddWithValue("@flow_id", flowId);
 295        command.Parameters.AddWithValue("@lease_id", leaseId);
 296        command.Parameters.AddWithValue("@lease_ms", DurableFlowStoreShared.ServerClockTtlMilliseconds(leaseDuration));
 297        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 298    }
 299
 300    private async Task<MySqlConnection> OpenConnectionAsync(CancellationToken cancellationToken)
 301    {
 302        // Row-count semantics guard: this store's lease renewal (and update fencing) treats
 303        // ExecuteNonQuery's result as ROWS MATCHED, which is MySqlConnector's default
 304        // (UseAffectedRows=false). A connection string with UseAffectedRows=true switches the
 305        // result to ROWS CHANGED, and a renewal that lands in the same microsecond as the current
 306        // lease expiry would report 0 and abort a healthy execution. Do not set
 307        // UseAffectedRows=true on this store's connection string.
 308        var connection = new MySqlConnection(_options.ConnectionString);
 309        try
 310        {
 311            await connection.OpenAsync(cancellationToken).ConfigureAwait(false);
 312            return connection;
 313        }
 314        catch
 315        {
 316            await connection.DisposeAsync().ConfigureAwait(false);
 317            throw;
 318        }
 319    }
 320
 321    private string Table => Quote(_options.TableName);
 322    private string IndexName => Quote($"{_options.TableName}_expires_idx");
 323    private static string Quote(string identifier) => "`" + identifier.Replace("`", "``", StringComparison.Ordinal) + "`
 324}
 325}
Generated by: ReportGenerator 5.5.10.0
08/03/2026 - 08:59:44
GitHub | reportgenerator.io

Methods/Properties

WithMySqlDurableFlows(Microsoft.Extensions.DependencyInjection.AsyncResponseRegistrationBuilder, System.Action<AsyncResponse.DurableFlows.MySql.MySqlDurableFlowOptions>)