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

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

/_/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.
 20022            builder.Services.TryAddSingleton<MySqlFlowStateStore>();
 20023            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)
 90        => $"TIMESTAMPADD(MICROSECOND, {parameterName} * 1000, UTC_TIMESTAMP(6))";
 91
 92    private readonly MySqlDurableFlowOptions _options;
 93    private readonly SemaphoreSlim _ensureGate = new(1, 1);
 94    private long _lastPruneTicks;
 95    private volatile bool _created;
 96
 97    public MySqlFlowStateStore(IOptions<MySqlDurableFlowOptions> options, ILogger<MySqlFlowStateStore>? logger = null)
 98    {
 99        _logger = logger;
 100        _options = options.Value;
 101        _options.Validate();
 102    }
 103
 104    public async Task<FlowState?> LoadAsync(string flowId, CancellationToken cancellationToken = default)
 105    {
 106        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 107        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 108
 109        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 110        await using var command = connection.CreateCommand();
 111        command.CommandText = $"SELECT state_json, revision FROM {Table} WHERE flow_id = @flow_id AND expires_at_utc > U
 112        command.Parameters.AddWithValue("@flow_id", flowId);
 113
 114        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 115        if (!await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 116            return null;
 117
 118        return DurableFlowStoreShared.ReadState(flowId, reader.GetString(0), reader.GetInt64(1));
 119    }
 120
 121    /// <inheritdoc />
 122    public void ValidateCreate(string flowId, FlowState state, TimeSpan ttl)
 123    {
 124        DurableFlowStoreShared.ValidateCreate(flowId, state, ttl);
 125        if (_options.MaxStateBytes is not null)
 126            _ = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "MySQL");
 127    }
 128
 129    public async Task<bool> TryCreateAsync(string flowId, FlowState state, TimeSpan ttl, CancellationToken cancellationT
 130    {
 131        DurableFlowStoreShared.ValidateCreate(flowId, state, ttl);
 132        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "MySQL");
 133        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 134        if (DurableFlowStoreShared.ShouldPrune(ref _lastPruneTicks, _options.PruneInterval))
 135            await DurableFlowStoreShared.PruneQuietlyAsync(() => PruneExpiredAsync(cancellationToken), _options.PruneBud
 136
 137        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 138        await using var command = connection.CreateCommand();
 139        command.CommandText =
 140            $"""
 141            INSERT INTO {Table} (flow_id, state_json, expires_at_utc, updated_at_utc, revision)
 142            VALUES (@flow_id, @state_json, {AddMilliseconds("@ttl_ms")}, UTC_TIMESTAMP(6), @revision);
 143            """;
 144        command.Parameters.AddWithValue("@flow_id", flowId);
 145        command.Parameters.AddWithValue("@state_json", stateJson);
 146        command.Parameters.AddWithValue("@ttl_ms", DurableFlowStoreShared.ServerClockTtlMilliseconds(ttl));
 147        command.Parameters.AddWithValue("@revision", state.Revision);
 148        try
 149        {
 150            await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 151            return true;
 152        }
 153        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.
 163            if (!await ExistsAsync(connection, flowId, cancellationToken).ConfigureAwait(false))
 164                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.
 173        command.CommandText =
 174            $"""
 175            UPDATE {Table}
 176            SET state_json = @state_json,
 177                revision = @revision,
 178                lease_id = NULL,
 179                lease_expires_at_utc = NULL,
 180                updated_at_utc = UTC_TIMESTAMP(6),
 181                expires_at_utc = {AddMilliseconds("@ttl_ms")}
 182            WHERE flow_id = @flow_id AND expires_at_utc <= UTC_TIMESTAMP(6);
 183            """;
 184        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 185    }
 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    {
 198        await using var command = connection.CreateCommand();
 199        command.CommandText = $"SELECT 1 FROM {Table} WHERE flow_id = @flow_id LIMIT 1;";
 200        command.Parameters.AddWithValue("@flow_id", flowId);
 201        return await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false) is not null;
 202    }
 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    {
 212        DurableFlowStoreShared.ValidateUpdate(flowId, state, expectedRevision, ttl);
 213        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "MySQL");
 214        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 215
 216        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 217        await using var command = connection.CreateCommand();
 218        command.CommandText =
 219            $"""
 220            UPDATE {Table}
 221            SET state_json = @state_json,
 222                expires_at_utc = {AddMilliseconds("@ttl_ms")},
 223                updated_at_utc = UTC_TIMESTAMP(6),
 224                revision = @new_revision
 225            WHERE flow_id = @flow_id
 226              AND revision = @expected_revision
 227              AND expires_at_utc > UTC_TIMESTAMP(6)
 228              AND (@lease_id IS NULL OR (lease_id = @lease_id AND lease_expires_at_utc > UTC_TIMESTAMP(6)));
 229            """;
 230        command.Parameters.AddWithValue("@flow_id", flowId);
 231        command.Parameters.AddWithValue("@state_json", stateJson);
 232        command.Parameters.AddWithValue("@ttl_ms", DurableFlowStoreShared.ServerClockTtlMilliseconds(ttl));
 233        command.Parameters.AddWithValue("@expected_revision", expectedRevision);
 234        command.Parameters.AddWithValue("@new_revision", state.Revision);
 235        command.Parameters.AddWithValue("@lease_id", (object?)leaseId ?? DBNull.Value);
 236        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 237    }
 238
 239    public Task<bool> TryAcquireLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken canc
 240        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: true, cancellationToken);
 241
 242    public Task<bool> TryRenewLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken cancel
 243        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: false, cancellationToken);
 244
 245    public async Task ReleaseLeaseAsync(string flowId, string leaseId, CancellationToken cancellationToken = default)
 246    {
 247        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 248        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 249        await using var command = connection.CreateCommand();
 250        command.CommandText = $"UPDATE {Table} SET lease_id = NULL, lease_expires_at_utc = NULL WHERE flow_id = @flow_id
 251        command.Parameters.AddWithValue("@flow_id", flowId);
 252        command.Parameters.AddWithValue("@lease_id", leaseId);
 253        await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 254    }
 255
 256    /// <inheritdoc />
 257    public async Task<FlowLeaseObservation?> ObserveLeaseAsync(string flowId, CancellationToken cancellationToken = defa
 258    {
 259        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 260        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.
 267        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 268        await using var command = connection.CreateCommand();
 269        command.CommandText = $"SELECT lease_id, lease_expires_at_utc FROM {Table} WHERE flow_id = @flow_id;";
 270        command.Parameters.AddWithValue("@flow_id", flowId);
 271
 272        await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 273        if (!await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 274            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.
 280        return DurableFlowStoreShared.LeaseObservation(
 281            reader.IsDBNull(0) ? null : reader.GetString(0),
 282            reader.IsDBNull(1) ? null : DateTime.SpecifyKind(reader.GetDateTime(1), DateTimeKind.Unspecified));
 283    }
 284
 285    public async Task<bool> TryDeleteAsync(string flowId, CancellationToken cancellationToken = default)
 286    {
 287        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 288        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 289
 290        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 291        await using var command = connection.CreateCommand();
 292        command.CommandText = $"DELETE FROM {Table} WHERE flow_id = @flow_id;";
 293        command.Parameters.AddWithValue("@flow_id", flowId);
 294        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 295    }
 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.
 304        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 305        await using var command = connection.CreateCommand();
 306        command.CommandText = $"DELETE FROM {Table} WHERE expires_at_utc <= UTC_TIMESTAMP(6) LIMIT {DurableFlowStoreShar
 307        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 308    }
 309
 310    private async Task EnsureCreatedAsync(CancellationToken cancellationToken)
 311    {
 312        if (_created)
 313            return;
 314
 315        await _ensureGate.WaitAsync(cancellationToken).ConfigureAwait(false);
 316        try
 317        {
 318            if (_created)
 319                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.
 329            if (new MySqlConnectionStringBuilder(_options.ConnectionString!).UseAffectedRows)
 330            {
 331                throw new InvalidOperationException(
 332                    $"{nameof(MySqlDurableFlowOptions)}.{nameof(MySqlDurableFlowOptions.ConnectionString)} sets UseAffec
 333                    "which switches ExecuteNonQuery from rows-MATCHED to rows-CHANGED semantics and silently breaks this
 334                    "lease renewal and update fencing. Remove UseAffectedRows from the connection string; the MySqlConne
 335                    "default (false) is required.");
 336            }
 337
 338            await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 339            if (_options.AutoCreateSchema)
 340            {
 341                await using var command = connection.CreateCommand();
 342                command.CommandText =
 343                    $"""
 344                    CREATE TABLE IF NOT EXISTS {Table} (
 345                        flow_id varchar(400) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL PRIMARY KEY,
 346                        state_json longtext CHARACTER SET utf8mb4 NOT NULL,
 347                        expires_at_utc datetime(6) NOT NULL,
 348                        updated_at_utc datetime(6) NOT NULL,
 349                        revision bigint NOT NULL DEFAULT 0,
 350                        lease_id varchar(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NULL,
 351                        lease_expires_at_utc datetime(6) NULL,
 352                        INDEX {IndexName} (expires_at_utc)
 353                    );
 354                    """;
 355                await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 356            }
 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.
 362            _created = await VerifyFlowTableAsync(connection, cancellationToken).ConfigureAwait(false);
 363        }
 364        finally
 365        {
 366            _ensureGate.Release();
 367        }
 368    }
 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    {
 393        var columns = new Dictionary<string, ActualColumn>(StringComparer.OrdinalIgnoreCase);
 394        await using (var command = connection.CreateCommand())
 395        {
 396            command.CommandText =
 397                """
 398                SELECT COLUMN_NAME, DATA_TYPE, COLUMN_TYPE, COLLATION_NAME, IS_NULLABLE,
 399                       CHARACTER_MAXIMUM_LENGTH, DATETIME_PRECISION, CHARACTER_SET_NAME,
 400                       COLUMN_DEFAULT, EXTRA
 401                FROM information_schema.COLUMNS
 402                WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = @table;
 403                """;
 404            command.Parameters.AddWithValue("@table", _options.TableName);
 405            await using var reader = await command.ExecuteReaderAsync(cancellationToken).ConfigureAwait(false);
 406            while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
 407            {
 408                columns[reader.GetString(0)] = new ActualColumn(
 409                    DataType: reader.GetString(1),
 410                    ColumnType: reader.GetString(2),
 411                    Collation: reader.IsDBNull(3) ? null : reader.GetString(3),
 412                    Nullable: string.Equals(reader.GetString(4), "YES", StringComparison.OrdinalIgnoreCase),
 413                    MaxLength: reader.IsDBNull(5) ? null : reader.GetInt64(5),
 414                    DateTimePrecision: reader.IsDBNull(6) ? null : reader.GetInt64(6),
 415                    CharacterSet: reader.IsDBNull(7) ? null : reader.GetString(7),
 416                    HasDefault: !reader.IsDBNull(8),
 417                    Extra: reader.IsDBNull(9) ? string.Empty : reader.GetString(9));
 418            }
 419        }
 420
 421        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.
 427            return false;
 428        }
 429
 430        foreach (var expected in ExpectedColumns)
 431        {
 432            if (!columns.TryGetValue(expected.Name, out var actual))
 433            {
 434                throw new InvalidOperationException(
 435                    $"The MySQL durable-flow table '{_options.TableName}' has no '{expected.Name}' column. It was create
 436                    "earlier build or by hand and does not match the shape this store reads and writes " +
 437                    $"({string.Join(", ", ExpectedColumns.Select(column => $"{column.Name} {column.Declaration}"))}). Re
 438                    "or add the missing columns — the DDL is in docs/durable-flow-state-stores.md.");
 439            }
 440
 441            if (expected.Mismatch(actual) is { } mismatch)
 442            {
 443                throw new InvalidOperationException(
 444                    $"The MySQL durable-flow table '{_options.TableName}' declares {expected.Name} as '{actual.ColumnTyp
 445                    $"{(actual.Nullable ? " NULL" : " NOT NULL")}', which {mismatch}. This store needs " +
 446                    $"{expected.Name} {expected.Declaration}. Fix it with " +
 447                    $"ALTER TABLE `{_options.TableName}` MODIFY {expected.Name} {expected.Declaration}; " +
 448                    "(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.
 456        foreach (var (name, actual) in columns)
 457        {
 458            if (ExpectedColumns.Any(expected => string.Equals(expected.Name, name, StringComparison.OrdinalIgnoreCase))
 459                || actual.IsWritableWithoutValue)
 460            {
 461                continue;
 462            }
 463
 464            throw new InvalidOperationException(
 465                $"The MySQL durable-flow table '{_options.TableName}' has an extra column '{name}' ({actual.ColumnType} 
 466                "with no default. This store writes only its own columns, so every flow creation would fail on that colu
 467                "it a default, make it nullable or generated, or move it to a table of your own.");
 468        }
 469
 470        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.
 476        if (flowIdColumn.CharacterSet is not { } characterSet
 477            || !characterSet.Equals("utf8mb4", StringComparison.OrdinalIgnoreCase))
 478        {
 479            throw new InvalidOperationException(
 480                $"The MySQL durable-flow table '{_options.TableName}' stores flow_id in the character set " +
 481                $"'{flowIdColumn.CharacterSet ?? "(none)"}', which cannot hold every id the engine accepts. Flow ids are
 482                "text — this store's contract bounds their length, not their alphabet — so a narrower set rejects or man
 483                $"that are perfectly valid. Fix it with ALTER TABLE `{_options.TableName}` MODIFY flow_id varchar(400) C
 484                "SET utf8mb4 COLLATE utf8mb4_bin NOT NULL; (tables this build creates get that character set automatical
 485        }
 486
 487        var collation = flowIdColumn.Collation;
 488        if (collation is null || !collation.EndsWith("_bin", StringComparison.OrdinalIgnoreCase))
 489        {
 490            throw new InvalidOperationException(
 491                $"The MySQL durable-flow table '{_options.TableName}' stores flow_id with the collation '{collation ?? "
 492                "which is not binary. Flow ids are compared ordinally by the engine, so ids differing only in case (or a
 493                "width) collide on the primary key: the second flow fails to start and a load returns the other run's st
 494                $"with ALTER TABLE `{_options.TableName}` MODIFY flow_id varchar(400) CHARACTER SET utf8mb4 COLLATE utf8
 495                "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.
 503        var stateJsonColumn = columns["state_json"];
 504        if (stateJsonColumn.CharacterSet is not { } stateJsonCharacterSet
 505            || !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.
 514            if (_options.AutoCreateSchema)
 515            {
 516                await using var repair = connection.CreateCommand();
 517                repair.CommandText = $"ALTER TABLE {Table} MODIFY state_json longtext CHARACTER SET utf8mb4 NOT NULL;";
 518                await repair.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false);
 519            }
 520            else
 521            {
 522                throw new InvalidOperationException(
 523                    $"The MySQL durable-flow table '{_options.TableName}' stores state_json in the character set " +
 524                    $"'{stateJsonColumn.CharacterSet ?? "(none)"}', which cannot hold every flow state the engine accept
 525                    "JSON is arbitrary text, so a narrower set rejects updates under strict mode or silently truncates t
 526                    $"state otherwise. Fix it with ALTER TABLE `{_options.TableName}` MODIFY state_json longtext CHARACT
 527                    "NOT NULL; (tables this build creates get that character set automatically).");
 528            }
 529        }
 530
 531        await VerifyFlowIdIsUniqueAsync(connection, cancellationToken).ConfigureAwait(false);
 532        return true;
 533    }
 534
 535    /// <summary>One column as <c>information_schema</c> reports it.</summary>
 536    private readonly record struct ActualColumn(
 537        string DataType,
 538        string ColumnType,
 539        string? Collation,
 540        bool Nullable,
 541        long? MaxLength,
 542        long? DateTimePrecision,
 543        string? CharacterSet,
 544        bool HasDefault,
 545        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
 552            => Nullable
 553                || HasDefault
 554                || Extra.Contains("auto_increment", StringComparison.OrdinalIgnoreCase)
 555                || 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>
 565    private sealed record ExpectedColumn(string Name, string Declaration, string DataType, bool Nullable, long? Minimum 
 566    {
 567        internal string? Mismatch(ActualColumn actual)
 568        {
 569            if (!string.Equals(actual.DataType, DataType, StringComparison.OrdinalIgnoreCase))
 570                return $"is a '{actual.DataType}'";
 571            if (actual.Nullable != Nullable)
 572                return Nullable ? "is NOT NULL (this store writes NULL to it)" : "is nullable";
 573            if (Minimum is not { } minimum)
 574                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.
 579            var actualSize = actual.MaxLength ?? actual.DateTimePrecision;
 580            return actualSize is { } size && size >= minimum
 581                ? null
 582                : $"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>
 591    private static readonly ExpectedColumn[] ExpectedColumns =
 592    [
 593        new("flow_id", "varchar(400) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL", "varchar", Nullable: false, Mi
 594        new("state_json", "longtext NOT NULL", "longtext", Nullable: false),
 595        new("expires_at_utc", "datetime(6) NOT NULL", "datetime", Nullable: false, Minimum: 6),
 596        new("updated_at_utc", "datetime(6) NOT NULL", "datetime", Nullable: false, Minimum: 6),
 597        new("revision", "bigint NOT NULL DEFAULT 0", "bigint", Nullable: false),
 598        new("lease_id", "varchar(64) NULL", "varchar", Nullable: true, Minimum: 64),
 599        new("lease_expires_at_utc", "datetime(6) NULL", "datetime", Nullable: true, Minimum: 6)
 600    ];
 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    {
 622        await using var command = connection.CreateCommand();
 623        command.CommandText =
 624            """
 625            SELECT 1
 626            FROM information_schema.STATISTICS s
 627            WHERE s.TABLE_SCHEMA = DATABASE() AND s.TABLE_NAME = @table
 628              AND s.NON_UNIQUE = 0 AND s.COLUMN_NAME = 'flow_id' AND s.SEQ_IN_INDEX = 1
 629              AND s.SUB_PART IS NULL
 630              AND NOT EXISTS (
 631                  SELECT 1 FROM information_schema.STATISTICS o
 632                  WHERE o.TABLE_SCHEMA = s.TABLE_SCHEMA AND o.TABLE_NAME = s.TABLE_NAME
 633                    AND o.INDEX_NAME = s.INDEX_NAME AND o.SEQ_IN_INDEX > 1)
 634            LIMIT 1;
 635            """;
 636        command.Parameters.AddWithValue("@table", _options.TableName);
 637
 638        if (await command.ExecuteScalarAsync(cancellationToken).ConfigureAwait(false) is not null)
 639            return;
 640
 641        throw new InvalidOperationException(
 642            $"The MySQL durable-flow table '{_options.TableName}' has no unique key on the whole of flow_id. Starting a 
 643            "insert-if-absent, and this store learns that a ledger already exists from MySQL's duplicate-key error. With
 644            "key nothing reports the duplicate, so two concurrent starts of one flow id both succeed and the flow runs t
 645            "a PREFIX key (flow_id(n)) the opposite happens, and two distinct ids sharing their first n characters colli
 646            $"second never starts. Fix it with ALTER TABLE `{_options.TableName}` ADD PRIMARY KEY (flow_id); (tables thi
 647            "creates declare it automatically).");
 648    }
 649
 650    private async Task<bool> UpdateLeaseAsync(
 651        string flowId,
 652        string leaseId,
 653        TimeSpan leaseDuration,
 654        bool acquire,
 655        CancellationToken cancellationToken)
 656    {
 657        DurableFlowStoreShared.ValidateLeaseArgs(flowId, leaseId, leaseDuration);
 658
 659        await EnsureCreatedAsync(cancellationToken).ConfigureAwait(false);
 660        await using var connection = await OpenConnectionAsync(cancellationToken).ConfigureAwait(false);
 661        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.
 665        command.CommandText =
 666            $"""
 667            UPDATE {Table}
 668            SET lease_id = @lease_id, lease_expires_at_utc = {AddMilliseconds("@lease_ms")}
 669            WHERE flow_id = @flow_id
 670              AND expires_at_utc > UTC_TIMESTAMP(6)
 671              AND {(acquire ? "(lease_id IS NULL OR lease_expires_at_utc <= UTC_TIMESTAMP(6) OR lease_id = @lease_id)" :
 672            """;
 673        command.Parameters.AddWithValue("@flow_id", flowId);
 674        command.Parameters.AddWithValue("@lease_id", leaseId);
 675        command.Parameters.AddWithValue("@lease_ms", DurableFlowStoreShared.ServerClockTtlMilliseconds(leaseDuration));
 676        return await command.ExecuteNonQueryAsync(cancellationToken).ConfigureAwait(false) > 0;
 677    }
 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)
 684        => DurableFlowStoreShared.OpenConnectionAsync<MySqlConnection>(_options.ConnectionString, cancellationToken);
 685
 686    private string Table => Quote(_options.TableName);
 687    private string IndexName => Quote(DurableFlowStoreShared.DerivedName(_options.TableName, "_expires_idx", 64));
 688    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

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