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

Information
Class: AsyncResponse.DurableFlows.MySql.MySqlFlowStateStore
Assembly: AsyncResponse.DurableFlows.MySql
File(s): /_/src/DurableFlows/AsyncResponse.DurableFlows.MySql/MySqlDurableFlows.cs
Line coverage
100%
Covered lines: 315
Uncovered lines: 0
Coverable lines: 315
Total lines: 690
Line coverage: 100%
Branch coverage
90%
Covered branches: 94
Total branches: 104
Branch coverage: 90.3%
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%44100%
ValidateCreate(...)50%22100%
TryCreateAsync()100%66100%
ExistsAsync()100%11100%
TryUpdateAsync()100%22100%
TryAcquireLeaseAsync(...)100%11100%
TryRenewLeaseAsync(...)100%11100%
ReleaseLeaseAsync()100%11100%
ObserveLeaseAsync()100%88100%
TryDeleteAsync()100%11100%
PruneExpiredAsync()100%11100%
EnsureCreatedAsync()90%101084.84%
VerifyFlowTableAsync()89.58%4848100%
get_DataType()100%11100%
get_ColumnType()100%11100%
get_Collation()100%11100%
get_Nullable()100%11100%
get_MaxLength()100%11100%
get_DateTimePrecision()100%11100%
get_CharacterSet()100%11100%
get_HasDefault()100%11100%
get_Extra()100%11100%
get_IsWritableWithoutValue()100%66100%
get_Name()100%11100%
Mismatch(...)77.77%1818100%
.cctor()100%11100%
VerifyFlowIdIsUniqueAsync()100%22100%
UpdateLeaseAsync()100%22100%
OpenConnectionAsync(...)100%11100%
get_Table()100%11100%
get_IndexName()100%11100%
Quote(...)100%11100%

File(s)

