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

Information
Class: AsyncResponse.DurableFlows.MySql.MySqlFlowStateStore
Assembly: AsyncResponse.DurableFlows.MySql
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/DurableFlows/AsyncResponse.DurableFlows.MySql/MySqlDurableFlows.cs
Line coverage
99%
Covered lines: 152
Uncovered lines: 1
Coverable lines: 153
Total lines: 325
Line coverage: 99.3%
Branch coverage
81%
Covered branches: 13
Total branches: 16
Branch coverage: 81.2%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
AddMilliseconds(...)100%11100%
.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()83.33%6696%
UpdateLeaseAsync()75%4495%
OpenConnectionAsync()50%2271.43%
get_Table()100%11100%
get_IndexName()100%11100%
Quote(...)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.
 21            builder.Services.TryAddSingleton<MySqlFlowStateStore>();
 22            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)
 179        => $"TIMESTAMPADD(MICROSECOND, {parameterName} * 1000, UTC_TIMESTAMP(6))";
 80
 81    private readonly MySqlDurableFlowOptions _options;
 382    private readonly SemaphoreSlim _ensureGate = new(1, 1);
 83    private long _lastPruneTicks;
 84    private bool _created;
 85
 386    public MySqlFlowStateStore(IOptions<MySqlDurableFlowOptions> options)
 87    {
 388        _options = options.Value;
 389        _options.Validate();
 390    }
 91
 92    public async Task<FlowState?> LoadAsync(string flowId, CancellationToken cancellationToken = default)
 93    {
 394        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 395        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 96
 397        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 198        await using var command = connection.CreateCommand();
 199        command.CommandText = $"SELECT state_json, revision FROM {Table} WHERE flow_id = @flow_id AND expires_at_utc > U
 1100        command.Parameters.AddWithValue("@flow_id", flowId);
 101
 1102        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 1103        if (!await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 1104            return null;
 105
 1106        return DurableFlowStoreShared.ReadState(flowId, reader.GetString(0), reader.GetInt64(1));
 1107    }
 108
 109    public async Task<bool> TryCreateAsync(string flowId, FlowState state, TimeSpan ttl, CancellationToken cancellationT
 110    {
 1111        DurableFlowStoreShared.ValidateCreate(flowId, state, ttl);
 1112        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "MySQL");
 1113        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 1114        if (DurableFlowStoreShared.ShouldPrune(ref _lastPruneTicks, _options.PruneInterval))
 1115            await PruneExpiredAsync(cancellationToken).ConfigureAwait(false);
 116
 1117        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1118        await using var command = connection.CreateCommand();
 1119        command.CommandText =
 1120            $"""
 1121            INSERT INTO {Table} (flow_id, state_json, expires_at_utc, updated_at_utc, revision)
 1122            VALUES (@flow_id, @state_json, {AddMilliseconds("@ttl_ms")}, UTC_TIMESTAMP(6), @revision);
 1123            """;
 1124        command.Parameters.AddWithValue("@flow_id", flowId);
 1125        command.Parameters.AddWithValue("@state_json", stateJson);
 1126        command.Parameters.AddWithValue("@ttl_ms", DurableFlowStoreShared.ServerClockTtlMilliseconds(ttl));
 1127        command.Parameters.AddWithValue("@revision", state.Revision);
 128        try
 129        {
 1130            await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 1131            return true;
 132        }
 1133        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.
 1137        }
 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.
 1142        command.CommandText =
 1143            $"""
 1144            UPDATE {Table}
 1145            SET state_json = @state_json,
 1146                revision = @revision,
 1147                lease_id = NULL,
 1148                lease_expires_at_utc = NULL,
 1149                updated_at_utc = UTC_TIMESTAMP(6),
 1150                expires_at_utc = {AddMilliseconds("@ttl_ms")}
 1151            WHERE flow_id = @flow_id AND expires_at_utc <= UTC_TIMESTAMP(6);
 1152            """;
 1153        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 1154    }
 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    {
 1164        DurableFlowStoreShared.ValidateUpdate(flowId, state, expectedRevision, ttl);
 1165        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "MySQL");
 1166        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 167
 1168        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1169        await using var command = connection.CreateCommand();
 1170        command.CommandText =
 1171            $"""
 1172            UPDATE {Table}
 1173            SET state_json = @state_json,
 1174                expires_at_utc = {AddMilliseconds("@ttl_ms")},
 1175                updated_at_utc = UTC_TIMESTAMP(6),
 1176                revision = @new_revision
 1177            WHERE flow_id = @flow_id
 1178              AND revision = @expected_revision
 1179              AND expires_at_utc > UTC_TIMESTAMP(6)
 1180              AND (@lease_id IS NULL OR (lease_id = @lease_id AND lease_expires_at_utc > UTC_TIMESTAMP(6)));
 1181            """;
 1182        command.Parameters.AddWithValue("@flow_id", flowId);
 1183        command.Parameters.AddWithValue("@state_json", stateJson);
 1184        command.Parameters.AddWithValue("@ttl_ms", DurableFlowStoreShared.ServerClockTtlMilliseconds(ttl));
 1185        command.Parameters.AddWithValue("@expected_revision", expectedRevision);
 1186        command.Parameters.AddWithValue("@new_revision", state.Revision);
 1187        command.Parameters.AddWithValue("@lease_id", (object?)leaseId ?? DBNull.Value);
 1188        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 1189    }
 190
 191    public Task<bool> TryAcquireLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken canc
 3192        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: true, cancellationToken);
 193
 194    public Task<bool> TryRenewLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken cancel
 1195        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: false, cancellationToken);
 196
 197    public async Task ReleaseLeaseAsync(string flowId, string leaseId, CancellationToken cancellationToken = default)
 198    {
 1199        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 1200        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1201        await using var command = connection.CreateCommand();
 1202        command.CommandText = $"UPDATE {Table} SET lease_id = NULL, lease_expires_at_utc = NULL WHERE flow_id = @flow_id
 1203        command.Parameters.AddWithValue("@flow_id", flowId);
 1204        command.Parameters.AddWithValue("@lease_id", leaseId);
 1205        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 1206    }
 207
 208    public async Task<bool> TryDeleteAsync(string flowId, CancellationToken cancellationToken = default)
 209    {
 1210        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 1211        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 212
 1213        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1214        await using var command = connection.CreateCommand();
 1215        command.CommandText = $"DELETE FROM {Table} WHERE flow_id = @flow_id;";
 1216        command.Parameters.AddWithValue("@flow_id", flowId);
 1217        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 1218    }
 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.
 1226        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1227        await using var command = connection.CreateCommand();
 1228        command.CommandText = $"DELETE FROM {Table} WHERE expires_at_utc <= UTC_TIMESTAMP(6) LIMIT {PruneBatchSize};";
 1229        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 1230    }
 231
 232    private async Task EnsureCreatedAsync(CancellationToken cancellationToken)
 233    {
 3234        if (_created || !_options.AutoCreateSchema)
 3235            return;
 236
 1237        await _ensureGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 238        try
 239        {
 1240            if (_created)
 0241                return;
 242
 1243            await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1244            await using var command = connection.CreateCommand();
 1245            command.CommandText =
 1246                $"""
 1247                CREATE TABLE IF NOT EXISTS {Table} (
 1248                    flow_id varchar(400) NOT NULL PRIMARY KEY,
 1249                    state_json longtext NOT NULL,
 1250                    expires_at_utc datetime(6) NOT NULL,
 1251                    updated_at_utc datetime(6) NOT NULL,
 1252                    revision bigint NOT NULL DEFAULT 0,
 1253                    lease_id varchar(64) NULL,
 1254                    lease_expires_at_utc datetime(6) NULL,
 1255                    INDEX {IndexName} (expires_at_utc)
 1256                );
 1257                """;
 1258            await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 259
 1260            _created = true;
 1261        }
 262        finally
 263        {
 1264            _ensureGate.Release();
 265        }
 3266    }
 267
 268    private async Task<bool> UpdateLeaseAsync(
 269        string flowId,
 270        string leaseId,
 271        TimeSpan leaseDuration,
 272        bool acquire,
 273        CancellationToken cancellationToken)
 274    {
 3275        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 3276        ArgumentException.ThrowIfNullOrWhiteSpace(leaseId);
 3277        if (leaseDuration <= TimeSpan.Zero)
 2278            throw new ArgumentOutOfRangeException(nameof(leaseDuration));
 279
 1280        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 1281        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 1282        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.
 1286        command.CommandText =
 1287            $"""
 1288            UPDATE {Table}
 1289            SET lease_id = @lease_id, lease_expires_at_utc = {AddMilliseconds("@lease_ms")}
 1290            WHERE flow_id = @flow_id
 1291              AND expires_at_utc > UTC_TIMESTAMP(6)
 1292              AND {(acquire ? "(lease_id IS NULL OR lease_expires_at_utc <= UTC_TIMESTAMP(6) OR lease_id = @lease_id)" :
 1293            """;
 1294        command.Parameters.AddWithValue("@flow_id", flowId);
 1295        command.Parameters.AddWithValue("@lease_id", leaseId);
 1296        command.Parameters.AddWithValue("@lease_ms", DurableFlowStoreShared.ServerClockTtlMilliseconds(leaseDuration));
 1297        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 1298    }
 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.
 3308        var connection = new MySqlConnection(_options.ConnectionString);
 309        try
 310        {
 3311            await connection.OpenAsync(cancellationToken).ConfigureAwait(false);
 1312            return connection;
 313        }
 2314        catch
 315        {
 2316            await connection.DisposeAsync().ConfigureAwait(false);
 2317            throw;
 318        }
 1319    }
 320
 1321    private string Table => Quote(_options.TableName);
 1322    private string IndexName => Quote($"{_options.TableName}_expires_idx");
 1323    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

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