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

Information
Class: AsyncResponse.DurableFlows.EFCore.EFCoreDurableFlowModelBuilderExtensions
Assembly: AsyncResponse.DurableFlows.EFCore
File(s): /_/src/DurableFlows/AsyncResponse.DurableFlows.EFCore/EFCoreDurableFlows.cs
Line coverage
100%
Covered lines: 25
Uncovered lines: 0
Coverable lines: 25
Total lines: 603
Line coverage: 100%
Branch coverage
100%
Covered branches: 2
Total branches: 2
Branch coverage: 100%
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Crap Score Cyclomatic complexity Line coverage
ConfigureAsyncResponseDurableFlows(...)100%22100%

File(s)

/_/src/DurableFlows/AsyncResponse.DurableFlows.EFCore/EFCoreDurableFlows.cs

#LineLine coverage
 1using System.Diagnostics.CodeAnalysis;
 2using AsyncResponse;
 3using AsyncResponse.DurableFlows.EFCore;
 4using AsyncResponse.DurableFlows.Internal;
 5using Microsoft.EntityFrameworkCore;
 6using Microsoft.EntityFrameworkCore.Infrastructure;
 7using Microsoft.Extensions.DependencyInjection;
 8using Microsoft.Extensions.DependencyInjection.Extensions;
 9using Microsoft.Extensions.Logging;
 10using Microsoft.Extensions.Options;
 11
 12namespace Microsoft.Extensions.DependencyInjection
 13{
 14    /// <summary>DI registration for the Entity Framework Core durable-flow state store.</summary>
 15    public static class EFCoreDurableFlowServiceCollectionExtensions
 16    {
 17        /// <summary>
 18        /// Stores durable-flow state in a table hosted by the application's own
 19        /// <typeparamref name="TContext"/>. Map the table into the context's model with
 20        /// <see cref="EFCoreDurableFlowModelBuilderExtensions.ConfigureAsyncResponseDurableFlows"/>
 21        /// so it rides the application's migration pipeline; the store itself never runs DDL.
 22        /// <para>
 23        /// Each operation resolves a fresh context: from <see cref="IDbContextFactory{TContext}"/>
 24        /// when one is registered (<c>AddDbContextFactory</c>), otherwise the scoped
 25        /// <typeparamref name="TContext"/> from a new service scope (<c>AddDbContext</c>). Parallel
 26        /// flow executions therefore never share a <see cref="DbContext"/> instance.
 27        /// </para>
 28        /// </summary>
 29        public static AsyncResponseRegistrationBuilder WithEFCoreDurableFlows<[DynamicallyAccessedMembers(DynamicallyAcc
 30            this AsyncResponseRegistrationBuilder builder,
 31            Action<EFCoreDurableFlowOptions>? configure = null)
 32            where TContext : DbContext
 33        {
 34            // Singleton on purpose: the store holds no DbContext (each operation leases one, see
 35            // above), and the executor resolves the store from a fresh scope per flow execution —
 36            // a scoped store would redo the mapped-model check on every run.
 37            builder.Services.TryAddSingleton<EFCoreFlowStateStore<TContext>>();
 38            return builder.WithDurableFlows<EFCoreFlowStateStore<TContext>, EFCoreDurableFlowOptions>(configure);
 39        }
 40    }
 41}
 42
 43namespace AsyncResponse.DurableFlows.EFCore
 44{
 45/// <summary>Options for the Entity Framework Core durable-flow state store.</summary>
 46public sealed class EFCoreDurableFlowOptions : DurableFlowOptions
 47{
 48    /// <summary>
 49    /// How often <see cref="EFCoreFlowStateStore{TContext}.TryCreateAsync"/> opportunistically deletes
 50    /// one bounded batch (1000 rows) of expired rows (loads already treat expired state as absent;
 51    /// pruning bounds table growth). Zero or negative prunes on every save. Default: 5 minutes.
 52    /// </summary>
 53    public TimeSpan PruneInterval { get; set; } = TimeSpan.FromMinutes(5);
 54
 55    /// <summary>
 56    /// Wall-clock budget one opportunistic prune may spend draining expired rows in batches of
 57    /// 1000 after its first batch (the first always runs). A single batch per interval capped
 58    /// cleanup at ~3 rows/second, which any busier instance outgrew forever; the prune now drains
 59    /// batches until one comes back short or this budget lapses, and reports the outcome on the
 60    /// <c>AsyncResponse</c> meter (<c>asyncresponse.flow_state.pruned_rows</c>,
 61    /// <c>prune_failures</c>, <c>prune_budget_exhausted</c>) and the store's logger. The create
 62    /// that triggers the prune waits for it, so this bounds that create's added latency. Zero
 63    /// keeps the historical single batch. Default: 2 seconds.
 64    /// </summary>
 65    public TimeSpan PruneBudget { get; set; } = DurableFlowStoreShared.DefaultPruneBudget;
 66
 67    /// <summary>
 68    /// Maximum serialized flow-state size in bytes accepted by writes; oversized ledgers fail fast
 69    /// with an actionable error instead of an opaque provider error. Default: <c>null</c>
 70    /// (unlimited — relational text/blob columns are effectively unbounded), settable as an
 71    /// operator budget.
 72    /// </summary>
 73    public long? MaxStateBytes { get; set; }
 74}
 75
 76/// <summary>
 77/// One durable-flow ledger row, mapped into the application's <see cref="DbContext"/> by
 78/// <see cref="EFCoreDurableFlowModelBuilderExtensions.ConfigureAsyncResponseDurableFlows"/>.
 79/// Column names match the other AsyncResponse.DurableFlows.* relational packages, so the table is
 80/// interchangeable with theirs.
 81/// </summary>
 82public sealed class DurableFlowStateRecord
 83{
 84    /// <summary>The flow run id (primary key).</summary>
 85    public string FlowId { get; set; } = string.Empty;
 86
 87    /// <summary>The serialized <see cref="FlowState"/> ledger.</summary>
 88    public string StateJson { get; set; } = string.Empty;
 89
 90    /// <summary>UTC instant after which the row is treated as absent and eligible for pruning.</summary>
 91    public DateTime ExpiresAtUtc { get; set; }
 92
 93    /// <summary>UTC instant of the last save.</summary>
 94    public DateTime UpdatedAtUtc { get; set; }
 95
 96    /// <summary>Optimistic-concurrency revision of the durable ledger.</summary>
 97    public long Revision { get; set; }
 98
 99    /// <summary>Current execution-lease owner, when a worker is running the flow.</summary>
 100    public string? LeaseId { get; set; }
 101
 102    /// <summary>UTC expiry of the current execution lease.</summary>
 103    public DateTime? LeaseExpiresAtUtc { get; set; }
 104}
 105
 106/// <summary>
 107/// What a provider needs from the <c>flow_id</c> collation, when it needs anything at all. Kept
 108/// apart from the store so the rules are one table rather than a chain of conditions inside a
 109/// generic type — and so both branches can be exercised without dragging in every EF Core provider.
 110/// </summary>
 111internal static class FlowIdCollationRules
 112{
 113    /// <summary>
 114    /// The rules for a provider whose DEFAULT collation folds case, or <c>null</c> when the default
 115    /// is already ordinal (PostgreSQL and SQLite compare byte-wise) or the provider is unknown — a
 116    /// third-party provider gets the benefit of the doubt rather than a startup failure it has no
 117    /// documented way to satisfy.
 118    /// </summary>
 119    internal static CaseFoldingProviderRules? CaseFoldingProvider(string? providerName) => providerName switch
 120    {
 121        not null when providerName.Contains("SqlServer", StringComparison.OrdinalIgnoreCase) => new(
 122            "SQL Server",
 123            nameof(AsyncResponseFlowIdCollations.SqlServer),
 124            AsyncResponseFlowIdCollations.SqlServer,
 125            "_BIN2 collation",
 126            static c => c.Contains("_BIN", StringComparison.OrdinalIgnoreCase)),
 127        not null when providerName.Contains("MySql", StringComparison.OrdinalIgnoreCase)
 128            || providerName.Contains("Pomelo", StringComparison.OrdinalIgnoreCase) => new(
 129            "MySQL",
 130            nameof(AsyncResponseFlowIdCollations.MySql),
 131            AsyncResponseFlowIdCollations.MySql,
 132            "_bin collation",
 133            static c => c.EndsWith("_bin", StringComparison.OrdinalIgnoreCase)),
 134        _ => null
 135    };
 136
 137    /// <summary>One provider's answer to "which collations compare byte-wise, and what to suggest".</summary>
 138    internal sealed record CaseFoldingProviderRules(
 139        string Name,
 140        string ConstantName,
 141        string Recommended,
 142        string OrdinalDescription,
 143        Func<string, bool> IsOrdinal);
 144}
 145
 146/// <summary>
 147/// Well-known case-sensitive collations for the <c>flow_id</c> key column, one per mainstream
 148/// provider. Pass one to <see cref="EFCoreDurableFlowModelBuilderExtensions.ConfigureAsyncResponseDurableFlows"/>
 149/// when the database's own collation is case-insensitive — the SQL Server and MySQL defaults are,
 150/// and under them two flow ids differing only in case collide on the primary key while the engine
 151/// treats them as two different runs.
 152/// </summary>
 153public static class AsyncResponseFlowIdCollations
 154{
 155    /// <summary>SQL Server: binary, code-point ordered.</summary>
 156    public const string SqlServer = "Latin1_General_100_BIN2";
 157
 158    /// <summary>MySQL / MariaDB: binary comparison over utf8mb4.</summary>
 159    public const string MySql = "utf8mb4_bin";
 160
 161    /// <summary>PostgreSQL: the C locale, which compares byte-wise.</summary>
 162    public const string PostgreSql = "C";
 163
 164    /// <summary>SQLite: the default, already case-sensitive — named for completeness.</summary>
 165    public const string Sqlite = "BINARY";
 166}
 167
 168/// <summary>Maps the durable-flow state table into an application model.</summary>
 169public static class EFCoreDurableFlowModelBuilderExtensions
 170{
 171    /// <summary>Default durable-flow state table name (shared with the other relational packages).</summary>
 172    public const string DefaultTableName = "asyncresponse_flow_state";
 173
 174    /// <summary>
 175    /// Model annotation recording the <c>flowIdCollation</c> the mapping was configured with. The
 176    /// store reads it at startup to refuse a case-folding provider left on its default; the
 177    /// property's own collation cannot be used for that, because EF Core strips relational
 178    /// configuration the runtime never reads out of the runtime model.
 179    /// </summary>
 180    internal const string FlowIdCollationAnnotation = "AsyncResponse:FlowIdCollation";
 181
 182    /// <summary>
 183    /// Maps <see cref="DurableFlowStateRecord"/> to the durable-flow state table. Call from
 184    /// <c>OnModelCreating</c>; the table then flows through the application's normal EF Core
 185    /// migrations (or <c>EnsureCreated</c>) like any other entity.
 186    /// </summary>
 187    /// <param name="modelBuilder">The application model builder.</param>
 188    /// <param name="tableName">Table name. Default: <see cref="DefaultTableName"/>.</param>
 189    /// <param name="schema">Optional schema; <c>null</c> uses the provider default.</param>
 190    /// <param name="flowIdCollation">
 191    /// Collation for the <c>flow_id</c> key column. Flow ids are compared ORDINALLY by the engine,
 192    /// so the column must be case-sensitive — but this package runs no DDL and cannot know which
 193    /// provider the application points at, and both the SQL Server and MySQL defaults are
 194    /// case-INSENSITIVE, which makes <c>flow-a</c> and <c>FLOW-A</c> one key: the second create
 195    /// fails as a duplicate and a load returns the other run's state. Pass the matching
 196    /// <see cref="AsyncResponseFlowIdCollations"/> constant (the sibling PostgreSQL, SQL Server,
 197    /// and MySQL packages pin this in their own DDL). <c>null</c> keeps the database default.
 198    /// </param>
 199    public static ModelBuilder ConfigureAsyncResponseDurableFlows(
 200        this ModelBuilder modelBuilder,
 201        string tableName = DefaultTableName,
 202        string? schema = null,
 203        string? flowIdCollation = null)
 204    {
 85205        ArgumentNullException.ThrowIfNull(modelBuilder);
 85206        ArgumentException.ThrowIfNullOrWhiteSpace(tableName);
 207
 85208        modelBuilder.Entity<DurableFlowStateRecord>(entity =>
 85209        {
 85210            entity.ToTable(tableName, schema);
 85211            entity.HasKey(r => r.FlowId);
 85212            // 400 matches the sibling packages' key column and stays inside every mainstream
 85213            // provider's index-key size limit (SQL Server 900 bytes, MySQL 3072 bytes).
 85214            entity.Property(r => r.FlowId).HasColumnName("flow_id").HasMaxLength(400);
 85215            if (!string.IsNullOrWhiteSpace(flowIdCollation))
 85216            {
 78217                entity.Property(r => r.FlowId).UseCollation(flowIdCollation);
 85218                // Also recorded as a model annotation, which survives into the runtime model the
 85219                // store can actually read at startup.
 78220                entity.HasAnnotation(FlowIdCollationAnnotation, flowIdCollation);
 85221            }
 85222            entity.Property(r => r.StateJson).HasColumnName("state_json").IsRequired();
 85223            entity.Property(r => r.ExpiresAtUtc).HasColumnName("expires_at_utc");
 85224            entity.Property(r => r.UpdatedAtUtc).HasColumnName("updated_at_utc");
 85225            entity.Property(r => r.Revision).HasColumnName("revision").HasDefaultValue(0L);
 85226            entity.Property(r => r.LeaseId).HasColumnName("lease_id").HasMaxLength(64);
 85227            entity.Property(r => r.LeaseExpiresAtUtc).HasColumnName("lease_expires_at_utc");
 85228            entity.HasIndex(r => r.ExpiresAtUtc).HasDatabaseName($"{tableName}_expires_idx");
 170229        });
 230
 85231        return modelBuilder;
 232    }
 233}
 234
 235/// <summary>
 236/// Entity Framework Core implementation of <see cref="IFlowStateStore"/> over an
 237/// application-owned <typeparamref name="TContext"/>. Requires a relational provider
 238/// (deletes and updates use <c>ExecuteDeleteAsync</c>/<c>ExecuteUpdateAsync</c>).
 239/// </summary>
 240public sealed class EFCoreFlowStateStore<[DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.PublicConstructors |
 241    where TContext : DbContext
 242{
 243    private readonly ILogger<EFCoreFlowStateStore<TContext>>? _logger;
 244
 245    // Time authority: this store deliberately keeps the app clock (DateTime.UtcNow) for expiry
 246    // and lease comparisons. It is provider-agnostic LINQ — there is no portable way to reference
 247    // the database server's clock in a translated expression — so multi-node deployments should
 248    // either keep worker clocks synchronized well inside the lease window or use one of the
 249    // provider-specific relational stores, which run all time math on the database clock.
 250    private readonly IServiceScopeFactory _scopeFactory;
 251    private readonly EFCoreDurableFlowOptions _options;
 252    private long _lastPruneTicks;
 253    private volatile bool _modelChecked;
 254
 255    public EFCoreFlowStateStore(IServiceScopeFactory scopeFactory, IOptions<EFCoreDurableFlowOptions> options, ILogger<E
 256    {
 257        _scopeFactory = scopeFactory;
 258        _options = options.Value;
 259        _logger = logger;
 260        DurableFlowStoreShared.ValidateMaxStateBytes(_options.MaxStateBytes, nameof(EFCoreDurableFlowOptions));
 261        DurableFlowStoreShared.ValidatePruneBudget(_options.PruneBudget, nameof(EFCoreDurableFlowOptions));
 262    }
 263
 264    public async Task<FlowState?> LoadAsync(string flowId, CancellationToken cancellationToken = default)
 265    {
 266        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 267        await using var lease = await LeaseContextAsync(cancellationToken).ConfigureAwait(false);
 268
 269        var now = DateTime.UtcNow;
 270        // Named-record projection, not an anonymous type: anonymous projections lower to the
 271        // RequiresUnreferencedCode Expression.New(ctor, args, members) overload, which ILC trim
 272        // analysis rejects in Native AOT publishes even though the Roslyn analyzer stays quiet.
 273        var record = await Records(lease.Context)
 274            .AsNoTracking()
 275            .Where(r => r.FlowId == flowId && r.ExpiresAtUtc > now)
 276            .Select(r => new StateRow(r.StateJson, r.Revision))
 277            .FirstOrDefaultAsync(cancellationToken)
 278            .ConfigureAwait(false);
 279
 280        if (record is null)
 281            return null;
 282
 283        return DurableFlowStoreShared.ReadState(flowId, record.StateJson, record.Revision);
 284    }
 285
 286    /// <inheritdoc />
 287    public void ValidateCreate(string flowId, FlowState state, TimeSpan ttl)
 288    {
 289        DurableFlowStoreShared.ValidateCreate(flowId, state, ttl);
 290        if (_options.MaxStateBytes is not null)
 291            _ = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "EF Core");
 292    }
 293
 294    public async Task<bool> TryCreateAsync(string flowId, FlowState state, TimeSpan ttl, CancellationToken cancellationT
 295    {
 296        DurableFlowStoreShared.ValidateCreate(flowId, state, ttl);
 297        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "EF Core");
 298        await using var lease = await LeaseContextAsync(cancellationToken).ConfigureAwait(false);
 299        var db = lease.Context;
 300        var now = DateTime.UtcNow;
 301
 302        if (DurableFlowStoreShared.ShouldPrune(ref _lastPruneTicks, _options.PruneInterval))
 303            await DurableFlowStoreShared.PruneQuietlyAsync(() => PruneExpiredAsync(db, cancellationToken), _options.Prun
 304
 305        // Replace an expired ledger IN PLACE, in one statement (sibling parity: PostgreSQL
 306        // `ON CONFLICT ... DO UPDATE ... WHERE expired`, SQL Server/Oracle `MERGE ... WHEN MATCHED
 307        // ... WHERE`). Delete-then-insert spanned two transactions, and a failure between them
 308        // destroyed the expired row with no replacement. The lease columns are cleared as the
 309        // siblings clear them: the replaced ledger is a fresh, unleased run.
 310        var expiresAtUtc = DurableFlowStoreShared.AddSaturating(now, ttl);
 311        var revision = state.Revision;
 312        var replaced = await Records(db)
 313            .Where(r => r.FlowId == flowId && r.ExpiresAtUtc <= now)
 314            .ExecuteUpdateAsync(setters => setters
 315                .SetProperty(r => r.StateJson, stateJson)
 316                .SetProperty(r => r.ExpiresAtUtc, expiresAtUtc)
 317                .SetProperty(r => r.UpdatedAtUtc, now)
 318                .SetProperty(r => r.Revision, revision)
 319                .SetProperty(r => r.LeaseId, (string?)null)
 320                .SetProperty(r => r.LeaseExpiresAtUtc, (DateTime?)null), cancellationToken)
 321            .ConfigureAwait(false);
 322        if (replaced > 0)
 323            return true;
 324
 325        db.Add(new DurableFlowStateRecord
 326        {
 327            FlowId = flowId,
 328            StateJson = stateJson,
 329            ExpiresAtUtc = expiresAtUtc,
 330            UpdatedAtUtc = now,
 331            Revision = revision
 332        });
 333        try
 334        {
 335            await db.SaveChangesAsync(cancellationToken).ConfigureAwait(false);
 336            return true;
 337        }
 338        catch (DbUpdateException)
 339        {
 340            db.ChangeTracker.Clear();
 341            // Provider-agnostic duplicate-key detection is not reliable. Verify that another
 342            // creator actually owns this id; otherwise preserve the real database failure instead
 343            // of misreporting truncation, trigger, permission, or schema errors as "already exists".
 344            if (await Records(db)
 345                    .AsNoTracking()
 346                    .AnyAsync(r => r.FlowId == flowId, cancellationToken)
 347                    .ConfigureAwait(false))
 348                return false;
 349
 350            throw;
 351        }
 352    }
 353
 354    public async Task<bool> TryUpdateAsync(
 355        string flowId,
 356        FlowState state,
 357        long expectedRevision,
 358        TimeSpan ttl,
 359        string? leaseId = null,
 360        CancellationToken cancellationToken = default)
 361    {
 362        DurableFlowStoreShared.ValidateUpdate(flowId, state, expectedRevision, ttl);
 363        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "EF Core");
 364        await using var lease = await LeaseContextAsync(cancellationToken).ConfigureAwait(false);
 365        var now = DateTime.UtcNow;
 366        var expiresAtUtc = DurableFlowStoreShared.AddSaturating(now, ttl);
 367        var query = Records(lease.Context).Where(r =>
 368            r.FlowId == flowId
 369            && r.Revision == expectedRevision
 370            && r.ExpiresAtUtc > now
 371            && (leaseId == null || (r.LeaseId == leaseId && r.LeaseExpiresAtUtc > now)));
 372        var updated = await query.ExecuteUpdateAsync(setters => setters
 373                .SetProperty(r => r.StateJson, stateJson)
 374                .SetProperty(r => r.ExpiresAtUtc, expiresAtUtc)
 375                .SetProperty(r => r.UpdatedAtUtc, now)
 376                .SetProperty(r => r.Revision, state.Revision), cancellationToken)
 377            .ConfigureAwait(false);
 378        return updated > 0;
 379    }
 380
 381    public Task<bool> TryAcquireLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken canc
 382        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: true, cancellationToken);
 383
 384    public Task<bool> TryRenewLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken cancel
 385        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: false, cancellationToken);
 386
 387    public async Task ReleaseLeaseAsync(string flowId, string leaseId, CancellationToken cancellationToken = default)
 388    {
 389        await using var lease = await LeaseContextAsync(cancellationToken).ConfigureAwait(false);
 390        await Records(lease.Context)
 391            .Where(r => r.FlowId == flowId && r.LeaseId == leaseId)
 392            .ExecuteUpdateAsync(setters => setters
 393                .SetProperty(r => r.LeaseId, (string?)null)
 394                .SetProperty(r => r.LeaseExpiresAtUtc, (DateTime?)null), cancellationToken)
 395            .ConfigureAwait(false);
 396    }
 397
 398    /// <inheritdoc />
 399    public async Task<FlowLeaseObservation?> ObserveLeaseAsync(string flowId, CancellationToken cancellationToken = defa
 400    {
 401        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 402        await using var lease = await LeaseContextAsync(cancellationToken).ConfigureAwait(false);
 403
 404        // The two lease columns exactly as stored — deliberately no comparison with `now`, unlike
 405        // every other query in this store: an expired lease nobody has taken over must keep
 406        // reading as the same lease, because the engine's proof of a live holder is that two
 407        // observations DIFFER. Whether it has lapsed stays UpdateLeaseAsync's call. A no-tracking
 408        // named-record projection (see LoadAsync for why not an anonymous type), so state_json is
 409        // never selected and nothing is cached on the per-operation context.
 410        var row = await Records(lease.Context)
 411            .AsNoTracking()
 412            .Where(r => r.FlowId == flowId)
 413            .Select(r => new LeaseRow(r.LeaseId, r.LeaseExpiresAtUtc))
 414            .FirstOrDefaultAsync(cancellationToken)
 415            .ConfigureAwait(false);
 416
 417        // Providers disagree on the kind a zone-less column reads back with (SQLite and SQL Server:
 418        // Unspecified; Npgsql timestamptz: Utc). UpdateLeaseAsync wrote a UTC instant, and the
 419        // shared shaper stamps it so.
 420        return DurableFlowStoreShared.LeaseObservation(row?.LeaseId, row?.LeaseExpiresAtUtc);
 421    }
 422
 423    public async Task<bool> TryDeleteAsync(string flowId, CancellationToken cancellationToken = default)
 424    {
 425        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 426        await using var lease = await LeaseContextAsync(cancellationToken).ConfigureAwait(false);
 427
 428        var deleted = await Records(lease.Context)
 429            .Where(r => r.FlowId == flowId)
 430            .ExecuteDeleteAsync(cancellationToken)
 431            .ConfigureAwait(false);
 432        return deleted > 0;
 433    }
 434
 435    private static async Task<int> PruneExpiredAsync(TContext db, CancellationToken cancellationToken)
 436    {
 437        // One bounded batch per call; DurableFlowStoreShared.PruneQuietlyAsync repeats it under
 438        // the PruneBudget while batches come back full (policy shared by all relational stores):
 439        // an unbatched delete over a large expired backlog holds row locks and bloats one
 440        // transaction for the unlucky create that triggered the prune. Loads already filter on
 441        // expiry, so any backlog beyond the budget just waits for the next interval. The OrderBy
 442        // makes the row-limited delete deterministic (and keeps providers from warning about an
 443        // unordered Take).
 444        var now = DateTime.UtcNow;
 445        return await Records(db)
 446            .Where(r => r.ExpiresAtUtc <= now)
 447            .OrderBy(r => r.FlowId)
 448            .Take(DurableFlowStoreShared.PruneBatchSize)
 449            .ExecuteDeleteAsync(cancellationToken)
 450            .ConfigureAwait(false);
 451    }
 452
 453    private static DbSet<DurableFlowStateRecord> Records(TContext db) => db.Set<DurableFlowStateRecord>();
 454
 455    private async Task<bool> UpdateLeaseAsync(
 456        string flowId,
 457        string leaseId,
 458        TimeSpan leaseDuration,
 459        bool acquire,
 460        CancellationToken cancellationToken)
 461    {
 462        DurableFlowStoreShared.ValidateLeaseArgs(flowId, leaseId, leaseDuration);
 463
 464        await using var contextLease = await LeaseContextAsync(cancellationToken).ConfigureAwait(false);
 465        var now = DateTime.UtcNow;
 466        var leaseExpiresAtUtc = DurableFlowStoreShared.AddSaturating(now, leaseDuration);
 467        var query = Records(contextLease.Context).Where(r =>
 468            r.FlowId == flowId
 469            && r.ExpiresAtUtc > now
 470            && (acquire
 471                ? r.LeaseId == null || r.LeaseExpiresAtUtc <= now || r.LeaseId == leaseId
 472                : r.LeaseId == leaseId && r.LeaseExpiresAtUtc > now));
 473        var updated = await query.ExecuteUpdateAsync(setters => setters
 474                .SetProperty(r => r.LeaseId, leaseId)
 475                .SetProperty(r => r.LeaseExpiresAtUtc, leaseExpiresAtUtc), cancellationToken)
 476            .ConfigureAwait(false);
 477        return updated > 0;
 478    }
 479
 480    /// <summary>
 481    /// Leases a context for one operation: an <see cref="IDbContextFactory{TContext}"/>-created
 482    /// context when a factory is registered, otherwise the scoped <typeparamref name="TContext"/>
 483    /// owned by a fresh scope. Never caches a context — <see cref="DbContext"/> is not thread-safe
 484    /// and this store is a singleton used by parallel flow executions.
 485    /// </summary>
 486    private async ValueTask<ContextLease> LeaseContextAsync(CancellationToken cancellationToken)
 487    {
 488        var scope = _scopeFactory.CreateAsyncScope();
 489        try
 490        {
 491            var factory = scope.ServiceProvider.GetService<IDbContextFactory<TContext>>();
 492            var ownsContext = factory is not null;
 493            var context = ownsContext
 494                ? await factory!.CreateDbContextAsync(cancellationToken).ConfigureAwait(false)
 495                : scope.ServiceProvider.GetRequiredService<TContext>();
 496            try
 497            {
 498                EnsureMapped(context);
 499                return new ContextLease(context, scope, ownsContext);
 500            }
 501            catch
 502            {
 503                if (ownsContext)
 504                    await context.DisposeAsync().ConfigureAwait(false);
 505                throw;
 506            }
 507        }
 508        catch
 509        {
 510            await scope.DisposeAsync().ConfigureAwait(false);
 511            throw;
 512        }
 513    }
 514
 515    private void EnsureMapped(TContext context)
 516    {
 517        if (_modelChecked)
 518            return;
 519
 520        var entity = context.Model.FindEntityType(typeof(DurableFlowStateRecord))
 521            ?? throw new InvalidOperationException(
 522                $"'{typeof(TContext).Name}' does not map {nameof(DurableFlowStateRecord)}. Call " +
 523                $"modelBuilder.{nameof(EFCoreDurableFlowModelBuilderExtensions.ConfigureAsyncResponseDurableFlows)}() " 
 524                "in OnModelCreating and add a migration for the durable-flow state table.");
 525
 526        // The flow_id column is a KEY the engine compares ordinally. This package owns no DDL, so
 527        // it cannot pin the collation itself — but it can refuse to run against a mapping that
 528        // leaves it to a provider whose default folds case. On SQL Server and MySQL that default
 529        // makes 'flow-a' and 'FLOW-A' one primary key: the second StartAsync fails as a duplicate
 530        // and a load returns the other run's state. Silence there is not an acceptable default.
 531        // Read the decision from the annotation the mapping records, not from the property's
 532        // collation: EF Core strips relational configuration the runtime never reads out of
 533        // context.Model, and asking a runtime property for its collation throws outright.
 534        if (FlowIdCollationRules.CaseFoldingProvider(context.Database.ProviderName) is not { } provider)
 535        {
 536            _modelChecked = true;
 537            return;
 538        }
 539
 540        var collation = entity.FindAnnotation(EFCoreDurableFlowModelBuilderExtensions.FlowIdCollationAnnotation)?.Value 
 541        if (string.IsNullOrWhiteSpace(collation))
 542        {
 543            throw new InvalidOperationException(
 544                $"'{typeof(TContext).Name}' maps {nameof(DurableFlowStateRecord)}.{nameof(DurableFlowStateRecord.FlowId)
 545                $"collation, and {provider.Name} defaults to a case-insensitive one. Flow ids are compared ordinally, so
 546                "differing only in case would collide on the primary key — the second flow fails to start and a load ret
 547                $"other run's state. Pass {nameof(AsyncResponseFlowIdCollations)}.{provider.ConstantName} to " +
 548                $"{nameof(EFCoreDurableFlowModelBuilderExtensions.ConfigureAsyncResponseDurableFlows)}(flowIdCollation: 
 549                "migration.");
 550        }
 551
 552        // A declared collation is a claim, not a proof: "I chose one" and "I chose an ordinal one"
 553        // are different statements, and only the second is what the primary key needs. On these
 554        // providers the difference is namable, so name it — Latin1_General_100_CS_AS is a perfectly
 555        // valid SQL Server collation that still folds full-width forms, and every _CS_AI collation
 556        // folds accents. Only _BIN/_BIN2 (SQL Server) and _bin (MySQL) compare byte-wise.
 557        if (!provider.IsOrdinal(collation))
 558        {
 559            throw new InvalidOperationException(
 560                $"'{typeof(TContext).Name}' maps {nameof(DurableFlowStateRecord)}.{nameof(DurableFlowStateRecord.FlowId)
 561                $"collation '{collation}', which {provider.Name} does not compare byte-wise. Case sensitivity alone is n
 562                "case-sensitive collation still folds accents or full-width forms, so two flow ids the library treats as
 563                "collide on the primary key — the second flow fails to start and a load returns the other run's state. P
 564                $"{nameof(AsyncResponseFlowIdCollations)}.{provider.ConstantName} ('{provider.Recommended}') instead, or
 565                $"{provider.OrdinalDescription}, and add a migration.");
 566        }
 567
 568        _modelChecked = true;
 569    }
 570
 571    private readonly struct ContextLease : IAsyncDisposable
 572    {
 573        private readonly AsyncServiceScope _scope;
 574        private readonly bool _ownsContext;
 575
 576        public ContextLease(TContext context, AsyncServiceScope scope, bool ownsContext)
 577        {
 578            Context = context;
 579            _scope = scope;
 580            _ownsContext = ownsContext;
 581        }
 582
 583        public TContext Context { get; }
 584
 585        public async ValueTask DisposeAsync()
 586        {
 587            // Factory-created contexts are not owned by the scope; scoped ones are disposed with it.
 588            if (_ownsContext)
 589                await Context.DisposeAsync().ConfigureAwait(false);
 590            await _scope.DisposeAsync().ConfigureAwait(false);
 591        }
 592    }
 593
 594    /// <summary>
 595    /// Ledger-row projection for <see cref="LoadAsync"/>. A named type keeps the LINQ projection
 596    /// off the anonymous-type Expression.New overload that Native AOT trim analysis rejects.
 597    /// </summary>
 598    private sealed record StateRow(string StateJson, long Revision);
 599
 600    /// <summary>Lease-column projection for <see cref="ObserveLeaseAsync"/>; named for the same AOT reason as <see cref
 601    private sealed record LeaseRow(string? LeaseId, DateTime? LeaseExpiresAtUtc);
 602}
 603}
Generated by: ReportGenerator 5.5.11.0
09/22/2026 - 09:41:04
GitHub | reportgenerator.io

Methods/Properties

ConfigureAsyncResponseDurableFlows(Microsoft.EntityFrameworkCore.ModelBuilder,System.String,System.String,System.String)