| | | 1 | | using System.Diagnostics.CodeAnalysis; |
| | | 2 | | using AsyncResponse; |
| | | 3 | | using AsyncResponse.DurableFlows.EFCore; |
| | | 4 | | using AsyncResponse.DurableFlows.Internal; |
| | | 5 | | using Microsoft.EntityFrameworkCore; |
| | | 6 | | using Microsoft.Extensions.DependencyInjection; |
| | | 7 | | using Microsoft.Extensions.DependencyInjection.Extensions; |
| | | 8 | | using Microsoft.Extensions.Options; |
| | | 9 | | |
| | | 10 | | namespace 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. |
| | | 35 | | builder.Services.TryAddSingleton<EFCoreFlowStateStore<TContext>>(); |
| | | 36 | | return builder.WithDurableFlows<EFCoreFlowStateStore<TContext>, EFCoreDurableFlowOptions>(configure); |
| | | 37 | | } |
| | | 38 | | } |
| | | 39 | | } |
| | | 40 | | |
| | | 41 | | namespace AsyncResponse.DurableFlows.EFCore |
| | | 42 | | { |
| | | 43 | | /// <summary>Options for the Entity Framework Core durable-flow state store.</summary> |
| | | 44 | | public 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> |
| | | 68 | | public 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> |
| | | 93 | | public 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> |
| | | 139 | | public 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 | | { |
| | 1 | 164 | | ArgumentException.ThrowIfNullOrWhiteSpace(flowId); |
| | 1 | 165 | | await using var lease = await LeaseContextAsync(cancellationToken).ConfigureAwait(false); |
| | | 166 | | |
| | 1 | 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. |
| | 1 | 171 | | var record = await Records(lease.Context) |
| | 1 | 172 | | .AsNoTracking() |
| | 1 | 173 | | .Where(r => r.FlowId == flowId && r.ExpiresAtUtc > now) |
| | 1 | 174 | | .Select(r => new StateRow(r.StateJson, r.Revision)) |
| | 1 | 175 | | .FirstOrDefaultAsync(cancellationToken) |
| | 1 | 176 | | .ConfigureAwait(false); |
| | | 177 | | |
| | 1 | 178 | | if (record is null) |
| | 1 | 179 | | return null; |
| | | 180 | | |
| | 1 | 181 | | return DurableFlowStoreShared.ReadState(flowId, record.StateJson, record.Revision); |
| | 1 | 182 | | } |
| | | 183 | | |
| | | 184 | | public async Task<bool> TryCreateAsync(string flowId, FlowState state, TimeSpan ttl, CancellationToken cancellationT |
| | | 185 | | { |
| | 1 | 186 | | DurableFlowStoreShared.ValidateCreate(flowId, state, ttl); |
| | 1 | 187 | | var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "EF Core"); |
| | 1 | 188 | | await using var lease = await LeaseContextAsync(cancellationToken).ConfigureAwait(false); |
| | 1 | 189 | | var db = lease.Context; |
| | 1 | 190 | | var now = DateTime.UtcNow; |
| | | 191 | | |
| | 1 | 192 | | if (DurableFlowStoreShared.ShouldPrune(ref _lastPruneTicks, _options.PruneInterval)) |
| | 1 | 193 | | await PruneExpiredAsync(db, cancellationToken).ConfigureAwait(false); |
| | | 194 | | |
| | 1 | 195 | | await Records(db) |
| | 1 | 196 | | .Where(r => r.FlowId == flowId && r.ExpiresAtUtc <= now) |
| | 1 | 197 | | .ExecuteDeleteAsync(cancellationToken) |
| | 1 | 198 | | .ConfigureAwait(false); |
| | 1 | 199 | | db.Add(new DurableFlowStateRecord |
| | 1 | 200 | | { |
| | 1 | 201 | | FlowId = flowId, |
| | 1 | 202 | | StateJson = stateJson, |
| | 1 | 203 | | ExpiresAtUtc = DurableFlowStoreShared.AddSaturating(now, ttl), |
| | 1 | 204 | | UpdatedAtUtc = now, |
| | 1 | 205 | | Revision = state.Revision |
| | 1 | 206 | | }); |
| | | 207 | | try |
| | | 208 | | { |
| | 1 | 209 | | await db.SaveChangesAsync(cancellationToken).ConfigureAwait(false); |
| | 1 | 210 | | return true; |
| | | 211 | | } |
| | 1 | 212 | | catch (DbUpdateException) |
| | | 213 | | { |
| | 1 | 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". |
| | 1 | 218 | | if (await Records(db) |
| | 1 | 219 | | .AsNoTracking() |
| | 1 | 220 | | .AnyAsync(r => r.FlowId == flowId, cancellationToken) |
| | 1 | 221 | | .ConfigureAwait(false)) |
| | 1 | 222 | | return false; |
| | | 223 | | |
| | 1 | 224 | | throw; |
| | | 225 | | } |
| | 1 | 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 | | { |
| | 1 | 236 | | DurableFlowStoreShared.ValidateUpdate(flowId, state, expectedRevision, ttl); |
| | 1 | 237 | | var stateJson = DurableFlowStoreShared.SerializeBounded(flowId, state, _options.MaxStateBytes, "EF Core"); |
| | 1 | 238 | | await using var lease = await LeaseContextAsync(cancellationToken).ConfigureAwait(false); |
| | 1 | 239 | | var now = DateTime.UtcNow; |
| | 1 | 240 | | var expiresAtUtc = DurableFlowStoreShared.AddSaturating(now, ttl); |
| | 1 | 241 | | var query = Records(lease.Context).Where(r => |
| | 1 | 242 | | r.FlowId == flowId |
| | 1 | 243 | | && r.Revision == expectedRevision |
| | 1 | 244 | | && r.ExpiresAtUtc > now |
| | 1 | 245 | | && (leaseId == null || (r.LeaseId == leaseId && r.LeaseExpiresAtUtc > now))); |
| | 1 | 246 | | var updated = await query.ExecuteUpdateAsync(setters => setters |
| | 1 | 247 | | .SetProperty(r => r.StateJson, stateJson) |
| | 1 | 248 | | .SetProperty(r => r.ExpiresAtUtc, expiresAtUtc) |
| | 1 | 249 | | .SetProperty(r => r.UpdatedAtUtc, now) |
| | 1 | 250 | | .SetProperty(r => r.Revision, state.Revision), cancellationToken) |
| | 1 | 251 | | .ConfigureAwait(false); |
| | 1 | 252 | | return updated > 0; |
| | 1 | 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 | | { |
| | 1 | 263 | | await using var lease = await LeaseContextAsync(cancellationToken).ConfigureAwait(false); |
| | 1 | 264 | | await Records(lease.Context) |
| | 1 | 265 | | .Where(r => r.FlowId == flowId && r.LeaseId == leaseId) |
| | 1 | 266 | | .ExecuteUpdateAsync(setters => setters |
| | 1 | 267 | | .SetProperty(r => r.LeaseId, (string?)null) |
| | 1 | 268 | | .SetProperty(r => r.LeaseExpiresAtUtc, (DateTime?)null), cancellationToken) |
| | 1 | 269 | | .ConfigureAwait(false); |
| | 1 | 270 | | } |
| | | 271 | | |
| | | 272 | | public async Task<bool> TryDeleteAsync(string flowId, CancellationToken cancellationToken = default) |
| | | 273 | | { |
| | 1 | 274 | | ArgumentException.ThrowIfNullOrWhiteSpace(flowId); |
| | 1 | 275 | | await using var lease = await LeaseContextAsync(cancellationToken).ConfigureAwait(false); |
| | | 276 | | |
| | 1 | 277 | | var deleted = await Records(lease.Context) |
| | 1 | 278 | | .Where(r => r.FlowId == flowId) |
| | 1 | 279 | | .ExecuteDeleteAsync(cancellationToken) |
| | 1 | 280 | | .ConfigureAwait(false); |
| | 1 | 281 | | return deleted > 0; |
| | 1 | 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). |
| | 1 | 292 | | var now = DateTime.UtcNow; |
| | 1 | 293 | | await Records(db) |
| | 1 | 294 | | .Where(r => r.ExpiresAtUtc <= now) |
| | 1 | 295 | | .OrderBy(r => r.FlowId) |
| | 1 | 296 | | .Take(PruneBatchSize) |
| | 1 | 297 | | .ExecuteDeleteAsync(cancellationToken) |
| | 1 | 298 | | .ConfigureAwait(false); |
| | 1 | 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 | | { |
| | 1 | 310 | | ArgumentException.ThrowIfNullOrWhiteSpace(flowId); |
| | 1 | 311 | | ArgumentException.ThrowIfNullOrWhiteSpace(leaseId); |
| | 1 | 312 | | if (leaseDuration <= TimeSpan.Zero) |
| | 1 | 313 | | throw new ArgumentOutOfRangeException(nameof(leaseDuration)); |
| | | 314 | | |
| | 1 | 315 | | await using var contextLease = await LeaseContextAsync(cancellationToken).ConfigureAwait(false); |
| | 1 | 316 | | var now = DateTime.UtcNow; |
| | 1 | 317 | | var leaseExpiresAtUtc = DurableFlowStoreShared.AddSaturating(now, leaseDuration); |
| | 1 | 318 | | var query = Records(contextLease.Context).Where(r => |
| | 1 | 319 | | r.FlowId == flowId |
| | 1 | 320 | | && r.ExpiresAtUtc > now |
| | 1 | 321 | | && (acquire |
| | 1 | 322 | | ? r.LeaseId == null || r.LeaseExpiresAtUtc <= now || r.LeaseId == leaseId |
| | 1 | 323 | | : r.LeaseId == leaseId && r.LeaseExpiresAtUtc > now)); |
| | 1 | 324 | | var updated = await query.ExecuteUpdateAsync(setters => setters |
| | 1 | 325 | | .SetProperty(r => r.LeaseId, leaseId) |
| | 1 | 326 | | .SetProperty(r => r.LeaseExpiresAtUtc, leaseExpiresAtUtc), cancellationToken) |
| | 1 | 327 | | .ConfigureAwait(false); |
| | 1 | 328 | | return updated > 0; |
| | 1 | 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 | | { |
| | 1 | 339 | | var scope = _scopeFactory.CreateAsyncScope(); |
| | | 340 | | try |
| | | 341 | | { |
| | 1 | 342 | | var factory = scope.ServiceProvider.GetService<IDbContextFactory<TContext>>(); |
| | 1 | 343 | | var ownsContext = factory is not null; |
| | 1 | 344 | | var context = ownsContext |
| | 1 | 345 | | ? await factory!.CreateDbContextAsync(cancellationToken).ConfigureAwait(false) |
| | 1 | 346 | | : scope.ServiceProvider.GetRequiredService<TContext>(); |
| | | 347 | | try |
| | | 348 | | { |
| | 1 | 349 | | EnsureMapped(context); |
| | 1 | 350 | | return new ContextLease(context, scope, ownsContext); |
| | | 351 | | } |
| | 1 | 352 | | catch |
| | | 353 | | { |
| | 1 | 354 | | if (ownsContext) |
| | 1 | 355 | | await context.DisposeAsync().ConfigureAwait(false); |
| | 1 | 356 | | throw; |
| | | 357 | | } |
| | 0 | 358 | | } |
| | 1 | 359 | | catch |
| | | 360 | | { |
| | 1 | 361 | | await scope.DisposeAsync().ConfigureAwait(false); |
| | 1 | 362 | | throw; |
| | | 363 | | } |
| | 1 | 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 | | { |
| | 1 | 387 | | Context = context; |
| | 1 | 388 | | _scope = scope; |
| | 1 | 389 | | _ownsContext = ownsContext; |
| | 1 | 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. |
| | 1 | 397 | | if (_ownsContext) |
| | 1 | 398 | | await Context.DisposeAsync().ConfigureAwait(false); |
| | 1 | 399 | | await _scope.DisposeAsync().ConfigureAwait(false); |
| | 1 | 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> |
| | 1 | 407 | | private sealed record StateRow(string StateJson, long Revision); |
| | | 408 | | } |
| | | 409 | | } |