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

Information
Class: Microsoft.Extensions.DependencyInjection.EFCoreDurableFlowServiceCollectionExtensions
Assembly: AsyncResponse.DurableFlows.EFCore
File(s): /home/runner/work/AsyncResponse/AsyncResponse/src/DurableFlows/AsyncResponse.DurableFlows.EFCore/EFCoreDurableFlows.cs
Line coverage
100%
Covered lines: 2
Uncovered lines: 0
Coverable lines: 2
Total lines: 409
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
WithEFCoreDurableFlows<TContext>(...)100%11100%

File(s)

/home/runner/work/AsyncResponse/AsyncResponse/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.Extensions.DependencyInjection;
 7using Microsoft.Extensions.DependencyInjection.Extensions;
 8using Microsoft.Extensions.Options;
 9
 10namespace Microsoft.Extensions.DependencyInjection
 11{
 12    /// <summary>DI registration for the Entity Framework Core durable-flow state store.</summary>
 13    public static class EFCoreDurableFlowServiceCollectionExtensions
 14    {
 15        /// <summary>
 16        /// Stores durable-flow state in a table hosted by the application's own
 17        /// <typeparamref name="TContext"/>. Map the table into the context's model with
 18        /// <see cref="EFCoreDurableFlowModelBuilderExtensions.ConfigureAsyncResponseDurableFlows"/>
 19        /// so it rides the application's migration pipeline; the store itself never runs DDL.
 20        /// <para>
 21        /// Each operation resolves a fresh context: from <see cref="IDbContextFactory{TContext}"/>
 22        /// when one is registered (<c>AddDbContextFactory</c>), otherwise the scoped
 23        /// <typeparamref name="TContext"/> from a new service scope (<c>AddDbContext</c>). Parallel
 24        /// flow executions therefore never share a <see cref="DbContext"/> instance.
 25        /// </para>
 26        /// </summary>
 27        public static AsyncResponseRegistrationBuilder WithEFCoreDurableFlows<[DynamicallyAccessedMembers(DynamicallyAcc
 28            this AsyncResponseRegistrationBuilder builder,
 29            Action<EFCoreDurableFlowOptions>? configure = null)
 30            where TContext : DbContext
 31        {
 32            // Singleton on purpose: the store holds no DbContext (each operation leases one, see
 33            // above), and the executor resolves the store from a fresh scope per flow execution —
 34            // a scoped store would redo the mapped-model check on every run.
 235            builder.Services.TryAddSingleton<EFCoreFlowStateStore<TContext>>();
 236            return builder.WithDurableFlows<EFCoreFlowStateStore<TContext>, EFCoreDurableFlowOptions>(configure);
 37        }
 38    }
 39}
 40
 41namespace AsyncResponse.DurableFlows.EFCore
 42{
 43/// <summary>Options for the Entity Framework Core durable-flow state store.</summary>
 44public sealed class EFCoreDurableFlowOptions : DurableFlowOptions
 45{
 46    /// <summary>
 47    /// How often <see cref="EFCoreFlowStateStore{TContext}.TryCreateAsync"/> opportunistically deletes
 48    /// one bounded batch (1000 rows) of expired rows (loads already treat expired state as absent;
 49    /// pruning bounds table growth). Zero or negative prunes on every save. Default: 5 minutes.
 50    /// </summary>
 51    public TimeSpan PruneInterval { get; set; } = TimeSpan.FromMinutes(5);
 52
 53    /// <summary>
 54    /// Maximum serialized flow-state size in bytes accepted by writes; oversized ledgers fail fast
 55    /// with an actionable error instead of an opaque provider error. Default: <c>null</c>
 56    /// (unlimited — relational text/blob columns are effectively unbounded), settable as an
 57    /// operator budget.
 58    /// </summary>
 59    public long? MaxStateBytes { get; set; }
 60}
 61
 62/// <summary>
 63/// One durable-flow ledger row, mapped into the application's <see cref="DbContext"/> by
 64/// <see cref="EFCoreDurableFlowModelBuilderExtensions.ConfigureAsyncResponseDurableFlows"/>.
 65/// Column names match the other AsyncResponse.DurableFlows.* relational packages, so the table is
 66/// interchangeable with theirs.
 67/// </summary>
 68public sealed class DurableFlowStateRecord
 69{
 70    /// <summary>The flow run id (primary key).</summary>
 71    public string FlowId { get; set; } = string.Empty;
 72
 73    /// <summary>The serialized <see cref="FlowState"/> ledger.</summary>
 74    public string StateJson { get; set; } = string.Empty;
 75
 76    /// <summary>UTC instant after which the row is treated as absent and eligible for pruning.</summary>
 77    public DateTime ExpiresAtUtc { get; set; }
 78
 79    /// <summary>UTC instant of the last save.</summary>
 80    public DateTime UpdatedAtUtc { get; set; }
 81
 82    /// <summary>Optimistic-concurrency revision of the durable ledger.</summary>
 83    public long Revision { get; set; }
 84
 85    /// <summary>Current execution-lease owner, when a worker is running the flow.</summary>
 86    public string? LeaseId { get; set; }
 87
 88    /// <summary>UTC expiry of the current execution lease.</summary>
 89    public DateTime? LeaseExpiresAtUtc { get; set; }
 90}
 91
 92/// <summary>Maps the durable-flow state table into an application model.</summary>
 93public static class EFCoreDurableFlowModelBuilderExtensions
 94{
 95    /// <summary>Default durable-flow state table name (shared with the other relational packages).</summary>
 96    public const string DefaultTableName = "asyncresponse_flow_state";
 97
 98    /// <summary>
 99    /// Maps <see cref="DurableFlowStateRecord"/> to the durable-flow state table. Call from
 100    /// <c>OnModelCreating</c>; the table then flows through the application's normal EF Core
 101    /// migrations (or <c>EnsureCreated</c>) like any other entity.
 102    /// </summary>
 103    /// <param name="modelBuilder">The application model builder.</param>
 104    /// <param name="tableName">Table name. Default: <see cref="DefaultTableName"/>.</param>
 105    /// <param name="schema">Optional schema; <c>null</c> uses the provider default.</param>
 106    public static ModelBuilder ConfigureAsyncResponseDurableFlows(
 107        this ModelBuilder modelBuilder,
 108        string tableName = DefaultTableName,
 109        string? schema = null)
 110    {
 111        ArgumentNullException.ThrowIfNull(modelBuilder);
 112        ArgumentException.ThrowIfNullOrWhiteSpace(tableName);
 113
 114        modelBuilder.Entity<DurableFlowStateRecord>(entity =>
 115        {
 116            entity.ToTable(tableName, schema);
 117            entity.HasKey(r => r.FlowId);
 118            // 400 matches the sibling packages' key column and stays inside every mainstream
 119            // provider's index-key size limit (SQL Server 900 bytes, MySQL 3072 bytes).
 120            entity.Property(r => r.FlowId).HasColumnName("flow_id").HasMaxLength(400);
 121            entity.Property(r => r.StateJson).HasColumnName("state_json").IsRequired();
 122            entity.Property(r => r.ExpiresAtUtc).HasColumnName("expires_at_utc");
 123            entity.Property(r => r.UpdatedAtUtc).HasColumnName("updated_at_utc");
 124            entity.Property(r => r.Revision).HasColumnName("revision").HasDefaultValue(0L);
 125            entity.Property(r => r.LeaseId).HasColumnName("lease_id").HasMaxLength(64);
 126            entity.Property(r => r.LeaseExpiresAtUtc).HasColumnName("lease_expires_at_utc");
 127            entity.HasIndex(r => r.ExpiresAtUtc).HasDatabaseName($"{tableName}_expires_idx");
 128        });
 129
 130        return modelBuilder;
 131    }
 132}
 133
 134/// <summary>
 135/// Entity Framework Core implementation of <see cref="IFlowStateStore"/> over an
 136/// application-owned <typeparamref name="TContext"/>. Requires a relational provider
 137/// (deletes and updates use <c>ExecuteDeleteAsync</c>/<c>ExecuteUpdateAsync</c>).
 138/// </summary>
 139public sealed class EFCoreFlowStateStore<[DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.PublicConstructors |
 140    where TContext : DbContext
 141{
 142    private const int PruneBatchSize = 1000;
 143
 144    // Time authority: this store deliberately keeps the app clock (DateTime.UtcNow) for expiry
 145    // and lease comparisons. It is provider-agnostic LINQ — there is no portable way to reference
 146    // the database server's clock in a translated expression — so multi-node deployments should
 147    // either keep worker clocks synchronized well inside the lease window or use one of the
 148    // provider-specific relational stores, which run all time math on the database clock.
 149    private readonly IServiceScopeFactory _scopeFactory;
 150    private readonly EFCoreDurableFlowOptions _options;
 151    private long _lastPruneTicks;
 152    private volatile bool _modelChecked;
 153
 154    public EFCoreFlowStateStore(IServiceScopeFactory scopeFactory, IOptions<EFCoreDurableFlowOptions> options)
 155    {
 156        _scopeFactory = scopeFactory;
 157        _options = options.Value;
 158        if (_options.MaxStateBytes is <= 0)
 159            throw new InvalidOperationException($"{nameof(EFCoreDurableFlowOptions)}.{nameof(EFCoreDurableFlowOptions.Ma
 160    }
 161
 162    public async Task<FlowState?> LoadAsync(string flowId, CancellationToken cancellationToken = default)
 163    {
 164        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 165        await using var lease = await LeaseContextAsync(cancellationToken).ConfigureAwait(false);
 166
 167        var now = DateTime.UtcNow;
 168        // Named-record projection, not an anonymous type: anonymous projections lower to the
 169        // RequiresUnreferencedCode Expression.New(ctor, args, members) overload, which ILC trim
 170        // analysis rejects in Native AOT publishes even though the Roslyn analyzer stays quiet.
 171        var record = await Records(lease.Context)
 172            .AsNoTracking()
 173            .Where(r => r.FlowId == flowId && r.ExpiresAtUtc > now)
 174            .Select(r => new StateRow(r.StateJson, r.Revision))
 175            .FirstOrDefaultAsync(cancellationToken)
 176            .ConfigureAwait(false);
 177
 178        if (record is null)
 179            return null;
 180
 181        return DurableFlowStoreShared.ReadState(flowId, record.StateJson, record.Revision);
 182    }
 183
 184    public async Task<bool> TryCreateAsync(string flowId, FlowState state, TimeSpan ttl, CancellationToken cancellationT
 185    {
 186        DurableFlowStoreShared.ValidateCreate(flowId, state, ttl);
 187        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "EF Core");
 188        await using var lease = await LeaseContextAsync(cancellationToken).ConfigureAwait(false);
 189        var db = lease.Context;
 190        var now = DateTime.UtcNow;
 191
 192        if (DurableFlowStoreShared.ShouldPrune(ref _lastPruneTicks, _options.PruneInterval))
 193            await PruneExpiredAsync(db, cancellationToken).ConfigureAwait(false);
 194
 195        await Records(db)
 196            .Where(r => r.FlowId == flowId && r.ExpiresAtUtc <= now)
 197            .ExecuteDeleteAsync(cancellationToken)
 198            .ConfigureAwait(false);
 199        db.Add(new DurableFlowStateRecord
 200        {
 201            FlowId = flowId,
 202            StateJson = stateJson,
 203            ExpiresAtUtc = DurableFlowStoreShared.AddSaturating(now, ttl),
 204            UpdatedAtUtc = now,
 205            Revision = state.Revision
 206        });
 207        try
 208        {
 209            await db.SaveChangesAsync(cancellationToken).ConfigureAwait(false);
 210            return true;
 211        }
 212        catch (DbUpdateException)
 213        {
 214            db.ChangeTracker.Clear();
 215            // Provider-agnostic duplicate-key detection is not reliable. Verify that another
 216            // creator actually owns this id; otherwise preserve the real database failure instead
 217            // of misreporting truncation, trigger, permission, or schema errors as "already exists".
 218            if (await Records(db)
 219                    .AsNoTracking()
 220                    .AnyAsync(r => r.FlowId == flowId, cancellationToken)
 221                    .ConfigureAwait(false))
 222                return false;
 223
 224            throw;
 225        }
 226    }
 227
 228    public async Task<bool> TryUpdateAsync(
 229        string flowId,
 230        FlowState state,
 231        long expectedRevision,
 232        TimeSpan ttl,
 233        string? leaseId = null,
 234        CancellationToken cancellationToken = default)
 235    {
 236        DurableFlowStoreShared.ValidateUpdate(flowId, state, expectedRevision, ttl);
 237        var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "EF Core");
 238        await using var lease = await LeaseContextAsync(cancellationToken).ConfigureAwait(false);
 239        var now = DateTime.UtcNow;
 240        var expiresAtUtc = DurableFlowStoreShared.AddSaturating(now, ttl);
 241        var query = Records(lease.Context).Where(r =>
 242            r.FlowId == flowId
 243            && r.Revision == expectedRevision
 244            && r.ExpiresAtUtc > now
 245            && (leaseId == null || (r.LeaseId == leaseId && r.LeaseExpiresAtUtc > now)));
 246        var updated = await query.ExecuteUpdateAsync(setters => setters
 247                .SetProperty(r => r.StateJson, stateJson)
 248                .SetProperty(r => r.ExpiresAtUtc, expiresAtUtc)
 249                .SetProperty(r => r.UpdatedAtUtc, now)
 250                .SetProperty(r => r.Revision, state.Revision), cancellationToken)
 251            .ConfigureAwait(false);
 252        return updated > 0;
 253    }
 254
 255    public Task<bool> TryAcquireLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken canc
 256        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: true, cancellationToken);
 257
 258    public Task<bool> TryRenewLeaseAsync(string flowId, string leaseId, TimeSpan leaseDuration, CancellationToken cancel
 259        => UpdateLeaseAsync(flowId, leaseId, leaseDuration, acquire: false, cancellationToken);
 260
 261    public async Task ReleaseLeaseAsync(string flowId, string leaseId, CancellationToken cancellationToken = default)
 262    {
 263        await using var lease = await LeaseContextAsync(cancellationToken).ConfigureAwait(false);
 264        await Records(lease.Context)
 265            .Where(r => r.FlowId == flowId && r.LeaseId == leaseId)
 266            .ExecuteUpdateAsync(setters => setters
 267                .SetProperty(r => r.LeaseId, (string?)null)
 268                .SetProperty(r => r.LeaseExpiresAtUtc, (DateTime?)null), cancellationToken)
 269            .ConfigureAwait(false);
 270    }
 271
 272    public async Task<bool> TryDeleteAsync(string flowId, CancellationToken cancellationToken = default)
 273    {
 274        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 275        await using var lease = await LeaseContextAsync(cancellationToken).ConfigureAwait(false);
 276
 277        var deleted = await Records(lease.Context)
 278            .Where(r => r.FlowId == flowId)
 279            .ExecuteDeleteAsync(cancellationToken)
 280            .ConfigureAwait(false);
 281        return deleted > 0;
 282    }
 283
 284    private static async Task PruneExpiredAsync(TContext db, CancellationToken cancellationToken)
 285    {
 286        // One bounded batch per prune interval (policy shared by all relational stores): an
 287        // unbatched delete over a large expired backlog holds row locks and bloats one
 288        // transaction for the unlucky create that triggered the prune. Loads already filter on
 289        // expiry, so any backlog beyond the batch just waits for the next interval. The OrderBy
 290        // makes the row-limited delete deterministic (and keeps providers from warning about an
 291        // unordered Take).
 292        var now = DateTime.UtcNow;
 293        await Records(db)
 294            .Where(r => r.ExpiresAtUtc <= now)
 295            .OrderBy(r => r.FlowId)
 296            .Take(PruneBatchSize)
 297            .ExecuteDeleteAsync(cancellationToken)
 298            .ConfigureAwait(false);
 299    }
 300
 301    private static DbSet<DurableFlowStateRecord> Records(TContext db) => db.Set<DurableFlowStateRecord>();
 302
 303    private async Task<bool> UpdateLeaseAsync(
 304        string flowId,
 305        string leaseId,
 306        TimeSpan leaseDuration,
 307        bool acquire,
 308        CancellationToken cancellationToken)
 309    {
 310        ArgumentException.ThrowIfNullOrWhiteSpace(flowId);
 311        ArgumentException.ThrowIfNullOrWhiteSpace(leaseId);
 312        if (leaseDuration <= TimeSpan.Zero)
 313            throw new ArgumentOutOfRangeException(nameof(leaseDuration));
 314
 315        await using var contextLease = await LeaseContextAsync(cancellationToken).ConfigureAwait(false);
 316        var now = DateTime.UtcNow;
 317        var leaseExpiresAtUtc = DurableFlowStoreShared.AddSaturating(now, leaseDuration);
 318        var query = Records(contextLease.Context).Where(r =>
 319            r.FlowId == flowId
 320            && r.ExpiresAtUtc > now
 321            && (acquire
 322                ? r.LeaseId == null || r.LeaseExpiresAtUtc <= now || r.LeaseId == leaseId
 323                : r.LeaseId == leaseId && r.LeaseExpiresAtUtc > now));
 324        var updated = await query.ExecuteUpdateAsync(setters => setters
 325                .SetProperty(r => r.LeaseId, leaseId)
 326                .SetProperty(r => r.LeaseExpiresAtUtc, leaseExpiresAtUtc), cancellationToken)
 327            .ConfigureAwait(false);
 328        return updated > 0;
 329    }
 330
 331    /// <summary>
 332    /// Leases a context for one operation: an <see cref="IDbContextFactory{TContext}"/>-created
 333    /// context when a factory is registered, otherwise the scoped <typeparamref name="TContext"/>
 334    /// owned by a fresh scope. Never caches a context — <see cref="DbContext"/> is not thread-safe
 335    /// and this store is a singleton used by parallel flow executions.
 336    /// </summary>
 337    private async ValueTask<ContextLease> LeaseContextAsync(CancellationToken cancellationToken)
 338    {
 339        var scope = _scopeFactory.CreateAsyncScope();
 340        try
 341        {
 342            var factory = scope.ServiceProvider.GetService<IDbContextFactory<TContext>>();
 343            var ownsContext = factory is not null;
 344            var context = ownsContext
 345                ? await factory!.CreateDbContextAsync(cancellationToken).ConfigureAwait(false)
 346                : scope.ServiceProvider.GetRequiredService<TContext>();
 347            try
 348            {
 349                EnsureMapped(context);
 350                return new ContextLease(context, scope, ownsContext);
 351            }
 352            catch
 353            {
 354                if (ownsContext)
 355                    await context.DisposeAsync().ConfigureAwait(false);
 356                throw;
 357            }
 358        }
 359        catch
 360        {
 361            await scope.DisposeAsync().ConfigureAwait(false);
 362            throw;
 363        }
 364    }
 365
 366    private void EnsureMapped(TContext context)
 367    {
 368        if (_modelChecked)
 369            return;
 370
 371        if (context.Model.FindEntityType(typeof(DurableFlowStateRecord)) is null)
 372            throw new InvalidOperationException(
 373                $"'{typeof(TContext).Name}' does not map {nameof(DurableFlowStateRecord)}. Call " +
 374                $"modelBuilder.{nameof(EFCoreDurableFlowModelBuilderExtensions.ConfigureAsyncResponseDurableFlows)}() " 
 375                "in OnModelCreating and add a migration for the durable-flow state table.");
 376
 377        _modelChecked = true;
 378    }
 379
 380    private readonly struct ContextLease : IAsyncDisposable
 381    {
 382        private readonly AsyncServiceScope _scope;
 383        private readonly bool _ownsContext;
 384
 385        public ContextLease(TContext context, AsyncServiceScope scope, bool ownsContext)
 386        {
 387            Context = context;
 388            _scope = scope;
 389            _ownsContext = ownsContext;
 390        }
 391
 392        public TContext Context { get; }
 393
 394        public async ValueTask DisposeAsync()
 395        {
 396            // Factory-created contexts are not owned by the scope; scoped ones are disposed with it.
 397            if (_ownsContext)
 398                await Context.DisposeAsync().ConfigureAwait(false);
 399            await _scope.DisposeAsync().ConfigureAwait(false);
 400        }
 401    }
 402
 403    /// <summary>
 404    /// Ledger-row projection for <see cref="LoadAsync"/>. A named type keeps the LINQ projection
 405    /// off the anonymous-type Expression.New overload that Native AOT trim analysis rejects.
 406    /// </summary>
 407    private sealed record StateRow(string StateJson, long Revision);
 408}
 409}
Generated by: ReportGenerator 5.5.10.0
08/03/2026 - 08:59:44
GitHub | reportgenerator.io

Methods/Properties

WithEFCoreDurableFlows<TContext>(Microsoft.Extensions.DependencyInjection.AsyncResponseRegistrationBuilder, System.Action<AsyncResponse.DurableFlows.EFCore.EFCoreDurableFlowOptions>)