/_/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.Logging;
 6using Microsoft.Extensions.Options;
 7using MySqlConnector;
 8
 9namespace Microsoft.Extensions.DependencyInjection
 10{
 11    /// <summary>DI registration for the MySQL/MariaDB durable-flow state store.</summary>
 12    public static class MySqlDurableFlowServiceCollectionExtensions
 13    {
 14        /// <summary>Stores durable-flow state in MySQL or MariaDB.</summary>
 15        public static AsyncResponseRegistrationBuilder WithMySqlDurableFlows(
 16            this AsyncResponseRegistrationBuilder builder,
 17            Action<MySqlDurableFlowOptions>? 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<MySqlFlowStateStore>();
 23            return builder.WithDurableFlows<MySqlFlowStateStore, MySqlDurableFlowOptions>(configure);
 24        }
 25    }
 26}
 27
 28namespace AsyncResponse.DurableFlows.MySql
 29{
 30/// <summary>Options for the MySQL/MariaDB durable-flow state store.</summary>
 31public sealed class MySqlDurableFlowOptions : DurableFlowOptions
 32{
 33    /// <summary>MySQL or MariaDB connection string. Required.</summary>
 34    public string? ConnectionString { get; set; }
 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="MySqlFlowStateStore.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 — <c>longtext</c> holds up to 4 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(MySqlDurableFlowOptions));
 72        DurableFlowStoreShared.ValidateIdentifier(TableName, $"{nameof(MySqlDurableFlowOptions)}.{nameof(TableName)}", "
 73        DurableFlowStoreShared.ValidateMaxStateBytes(MaxStateBytes, nameof(MySqlDurableFlowOptions));
 74        DurableFlowStoreShared.ValidatePruneBudget(PruneBudget, nameof(MySqlDurableFlowOptions));
 75    }
 76}
 77
 78/// <summary>MySQL/MariaDB implementation of <see cref="IFlowStateStore"/>.</summary>
 79public sealed class MySqlFlowStateStore : IFlowStateStore
 80{
 81    private readonly ILogger<MySqlFlowStateStore>? _logger;
 82
 83    /// <summary>
 84    /// SQL expression adding a millisecond bigint parameter to the database clock. All expiry and
 85    /// lease math runs on <c>UTC_TIMESTAMP(6)</c> (statement-stable, like <c>NOW()</c>) so app
 86    /// clock skew can never fence a lease in or out; microsecond arithmetic keeps
 87    /// <c>datetime(6)</c> precision.
 88    /// </summary>
 89    private static string AddMilliseconds(string parameterName)
 147490        => $"TIMESTAMPADD(MICROSECOND, {parameterName} * 1000, UTC_TIMESTAMP(6))";
 91
 92    private readonly MySqlDurableFlowOptions _options;
 23493    private readonly SemaphoreSlim _ensureGate = new(1, 1);
 94    private long _lastPruneTicks;
 95    private volatile bool _created;
 96
 23497    public MySqlFlowStateStore(IOptions<MySqlDurableFlowOptions> options, ILogger<MySqlFlowStateStore>? logger = null)
 98    {
 23499        _logger = logger;
 234100        _options = options.Value;
 234101        _options.Validate();
 234102    }
 103
 104    public async Task<FlowState?> LoadAsync(string flowId, CancellationToken cancellationToken = default)
 105    {
 681106        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 681107        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 108
 677109        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 677110        await using var command = connection.CreateCommand();
 677111        command.CommandText = $"SELECT state_json, revision FROM {Table} WHERE flow_id = @flow_id AND expires_at_utc > U
 677112        command.Parameters.AddWithValue("@flow_id", flowId);
 113
 677114        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 677115        if (!await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 5116            return null;
 117
 672118        return DurableFlowStoreShared.ReadState(flowId, reader.GetString(0), reader.GetInt64(1));
 677119    }
 120
 121    /// <inheritdoc />
 122    public void ValidateCreate(string flowId, FlowState state, TimeSpan ttl)
 123    {
 136124        DurableFlowStoreShared.ValidateCreate(flowId, state, ttl);
 136125        if (_options.MaxStateBytes is not null)
 4126            _ = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "MySQL");
 134127    }
 128
 129    public async Task<bool> TryCreateAsync(string flowId, FlowState state, TimeSpan ttl, CancellationToken cancellationT
 130    {
 316131        DurableFlowStoreShared.ValidateCreate(flowId, state, ttl);
 315132        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "MySQL");
 315133        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 302134        if (DurableFlowStoreShared.ShouldPrune(ref _lastPruneTicks, _options.PruneInterval))
 286135            await DurableFlowStoreShared.PruneQuietlyAsync(() => PruneExpiredAsync(cancellationToken), _options.PruneBud
 136
 302137        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 302138        await using var command = connection.CreateCommand();
 302139        command.CommandText =
 302140            $"""
 302141            INSERT INTO {Table} (flow_id, state_json, expires_at_utc, updated_at_utc, revision)
 302142            VALUES (@flow_id, @state_json, {AddMilliseconds("@ttl_ms")}, UTC_TIMESTAMP(6), @revision);
 302143            """;
 302144        command.Parameters.AddWithValue("@flow_id", flowId);
 302145        command.Parameters.AddWithValue("@state_json", stateJson);
 302146        command.Parameters.AddWithValue("@ttl_ms", DurableFlowStoreShared.ServerClockTtlMilliseconds(ttl));
 302147        command.Parameters.AddWithValue("@revision", state.Revision);
 148        try
 149        {
 302150            await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 149151            return true;
 152        }
 153153        catch (MySqlException exception) when (exception.Number == 1062)
 154        {
 155            // 1062 says "some unique constraint rejected this row" — NOT "this flow id exists".
 156            // The distinction is load-bearing on a table this build did not create: a legacy
 157            // PREFIX key alongside the required one (UNIQUE (flow_id(100))) raises 1062 for a
 158            // DIFFERENT id that happens to share a prefix, and reading that as "already exists"
 159            // would report a successful start for a flow with no row and no run. Startup
 160            // verification refuses such tables, but this store also runs against schemas it did not
 161            // get to inspect first (AutoCreateSchema off, table created later), so confirm the row
 162            // is actually there before believing the error.
 152163            if (!await ExistsAsync(connection, flowId, cancellationToken).ConfigureAwait(false))
 1164                throw;
 165
 166            // The id already exists. Only an expired row may be replaced below; do not use
 167            // INSERT IGNORE here because it also suppresses truncation and other data errors.
 168        }
 169
 170        // Exactly one caller can replace an expired ledger: after its conditional update, every
 171        // competing caller sees the new future expiry and returns false. This avoids relying on
 172        // MySQL's configurable "changed rows" versus "matched rows" result semantics.
 151173        command.CommandText =
 151174            $"""
 151175            UPDATE {Table}
 151176            SET state_json = @state_json,
 151177                revision = @revision,
 151178                lease_id = NULL,
 151179                lease_expires_at_utc = NULL,
 151180                updated_at_utc = UTC_TIMESTAMP(6),
 151181                expires_at_utc = {AddMilliseconds("@ttl_ms")}
 151182            WHERE flow_id = @flow_id AND expires_at_utc <= UTC_TIMESTAMP(6);
 151183            """;
 151184        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 300185    }
 186
 187    /// <summary>
 188    /// Whether a row with EXACTLY this flow id exists, expired or not — the question a 1062 does
 189    /// not answer on its own. Runs on the caller's already-open connection: <c>TryCreateAsync</c>
 190    /// holds its connection across the 1062 handling, and opening a SECOND one from inside that
 191    /// window meant every duplicate create occupied one pooled connection while waiting for
 192    /// another — with <c>MaximumPoolSize=1</c> a single duplicate start timed out with "All pooled
 193    /// connections are in use", and under concurrent idempotent starts the pool starved whatever
 194    /// size it had.
 195    /// </summary>
 196    private async Task<bool> ExistsAsync(MySqlConnection connection, string flowId, CancellationToken cancellationToken)
 197    {
 152198        await using var command = connection.CreateCommand();
 152199        command.CommandText = $"SELECT 1 FROM {Table} WHERE flow_id = @flow_id LIMIT 1;";
 152200        command.Parameters.AddWithValue("@flow_id", flowId);
 152201        return await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false) is not null;
 152202    }
 203
 204    public async Task<bool> TryUpdateAsync(
 205        string flowId,
 206        FlowState state,
 207        long expectedRevision,
 208        TimeSpan ttl,
 209        string? leaseId = null,
 210        CancellationToken cancellationToken = default)
 211    {
 863212        DurableFlowStoreShared.ValidateUpdate(flowId, state, expectedRevision, ttl);
 863213        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "MySQL");
 863214        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 215
 863216        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 863217        await using var command = connection.CreateCommand();
 863218        command.CommandText =
 863219            $"""
 863220            UPDATE {Table}
 863221            SET state_json = @state_json,
 863222                expires_at_utc = {AddMilliseconds("@ttl_ms")},
 863223                updated_at_utc = UTC_TIMESTAMP(6),
 863224                revision = @new_revision
 863225            WHERE flow_id = @flow_id
 863226              AND revision = @expected_revision
 863227              AND expires_at_utc > UTC_TIMESTAMP(6)
 863228              AND (@lease_id IS NULL OR (lease_id = @lease_id AND lease_expires_at_utc > UTC_TIMESTAMP(6)));
 863229            """;
 863230        command.Parameters.AddWithValue("@flow_id", flowId);
 863231        command.Parameters.AddWithValue("@state_json", stateJson);
 863232        command.Parameters.AddWithValue("@ttl_ms", DurableFlowStoreShared.ServerClockTtlMilliseconds(ttl));
 863233        command.Parameters.AddWithValue("@expected_revision", expectedRevision);
 863234        command.Parameters.AddWithValue("@new_revision", state.Revision);
 863235        command.Parameters.AddWithValue("@lease_id", (object?)leaseId ?? DBNull.Value);
 863236        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 863237    }
 238
 239    public Task<bool> TryAcquireLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken canc
 151240        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: true, cancellationToken);
 241
 242    public Task<bool> TryRenewLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken cancel
 9243        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: false, cancellationToken);
 244
 245    public async Task ReleaseLeaseAsync(string flowId, string leaseId, CancellationToken cancellationToken = default)
 246    {
 142247        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 142248        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 142249        await using var command = connection.CreateCommand();
 142250        command.CommandText = $"UPDATE {Table} SET lease_id = NULL, lease_expires_at_utc = NULL WHERE flow_id = @flow_id
 142251        command.Parameters.AddWithValue("@flow_id", flowId);
 142252        command.Parameters.AddWithValue("@lease_id", leaseId);
 142253        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 142254    }
 255
 256    /// <inheritdoc />
 257    public async Task<FlowLeaseObservation?> ObserveLeaseAsync(string flowId, CancellationToken cancellationToken = defa
 258    {
 18259        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 12260        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 261
 262        // The two lease columns exactly as stored — deliberately no UTC_TIMESTAMP(6) predicate,
 263        // unlike every other statement in this store: an expired lease nobody has taken over must
 264        // keep reading as the same lease, because the engine's proof of a live holder is that two
 265        // observations DIFFER. Whether it has lapsed stays UpdateLeaseAsync's call, on the
 266        // database clock.
 12267        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 12268        await using var command = connection.CreateCommand();
 12269        command.CommandText = $"SELECT lease_id, lease_expires_at_utc FROM {Table} WHERE flow_id = @flow_id;";
 12270        command.Parameters.AddWithValue("@flow_id", flowId);
 271
 12272        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 12273        if (!await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 2274            return FlowLeaseObservation.Unheld;
 275
 276        // datetime(6) carries no zone and the value is UTC_TIMESTAMP(6) arithmetic, so it IS UTC
 277        // whatever kind the connection labels it with: MySqlConnector's DateTimeKind=Local option
 278        // relabels the same digits as local time without converting them, and the shared shaper
 279        // would then shift a Local value by the host's offset. Strip the label first.
 10280        return DurableFlowStoreShared.LeaseObservation(
 10281            reader.IsDBNull(0) ? null : reader.GetString(0),
 10282            reader.IsDBNull(1) ? null : DateTime.SpecifyKind(reader.GetDateTime(1), DateTimeKind.Unspecified));
 12283    }
 284
 285    public async Task<bool> TryDeleteAsync(string flowId, CancellationToken cancellationToken = default)
 286    {
 10287        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 10288        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 289
 10290        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 10291        await using var command = connection.CreateCommand();
 10292        command.CommandText = $"DELETE FROM {Table} WHERE flow_id = @flow_id;";
 10293        command.Parameters.AddWithValue("@flow_id", flowId);
 10294        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 10295    }
 296
 297    private async Task<int> PruneExpiredAsync(CancellationToken cancellationToken)
 298    {
 299        // One bounded batch per call; DurableFlowStoreShared.PruneQuietlyAsync repeats it under
 300        // the PruneBudget while batches come back full (policy shared by all relational stores): an
 301        // unbatched DELETE over a large expired backlog holds row locks and bloats one
 302        // transaction for the unlucky create that triggered the prune. Loads already filter on
 303        // expiry, so any backlog beyond the batch just waits for the next interval.
 143304        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 143305        await using var command = connection.CreateCommand();
 143306        command.CommandText = $"DELETE FROM {Table} WHERE expires_at_utc <= UTC_TIMESTAMP(6) LIMIT {DurableFlowStoreShar
 143307        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 142308    }
 309
 310    private async Task EnsureCreatedAsync(CancellationToken cancellationToken)
 311    {
 2181312        if (_created)
 1907313            return;
 314
 274315        await _ensureGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 316        try
 317        {
 274318            if (_created)
 114319                return;
 320
 321            // Row-count semantics guard: lease renewal and update fencing treat ExecuteNonQuery's
 322            // result as ROWS MATCHED, MySqlConnector's default (UseAffectedRows=false). With
 323            // UseAffectedRows=true the result becomes rows CHANGED, so an UPDATE that rewrites
 324            // identical values (a renewal landing in the same microsecond as the stored expiry, a
 325            // stalled clock) reports 0 and a healthy execution aborts as "lease lost" — sporadic
 326            // and unattributable from logs. Every other silently-breaking property (charset,
 327            // collation, column shape, keys) fails startup in VerifyFlowTableAsync below; the
 328            // connection string gets the same treatment.
 160329            if (new MySqlConnectionStringBuilder(_options.ConnectionString!).UseAffectedRows)
 330            {
 2331                throw new InvalidOperationException(
 2332                    $"{nameof(MySqlDurableFlowOptions)}.{nameof(MySqlDurableFlowOptions.ConnectionString)} sets UseAffec
 2333                    "which switches ExecuteNonQuery from rows-MATCHED to rows-CHANGED semantics and silently breaks this
 2334                    "lease renewal and update fencing. Remove UseAffectedRows from the connection string; the MySqlConne
 2335                    "default (false) is required.");
 336            }
 337
 158338            await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 156339            if (_options.AutoCreateSchema)
 340            {
 138341                await using var command = connection.CreateCommand();
 138342                command.CommandText =
 138343                    $"""
 138344                    CREATE TABLE IF NOT EXISTS {Table} (
 138345                        flow_id varchar(400) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL PRIMARY KEY,
 138346                        state_json longtext CHARACTER SET utf8mb4 NOT NULL,
 138347                        expires_at_utc datetime(6) NOT NULL,
 138348                        updated_at_utc datetime(6) NOT NULL,
 138349                        revision bigint NOT NULL DEFAULT 0,
 138350                        lease_id varchar(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NULL,
 138351                        lease_expires_at_utc datetime(6) NULL,
 138352                        INDEX {IndexName} (expires_at_utc)
 138353                    );
 138354                    """;
 138355                await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 138356            }
 357
 358            // Latch only when the table was actually verified (Oracle parity): an absent table
 359            // under AutoCreateSchema = false must keep re-verifying, or a migration that lands
 360            // AFTER the first operation would never have its unique-key/collation/charset checks
 361            // run for the process lifetime.
 156362            _created = await VerifyFlowTableAsync(connection, cancellationToken).ConfigureAwait(false);
 143363        }
 364        finally
 365        {
 274366            _ensureGate.Release();
 367        }
 2164368    }
 369
 370    /// <summary>
 371    /// Checks the table this store will actually use, independently of who created it.
 372    /// <c>CREATE TABLE IF NOT EXISTS</c> leaves a table made by an earlier build (or by hand)
 373    /// exactly as it was, and <c>AutoCreateSchema = false</c> issues no DDL at all — so the DDL
 374    /// above only ever protects a table this build created. Two properties of that table are
 375    /// load-bearing and both fail SILENTLY when absent, which is why they are checked at startup
 376    /// rather than left to the first query:
 377    /// <list type="bullet">
 378    /// <item><description>
 379    /// A UNIQUE key on flow_id alone. <see cref="TryCreateAsync"/> is the engine's insert-if-absent
 380    /// primitive and detects "already exists" from MySQL's duplicate-key error 1062 — with no such
 381    /// key nothing raises 1062, so two concurrent starts of ONE flow id both report success and the
 382    /// ledger gets two rows.
 383    /// </description></item>
 384    /// <item><description>
 385    /// A binary collation on flow_id. MySQL's default is case-insensitive, which makes two ids
 386    /// differing only in case (or accent, or width) one key: the second start fails as a duplicate
 387    /// and a load returns the other run's state.
 388    /// </description></item>
 389    /// </list>
 390    /// </summary>
 391    private async Task<bool> VerifyFlowTableAsync(MySqlConnection connection, CancellationToken cancellationToken)
 392    {
 156393        var columns = new Dictionary<string, ActualColumn>(StringComparer.OrdinalIgnoreCase);
 156394        await using (var command = connection.CreateCommand())
 395        {
 156396            command.CommandText =
 156397                """
 156398                SELECT COLUMN_NAME, DATA_TYPE, COLUMN_TYPE, COLLATION_NAME, IS_NULLABLE,
 156399                       CHARACTER_MAXIMUM_LENGTH, DATETIME_PRECISION, CHARACTER_SET_NAME,
 156400                       COLUMN_DEFAULT, EXTRA
 156401                FROM information_schema.COLUMNS
 156402                WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = @table;
 156403                """;
 156404            command.Parameters.AddWithValue("@table", _options.TableName);
 156405            await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 1242406            while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 407            {
 1086408                columns[reader.GetString(0)] = new ActualColumn(
 1086409                    DataType: reader.GetString(1),
 1086410                    ColumnType: reader.GetString(2),
 1086411                    Collation: reader.IsDBNull(3) ? null : reader.GetString(3),
 1086412                    Nullable: string.Equals(reader.GetString(4), "YES", StringComparison.OrdinalIgnoreCase),
 1086413                    MaxLength: reader.IsDBNull(5) ? null : reader.GetInt64(5),
 1086414                    DateTimePrecision: reader.IsDBNull(6) ? null : reader.GetInt64(6),
 1086415                    CharacterSet: reader.IsDBNull(7) ? null : reader.GetString(7),
 1086416                    HasDefault: !reader.IsDBNull(8),
 1086417                    Extra: reader.IsDBNull(9) ? string.Empty : reader.GetString(9));
 418            }
 156419        }
 420
 156421        if (columns.Count == 0)
 422        {
 423            // The table does not exist: AutoCreateSchema = false and the migration has not run yet.
 424            // That surfaces at the first query with a clear MySQL error, and failing here would
 425            // break the documented "create it yourself, later" workflow. Returning false keeps
 426            // _created unlatched so the next operation re-verifies once the migration has run.
 1427            return false;
 428        }
 429
 2439430        foreach (var expected in ExpectedColumns)
 431        {
 1067432            if (!columns.TryGetValue(expected.Name, out var actual))
 433            {
 1434                throw new InvalidOperationException(
 1435                    $"The MySQL durable-flow table '{_options.TableName}' has no '{expected.Name}' column. It was create
 1436                    "earlier build or by hand and does not match the shape this store reads and writes " +
 7437                    $"({string.Join(", ", ExpectedColumns.Select(column => $"{column.Name} {column.Declaration}"))}). Re
 1438                    "or add the missing columns — the DDL is in docs/durable-flow-state-stores.md.");
 439            }
 440
 1066441            if (expected.Mismatch(actual) is { } mismatch)
 442            {
 4443                throw new InvalidOperationException(
 4444                    $"The MySQL durable-flow table '{_options.TableName}' declares {expected.Name} as '{actual.ColumnTyp
 4445                    $"{(actual.Nullable ? " NULL" : " NOT NULL")}', which {mismatch}. This store needs " +
 4446                    $"{expected.Name} {expected.Declaration}. Fix it with " +
 4447                    $"ALTER TABLE `{_options.TableName}` MODIFY {expected.Name} {expected.Declaration}; " +
 4448                    "(tables this build creates get that shape automatically).");
 449            }
 450        }
 451
 452        // Columns this store never names in an INSERT. One that the database cannot fill in for
 453        // itself makes EVERY create fail — the shape is otherwise perfect, so the failure arrives
 454        // at the first flow rather than at startup, which is the wrong end of the deployment.
 455        // Generated and auto-increment columns are fine; so is anything nullable or defaulted.
 2405456        foreach (var (name, actual) in columns)
 457        {
 5277458            if (ExpectedColumns.Any(expected => string.Equals(expected.Name, name, StringComparison.OrdinalIgnoreCase))
 1053459                || actual.IsWritableWithoutValue)
 460            {
 461                continue;
 462            }
 463
 1464            throw new InvalidOperationException(
 1465                $"The MySQL durable-flow table '{_options.TableName}' has an extra column '{name}' ({actual.ColumnType} 
 1466                "with no default. This store writes only its own columns, so every flow creation would fail on that colu
 1467                "it a default, make it nullable or generated, or move it to a table of your own.");
 468        }
 469
 149470        var flowIdColumn = columns["flow_id"];
 471        // utf8mb4 or nothing: MySQL's older `utf8` is three bytes and holds no supplementary
 472        // character, and a single-byte set like latin1 holds almost nothing. Either one turns a
 473        // perfectly legal flow id — an emoji, a Han character, most non-Latin text — into an insert
 474        // error or a mangled key, and the collation check above cannot see it because latin1_bin
 475        // ends in _bin just like utf8mb4_bin does.
 149476        if (flowIdColumn.CharacterSet is not { } characterSet
 149477            || !characterSet.Equals("utf8mb4", StringComparison.OrdinalIgnoreCase))
 478        {
 1479            throw new InvalidOperationException(
 1480                $"The MySQL durable-flow table '{_options.TableName}' stores flow_id in the character set " +
 1481                $"'{flowIdColumn.CharacterSet ?? "(none)"}', which cannot hold every id the engine accepts. Flow ids are
 1482                "text — this store's contract bounds their length, not their alphabet — so a narrower set rejects or man
 1483                $"that are perfectly valid. Fix it with ALTER TABLE `{_options.TableName}` MODIFY flow_id varchar(400) C
 1484                "SET utf8mb4 COLLATE utf8mb4_bin NOT NULL; (tables this build creates get that character set automatical
 485        }
 486
 148487        var collation = flowIdColumn.Collation;
 148488        if (collation is null || !collation.EndsWith("_bin", StringComparison.OrdinalIgnoreCase))
 489        {
 2490            throw new InvalidOperationException(
 2491                $"The MySQL durable-flow table '{_options.TableName}' stores flow_id with the collation '{collation ?? "
 2492                "which is not binary. Flow ids are compared ordinally by the engine, so ids differing only in case (or a
 2493                "width) collide on the primary key: the second flow fails to start and a load returns the other run's st
 2494                $"with ALTER TABLE `{_options.TableName}` MODIFY flow_id varchar(400) CHARACTER SET utf8mb4 COLLATE utf8
 2495                "NULL; (tables this build creates get that collation automatically).");
 496        }
 497
 498        // The ledger JSON needs the same alphabet as the ids it embeds: on a latin1-default
 499        // server a table that inherited the database charset stores state_json in latin1, so any
 500        // non-Latin-1 state (a name, an emoji in a step result) hard-fails every update under
 501        // strict mode or is silently truncated at the first bad byte otherwise — malformed JSON
 502        // that deserializes to null, a flow that can neither load nor be re-created.
 146503        var stateJsonColumn = columns["state_json"];
 146504        if (stateJsonColumn.CharacterSet is not { } stateJsonCharacterSet
 146505            || !stateJsonCharacterSet.Equals("utf8mb4", StringComparison.OrdinalIgnoreCase))
 506        {
 507            // Tables created by builds before this charset was pinned declared state_json with no
 508            // CHARACTER SET and inherited the server default, so an unconditional throw would
 509            // hard-fail every pre-existing deployment on a latin1/utf8mb3-default server with no
 510            // way back: CREATE TABLE IF NOT EXISTS cannot alter an existing table. Under
 511            // AutoCreateSchema this store owns the DDL, so it repairs the column in place — MODIFY
 512            // converts the stored text to utf8mb4, lossless for everything the old charset could
 513            // actually represent. Operator-managed schemas keep the throw, with the exact ALTER.
 2514            if (_options.AutoCreateSchema)
 515            {
 1516                await using var repair = connection.CreateCommand();
 1517                repair.CommandText = $"ALTER TABLE {Table} MODIFY state_json longtext CHARACTER SET utf8mb4 NOT NULL;";
 1518                await repair.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 1519            }
 520            else
 521            {
 1522                throw new InvalidOperationException(
 1523                    $"The MySQL durable-flow table '{_options.TableName}' stores state_json in the character set " +
 1524                    $"'{stateJsonColumn.CharacterSet ?? "(none)"}', which cannot hold every flow state the engine accept
 1525                    "JSON is arbitrary text, so a narrower set rejects updates under strict mode or silently truncates t
 1526                    $"state otherwise. Fix it with ALTER TABLE `{_options.TableName}` MODIFY state_json longtext CHARACT
 1527                    "NOT NULL; (tables this build creates get that character set automatically).");
 528            }
 529        }
 530
 145531        await VerifyFlowIdIsUniqueAsync(connection, cancellationToken).ConfigureAwait(false);
 142532        return true;
 143533    }
 534
 535    /// <summary>One column as <c>information_schema</c> reports it.</summary>
 536    private readonly record struct ActualColumn(
 1067537        string DataType,
 5538        string ColumnType,
 148539        string? Collation,
 1073540        bool Nullable,
 760541        long? MaxLength,
 455542        long? DateTimePrecision,
 297543        string? CharacterSet,
 3544        bool HasDefault,
 4545        string Extra)
 546    {
 547        /// <summary>
 548        /// Whether this store could insert a row without naming this column. True when the column
 549        /// is nullable, carries a default, auto-increments, or is computed by the database.
 550        /// </summary>
 551        internal bool IsWritableWithoutValue
 4552            => Nullable
 4553                || HasDefault
 4554                || Extra.Contains("auto_increment", StringComparison.OrdinalIgnoreCase)
 4555                || Extra.Contains("GENERATED", StringComparison.OrdinalIgnoreCase);
 556    }
 557
 558    /// <summary>
 559    /// What this store needs from one column, and the check that says so. Widths and precisions are
 560    /// MINIMA rather than exact matches: a wider flow_id or a higher-precision timestamp still
 561    /// satisfies every promise the store makes, and rejecting a more generous schema would be a
 562    /// false alarm. Too NARROW is not — <c>varchar(10)</c> passes a name-only check and then
 563    /// truncates or errors on the first 400-character id the public contract permits.
 564    /// </summary>
 8550565    private sealed record ExpectedColumn(string Name, string Declaration, string DataType, bool Nullable, long? Minimum 
 566    {
 567        internal string? Mismatch(ActualColumn actual)
 568        {
 1066569            if (!string.Equals(actual.DataType, DataType, StringComparison.OrdinalIgnoreCase))
 1570                return $"is a '{actual.DataType}'";
 1065571            if (actual.Nullable != Nullable)
 1572                return Nullable ? "is NOT NULL (this store writes NULL to it)" : "is nullable";
 1064573            if (Minimum is not { } minimum)
 304574                return null;
 575
 576            // CHARACTER_MAXIMUM_LENGTH for the string columns, DATETIME_PRECISION for the
 577            // timestamps: a datetime(0) column silently rounds the sub-second lease arithmetic this
 578            // store runs on UTC_TIMESTAMP(6), which is how two workers end up holding one lease.
 760579            var actualSize = actual.MaxLength ?? actual.DateTimePrecision;
 760580            return actualSize is { } size && size >= minimum
 760581                ? null
 760582                : $"holds {actualSize?.ToString() ?? "an unknown size"} where at least {minimum} is required";
 583        }
 584    }
 585
 586    /// <summary>
 587    /// The shape this store reads and writes. No default expressions are verified because none are
 588    /// load-bearing: every write names every column except the two lease fields, whose absence
 589    /// means NULL.
 590    /// </summary>
 4591    private static readonly ExpectedColumn[] ExpectedColumns =
 4592    [
 4593        new("flow_id", "varchar(400) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL", "varchar", Nullable: false, Mi
 4594        new("state_json", "longtext NOT NULL", "longtext", Nullable: false),
 4595        new("expires_at_utc", "datetime(6) NOT NULL", "datetime", Nullable: false, Minimum: 6),
 4596        new("updated_at_utc", "datetime(6) NOT NULL", "datetime", Nullable: false, Minimum: 6),
 4597        new("revision", "bigint NOT NULL DEFAULT 0", "bigint", Nullable: false),
 4598        new("lease_id", "varchar(64) NULL", "varchar", Nullable: true, Minimum: 64),
 4599        new("lease_expires_at_utc", "datetime(6) NULL", "datetime", Nullable: true, Minimum: 6)
 4600    ];
 601
 602    /// <summary>
 603    /// Requires a unique index keyed on the WHOLE of flow_id and nothing else. The PRIMARY KEY this
 604    /// store's DDL declares is the usual one, but any single-column unique index raises the 1062
 605    /// that <see cref="TryCreateAsync"/> reads as "already exists", so all of them are accepted.
 606    /// Two shapes are not:
 607    /// <list type="bullet">
 608    /// <item><description>
 609    /// A COMPOSITE unique key — it permits two rows with the same flow_id.
 610    /// </description></item>
 611    /// <item><description>
 612    /// A PREFIX key (<c>UNIQUE (flow_id(100))</c>, <c>SUB_PART = 100</c>) — the opposite failure,
 613    /// and the reason this asks about SUB_PART. It constrains only the first N characters, so two
 614    /// ids the library treats as distinct collide on 1062 and the second flow never starts. Ids run
 615    /// to 400 characters by contract, and prefix keys are a common way to fit an index under
 616    /// MySQL's key-length limit, so this is a plausible hand-written schema, not a hypothetical.
 617    /// </description></item>
 618    /// </list>
 619    /// </summary>
 620    private async Task VerifyFlowIdIsUniqueAsync(MySqlConnection connection, CancellationToken cancellationToken)
 621    {
 145622        await using var command = connection.CreateCommand();
 145623        command.CommandText =
 145624            """
 145625            SELECT 1
 145626            FROM information_schema.STATISTICS s
 145627            WHERE s.TABLE_SCHEMA = DATABASE() AND s.TABLE_NAME = @table
 145628              AND s.NON_UNIQUE = 0 AND s.COLUMN_NAME = 'flow_id' AND s.SEQ_IN_INDEX = 1
 145629              AND s.SUB_PART IS NULL
 145630              AND NOT EXISTS (
 145631                  SELECT 1 FROM information_schema.STATISTICS o
 145632                  WHERE o.TABLE_SCHEMA = s.TABLE_SCHEMA AND o.TABLE_NAME = s.TABLE_NAME
 145633                    AND o.INDEX_NAME = s.INDEX_NAME AND o.SEQ_IN_INDEX > 1)
 145634            LIMIT 1;
 145635            """;
 145636        command.Parameters.AddWithValue("@table", _options.TableName);
 637
 145638        if (await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false) is not null)
 639            return;
 640
 3641        throw new InvalidOperationException(
 3642            $"The MySQL durable-flow table '{_options.TableName}' has no unique key on the whole of flow_id. Starting a 
 3643            "insert-if-absent, and this store learns that a ledger already exists from MySQL's duplicate-key error. With
 3644            "key nothing reports the duplicate, so two concurrent starts of one flow id both succeed and the flow runs t
 3645            "a PREFIX key (flow_id(n)) the opposite happens, and two distinct ids sharing their first n characters colli
 3646            $"second never starts. Fix it with ALTER TABLE `{_options.TableName}` ADD PRIMARY KEY (flow_id); (tables thi
 3647            "creates declare it automatically).");
 142648    }
 649
 650    private async Task<bool> UpdateLeaseAsync(
 651        string flowId,
 652        string leaseId,
 653        TimeSpan leaseDuration,
 654        bool acquire,
 655        CancellationToken cancellationToken)
 656    {
 160657        DurableFlowStoreShared.ValidateLeaseArgs(flowId, leaseId, leaseDuration);
 658
 158659        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 158660        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 158661        await using var command = connection.CreateCommand();
 662        // Lease fencing runs entirely on the database clock: acquire steals only leases the
 663        // database considers expired, and renew/extend stays relative to UTC_TIMESTAMP(6), so
 664        // worker clock skew can never make two nodes hold the same lease.
 158665        command.CommandText =
 158666            $"""
 158667            UPDATE {Table}
 158668            SET lease_id = @lease_id, lease_expires_at_utc = {AddMilliseconds("@lease_ms")}
 158669            WHERE flow_id = @flow_id
 158670              AND expires_at_utc > UTC_TIMESTAMP(6)
 158671              AND {(acquire ? "(lease_id IS NULL OR lease_expires_at_utc <= UTC_TIMESTAMP(6) OR lease_id = @lease_id)" :
 158672            """;
 158673        command.Parameters.AddWithValue("@flow_id", flowId);
 158674        command.Parameters.AddWithValue("@lease_id", leaseId);
 158675        command.Parameters.AddWithValue("@lease_ms", DurableFlowStoreShared.ServerClockTtlMilliseconds(leaseDuration));
 158676        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 158677    }
 678
 679    // Row-count semantics: this store's lease renewal (and update fencing) treats
 680    // ExecuteNonQuery's result as ROWS MATCHED, which is MySqlConnector's default
 681    // (UseAffectedRows=false). EnsureCreatedAsync rejects a connection string that sets
 682    // UseAffectedRows=true before any of those UPDATEs can run.
 683    private Task<MySqlConnection> OpenConnectionAsync(CancellationToken cancellationToken)
 2465684        => DurableFlowStoreShared.OpenConnectionAsync<MySqlConnection>(_options.ConnectionString, cancellationToken);
 685
 2749686    private string Table => Quote(_options.TableName);
 138687    private string IndexName => Quote(DurableFlowStoreShared.DerivedName(_options.TableName, "_expires_idx", 64));
 2887688    private static string Quote(string identifier) => "`" + identifier.Replace("`", "``", StringComparison.Ordinal) + "`
 689}
 690}
Generated by: ReportGenerator 5.5.11.0
09/22/2026 - 09:41:04
GitHub | reportgenerator.io

Methods/Properties

AddMilliseconds(System.String)
.ctor(Microsoft.Extensions.Options.IOptions`1<AsyncResponse.DurableFlows.MySql.MySqlDurableFlowOptions>,Microsoft.Extensions.Logging.ILogger`1<AsyncResponse.DurableFlows.MySql.MySqlFlowStateStore>)
LoadAsync()
ValidateCreate(System.String,AsyncResponse.FlowState,System.TimeSpan)
TryCreateAsync()
ExistsAsync()
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()
get_DataType()
get_ColumnType()
get_Collation()
get_Nullable()
get_MaxLength()
get_DateTimePrecision()
get_CharacterSet()
get_HasDefault()
get_Extra()
get_IsWritableWithoutValue()
get_Name()
Mismatch(AsyncResponse.DurableFlows.MySql.MySqlFlowStateStore/ActualColumn)
.cctor()
VerifyFlowIdIsUniqueAsync()
UpdateLeaseAsync()
OpenConnectionAsync(System.Threading.CancellationToken)
get_Table()
get_IndexName()
Quote(System.String)