| | | 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.EntityFrameworkCore.Infrastructure; |
| | | 7 | | using Microsoft.Extensions.DependencyInjection; |
| | | 8 | | using Microsoft.Extensions.DependencyInjection.Extensions; |
| | | 9 | | using Microsoft.Extensions.Logging; |
| | | 10 | | using Microsoft.Extensions.Options; |
| | | 11 | | |
| | | 12 | | namespace 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 | | |
| | | 43 | | namespace AsyncResponse.DurableFlows.EFCore |
| | | 44 | | { |
| | | 45 | | /// <summary>Options for the Entity Framework Core durable-flow state store.</summary> |
| | | 46 | | public 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> |
| | | 82 | | public sealed class DurableFlowStateRecord |
| | | 83 | | { |
| | | 84 | | /// <summary>The flow run id (primary key).</summary> |
| | 2630 | 85 | | public string FlowId { get; set; } = string.Empty; |
| | | 86 | | |
| | | 87 | | /// <summary>The serialized <see cref="FlowState"/> ledger.</summary> |
| | 2630 | 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> |
| | 1313 | 91 | | public DateTime ExpiresAtUtc { get; set; } |
| | | 92 | | |
| | | 93 | | /// <summary>UTC instant of the last save.</summary> |
| | 1313 | 94 | | public DateTime UpdatedAtUtc { get; set; } |
| | | 95 | | |
| | | 96 | | /// <summary>Optimistic-concurrency revision of the durable ledger.</summary> |
| | 1313 | 97 | | public long Revision { get; set; } |
| | | 98 | | |
| | | 99 | | /// <summary>Current execution-lease owner, when a worker is running the flow.</summary> |
| | 0 | 100 | | public string? LeaseId { get; set; } |
| | | 101 | | |
| | | 102 | | /// <summary>UTC expiry of the current execution lease.</summary> |
| | 4 | 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> |
| | | 111 | | internal 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> |
| | | 153 | | public 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> |
| | | 169 | | public 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 | | { |
| | | 205 | | ArgumentNullException.ThrowIfNull(modelBuilder); |
| | | 206 | | ArgumentException.ThrowIfNullOrWhiteSpace(tableName); |
| | | 207 | | |
| | | 208 | | modelBuilder.Entity<DurableFlowStateRecord>(entity => |
| | | 209 | | { |
| | | 210 | | entity.ToTable(tableName, schema); |
| | | 211 | | entity.HasKey(r => r.FlowId); |
| | | 212 | | // 400 matches the sibling packages' key column and stays inside every mainstream |
| | | 213 | | // provider's index-key size limit (SQL Server 900 bytes, MySQL 3072 bytes). |
| | | 214 | | entity.Property(r => r.FlowId).HasColumnName("flow_id").HasMaxLength(400); |
| | | 215 | | if (!string.IsNullOrWhiteSpace(flowIdCollation)) |
| | | 216 | | { |
| | | 217 | | entity.Property(r => r.FlowId).UseCollation(flowIdCollation); |
| | | 218 | | // Also recorded as a model annotation, which survives into the runtime model the |
| | | 219 | | // store can actually read at startup. |
| | | 220 | | entity.HasAnnotation(FlowIdCollationAnnotation, flowIdCollation); |
| | | 221 | | } |
| | | 222 | | entity.Property(r => r.StateJson).HasColumnName("state_json").IsRequired(); |
| | | 223 | | entity.Property(r => r.ExpiresAtUtc).HasColumnName("expires_at_utc"); |
| | | 224 | | entity.Property(r => r.UpdatedAtUtc).HasColumnName("updated_at_utc"); |
| | | 225 | | entity.Property(r => r.Revision).HasColumnName("revision").HasDefaultValue(0L); |
| | | 226 | | entity.Property(r => r.LeaseId).HasColumnName("lease_id").HasMaxLength(64); |
| | | 227 | | entity.Property(r => r.LeaseExpiresAtUtc).HasColumnName("lease_expires_at_utc"); |
| | | 228 | | entity.HasIndex(r => r.ExpiresAtUtc).HasDatabaseName($"{tableName}_expires_idx"); |
| | | 229 | | }); |
| | | 230 | | |
| | | 231 | | 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> |
| | | 240 | | public 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 | | } |