| | | 1 | | using Microsoft.Extensions.Logging; |
| | | 2 | | |
| | | 3 | | namespace AsyncResponse; |
| | | 4 | | |
| | | 5 | | /// <summary>Coordinates atomic flow creation, optimistic updates, and one active executor per flow id.</summary> |
| | | 6 | | internal static class FlowStateConcurrency |
| | | 7 | | { |
| | | 8 | | private const int MaxUpdateAttempts = 8; |
| | | 9 | | |
| | | 10 | | public static Task<bool> TryCreateAsync( |
| | | 11 | | IFlowStateStore store, |
| | | 12 | | string flowId, |
| | | 13 | | FlowState state, |
| | | 14 | | TimeSpan ttl, |
| | | 15 | | CancellationToken cancellationToken = default) |
| | | 16 | | { |
| | 3219 | 17 | | EnsurePortableFlowId(flowId); |
| | | 18 | | |
| | 3217 | 19 | | state.Revision = 0; |
| | 3217 | 20 | | return store.TryCreateAsync(flowId, state, ttl, cancellationToken); |
| | | 21 | | } |
| | | 22 | | |
| | | 23 | | /// <summary> |
| | | 24 | | /// Throws <see cref="ArgumentException"/> for an id that fails <see cref="FlowIdNotPortable"/>. |
| | | 25 | | /// Called by every create, and by <c>IDurableFlows.StartAsync</c> BEFORE it publishes the start |
| | | 26 | | /// job — the publish is the start's commit point, so a job for an id no store would accept must |
| | | 27 | | /// never leave the process. |
| | | 28 | | /// </summary> |
| | | 29 | | internal static void EnsurePortableFlowId(string flowId) |
| | | 30 | | { |
| | 4823 | 31 | | if (FlowIdNotPortable(flowId) is { } rejection) |
| | 24 | 32 | | throw new ArgumentException(rejection, nameof(flowId)); |
| | 4799 | 33 | | } |
| | | 34 | | |
| | | 35 | | /// <summary> |
| | | 36 | | /// Whether an existing ledger describes the same start as the requested one: same flow type, |
| | | 37 | | /// same input type (both ordinal), and semantically identical input JSON. The one idempotency |
| | | 38 | | /// test for flow ids, shared by the starter (which reports a mismatch to its caller as |
| | | 39 | | /// <see cref="DurableFlowIdConflictException"/>) and the executor's start target (which drops |
| | | 40 | | /// the job on a mismatch) so the two can never disagree about what "the same run" means. |
| | | 41 | | /// </summary> |
| | | 42 | | internal static bool IsSameStart(FlowState existing, string? flowTypeName, string? inputTypeName, string? inputJson) |
| | 1552 | 43 | | => string.Equals(existing.FlowTypeName, flowTypeName, StringComparison.Ordinal) |
| | 1552 | 44 | | && string.Equals(existing.InputTypeName, inputTypeName, StringComparison.Ordinal) |
| | 1552 | 45 | | && FlowStateJson.JsonEquivalent(existing.InputJson, inputJson ?? string.Empty); |
| | | 46 | | |
| | | 47 | | /// <summary> |
| | | 48 | | /// Enforces the portable flow-id contract on every final id at creation — the single door all |
| | | 49 | | /// creates walk through. Three independent limits, because the stores disagree about what an |
| | | 50 | | /// id may be, and an id that works on one store and fails on another is not portable: |
| | | 51 | | /// <list type="bullet"> |
| | | 52 | | /// <item>length in UTF-16 code units, for the 400-unit <c>flow_id</c> columns (SQL Server, |
| | | 53 | | /// MySQL, Oracle, EF Core);</item> |
| | | 54 | | /// <item>length in UTF-8 <em>bytes</em>, for Cosmos DB, whose 1023-byte id limit a 400-unit id |
| | | 55 | | /// exceeds once the characters are non-ASCII (up to three bytes per unit, four for a |
| | | 56 | | /// surrogate pair);</item> |
| | | 57 | | /// <item>the characters themselves — Cosmos rejects <c>/</c>, <c>\</c>, <c>?</c> and <c>#</c> |
| | | 58 | | /// in an id, and control characters break every store's diagnostics;</item> |
| | | 59 | | /// <item>no surrounding spaces — SQL Server pads the shorter operand of an equality |
| | | 60 | | /// comparison (binary collations included) and MySQL's <c>utf8mb4_bin</c> is PAD SPACE, so |
| | | 61 | | /// <c>flow</c> and <c>flow </c> are ONE key to those databases while the engine treats |
| | | 62 | | /// them as two runs.</item> |
| | | 63 | | /// </list> |
| | | 64 | | /// Case is deliberately NOT folded here: ids are compared ordinally throughout, and the |
| | | 65 | | /// relational stores pin a binary collation on the column so the database agrees. |
| | | 66 | | /// Returns the rejection message, or <c>null</c> when the id is portable. |
| | | 67 | | /// </summary> |
| | | 68 | | internal static string? FlowIdNotPortable(string flowId) |
| | | 69 | | { |
| | | 70 | | // Length-guarded before the [0]/[^1] probe below: this is the single door every create |
| | | 71 | | // walks through and its contract is to RETURN a rejection, so an empty id must not throw |
| | | 72 | | // IndexOutOfRangeException out of the very method whose job is to explain bad ids. |
| | 5157 | 73 | | if (flowId.Length == 0) |
| | 4 | 74 | | return "Flow id is empty. A run needs an id to be addressable by its wake-ups, child flows, and recovery cal |
| | | 75 | | |
| | 5153 | 76 | | if (flowId.Length > DurableFlowOptions.MaxFlowIdLength) |
| | | 77 | | { |
| | 8 | 78 | | return $"Flow id '{Excerpt(flowId)}' is {flowId.Length} UTF-16 code units; the portable maximum is " + |
| | 8 | 79 | | $"{DurableFlowOptions.MaxFlowIdLength} ({nameof(DurableFlowOptions)}.{nameof(DurableFlowOptions.MaxFlowI |
| | 8 | 80 | | "column length in the SQL Server, MySQL, Oracle, and EF Core stores). " + BudgetGuidance; |
| | | 81 | | } |
| | | 82 | | |
| | | 83 | | // Checked BEFORE the byte count, which would otherwise be measured against the U+FFFD an |
| | | 84 | | // encoder substitutes rather than against the id the caller passed. |
| | 5145 | 85 | | if (PortableText.IndexOfIllFormedUtf16(flowId) is var illFormed and >= 0) |
| | 6 | 86 | | return PortableText.IllFormedUtf16Rejection("Flow id", Excerpt(flowId), flowId[illFormed], illFormed); |
| | | 87 | | |
| | 5139 | 88 | | var utf8Bytes = System.Text.Encoding.UTF8.GetByteCount(flowId); |
| | 5139 | 89 | | if (utf8Bytes > DurableFlowOptions.MaxFlowIdBytes) |
| | | 90 | | { |
| | 2 | 91 | | return $"Flow id '{Excerpt(flowId)}' is {utf8Bytes} UTF-8 bytes; the portable maximum is " + |
| | 2 | 92 | | $"{DurableFlowOptions.MaxFlowIdBytes} ({nameof(DurableFlowOptions)}.{nameof(DurableFlowOptions.MaxFlowId |
| | 2 | 93 | | "id limit). A non-ASCII character costs up to three bytes (four for a surrogate pair), so a count of cha |
| | 2 | 94 | | "does not bound the byte length. " + BudgetGuidance; |
| | | 95 | | } |
| | | 96 | | |
| | 5137 | 97 | | if (flowId[0] == ' ' || flowId[^1] == ' ') |
| | | 98 | | { |
| | 4 | 99 | | return $"Flow id '{Excerpt(flowId)}' begins or ends with a space. SQL Server pads the shorter operand of an |
| | 4 | 100 | | "comparison — binary collations included — and MySQL's utf8mb4_bin is PAD SPACE, so an id with trailing |
| | 4 | 101 | | "the SAME key as one without to those stores, while the engine compares them ordinally and treats them a |
| | 4 | 102 | | "different flows. Trim the id."; |
| | | 103 | | } |
| | | 104 | | |
| | 409030 | 105 | | foreach (var character in flowId) |
| | | 106 | | { |
| | 199391 | 107 | | if (character is '/' or '\\' or '?' or '#' || char.IsControl(character)) |
| | | 108 | | { |
| | 18 | 109 | | return $"Flow id '{Excerpt(flowId)}' contains the character '{(char.IsControl(character) ? $"\\u{(int)ch |
| | 18 | 110 | | "which is not portable: Cosmos DB rejects '/', '\\', '?' and '#' in an id, and control characters co |
| | 18 | 111 | | "diagnostics. Use a separator the stores agree on, such as ':' or '-'."; |
| | | 112 | | } |
| | | 113 | | } |
| | | 114 | | |
| | 5115 | 115 | | return null; |
| | | 116 | | } |
| | | 117 | | |
| | | 118 | | private const string BudgetGuidance = |
| | | 119 | | "Budget root ids for growth: child flows append \":{stepName}\" to the parent id, and scheduled flows wrap the s |
| | | 120 | | "name as \"sched:{name}:{timestamp}\"."; |
| | | 121 | | |
| | 38 | 122 | | private static string Excerpt(string flowId) => PortableText.Excerpt(flowId); |
| | | 123 | | |
| | | 124 | | public static async Task<FlowExecutionLease?> TryAcquireExecutionLeaseAsync( |
| | | 125 | | IFlowStateStore store, |
| | | 126 | | string flowId, |
| | | 127 | | DurableFlowOptions options, |
| | | 128 | | ILogger logger, |
| | | 129 | | TimeProvider? timeProvider = null, |
| | | 130 | | CancellationToken cancellationToken = default, |
| | | 131 | | string? jobTag = null) |
| | | 132 | | { |
| | 6301 | 133 | | ValidateOptions(options); |
| | | 134 | | |
| | 6301 | 135 | | var clock = timeProvider ?? TimeProvider.System; |
| | | 136 | | |
| | | 137 | | // The job driving this execution is recorded IN the lease id, so it lands atomically with |
| | | 138 | | // the acquire on every store and comes back through ObserveLeaseAsync: a later delivery of |
| | | 139 | | // that same job can then tell it is contending with its own first delivery (see |
| | | 140 | | // FlowLeaseContention). Without a job identity this is the plain 32-character id. |
| | 6301 | 141 | | var leaseId = FlowLeaseContention.NewLeaseId(jobTag); |
| | | 142 | | |
| | | 143 | | // Stamp the deadline BEFORE the call, not after it returns. The store starts the lease when |
| | | 144 | | // it executes the command; every millisecond after that — network latency, a delayed |
| | | 145 | | // continuation, a GC pause between the response arriving and this line running — is lease |
| | | 146 | | // time already spent. Anchoring afterwards handed that whole interval back to the client as |
| | | 147 | | // if it were still owned, so a worker could believe it held a 60s lease 20s past the point |
| | | 148 | | // another replica was free to take it. Anchoring first is conservative in the safe |
| | | 149 | | // direction: the client's deadline can only be EARLIER than the server's. |
| | 6301 | 150 | | var deadline = FlowExecutionLease.DeadlineFrom(clock, options.ExecutionLeaseDuration); |
| | | 151 | | |
| | 6301 | 152 | | if (!await store.TryAcquireLeaseAsync( |
| | 6301 | 153 | | flowId, |
| | 6301 | 154 | | leaseId, |
| | 6301 | 155 | | options.ExecutionLeaseDuration, |
| | 6301 | 156 | | cancellationToken).ConfigureAwait(false)) |
| | 4205 | 157 | | return null; |
| | | 158 | | |
| | | 159 | | // The constructor is throw-free after the option bounds above: it only assigns fields, |
| | | 160 | | // records the pre-call deadline, and starts the renewal loop (whose first Task.Delay faults |
| | | 161 | | // the loop task, never the constructor). Were that ever to change, lease expiry is the |
| | | 162 | | // backstop for the persisted row. |
| | 2096 | 163 | | return new FlowExecutionLease(store, flowId, leaseId, options, logger, clock, deadline); |
| | 6301 | 164 | | } |
| | | 165 | | |
| | | 166 | | public static async Task<bool> MutateAsync( |
| | | 167 | | IFlowStateStore store, |
| | | 168 | | string flowId, |
| | | 169 | | TimeSpan ttl, |
| | | 170 | | TimeProvider? timeProvider, |
| | | 171 | | Func<FlowState, bool> mutate, |
| | | 172 | | CancellationToken cancellationToken = default) |
| | | 173 | | { |
| | 192 | 174 | | for (var attempt = 0; attempt < MaxUpdateAttempts; attempt++) |
| | | 175 | | { |
| | 94 | 176 | | var state = await store.LoadAsync(flowId, cancellationToken).ConfigureAwait(false); |
| | 94 | 177 | | if (state is null) |
| | 6 | 178 | | return false; |
| | | 179 | | |
| | 88 | 180 | | if (!mutate(state)) |
| | 26 | 181 | | return true; |
| | | 182 | | |
| | 62 | 183 | | var expectedRevision = state.Revision; |
| | 62 | 184 | | state.Revision = checked(expectedRevision + 1); |
| | 62 | 185 | | var nowUtc = (timeProvider ?? TimeProvider.System).GetUtcNow().UtcDateTime; |
| | 62 | 186 | | state.UpdatedAtUtc = nowUtc; |
| | 62 | 187 | | if (await store.TryUpdateAsync( |
| | 62 | 188 | | flowId, |
| | 62 | 189 | | state, |
| | 62 | 190 | | expectedRevision, |
| | 62 | 191 | | // A lease-bypassing write (recovery, failure signal, operator) never shrinks a |
| | 62 | 192 | | // live run's ledger under a park it knows nothing about. |
| | 62 | 193 | | FlowStateRetention.EffectiveTtl(state, ttl, nowUtc), |
| | 62 | 194 | | leaseId: null, |
| | 62 | 195 | | cancellationToken).ConfigureAwait(false)) |
| | 46 | 196 | | return true; |
| | | 197 | | } |
| | | 198 | | |
| | 2 | 199 | | throw new InvalidOperationException( |
| | 2 | 200 | | $"Durable flow '{flowId}' changed repeatedly while applying a recovery update; retry the operation."); |
| | 78 | 201 | | } |
| | | 202 | | |
| | | 203 | | internal static void ValidateOptions(DurableFlowOptions options) |
| | | 204 | | { |
| | | 205 | | // Upper bounds close the "passes validation, throws mid-operation" gap — but only on the |
| | | 206 | | // knobs that actually reach the failing sink. StateExpiry and ExecutionLeaseDuration become |
| | | 207 | | // "now + value" stamps (store TTLs and lease deadlines) and never arm a timer themselves, |
| | | 208 | | // so they get the persistence bound; DefaultStepTimeout and ExecutionLeaseRenewInterval arm |
| | | 209 | | // BCL timers, so they get the timer ceiling; ProgressPersistenceInterval is only ever |
| | | 210 | | // compared against elapsed time (DurableFlowContext.ReportProgressAsync), so any |
| | | 211 | | // non-negative value is representable — a 60-day lease or progress throttle is a valid |
| | | 212 | | // configuration and must not fail startup. |
| | 9619 | 213 | | AsyncResponseChannelOptions.EnsurePersistedTtl(options.StateExpiry, nameof(DurableFlowOptions), nameof(options.S |
| | 9615 | 214 | | if (options.DefaultStepTimeout is { } defaultStepTimeout) |
| | 26 | 215 | | AsyncResponseChannelOptions.EnsureTimerBacked(defaultStepTimeout, nameof(DurableFlowOptions), nameof(options |
| | 9611 | 216 | | AsyncResponseChannelOptions.EnsurePersistedTtl(options.ExecutionLeaseDuration, nameof(DurableFlowOptions), nameo |
| | 9607 | 217 | | AsyncResponseChannelOptions.EnsureTimerBacked(options.ExecutionLeaseRenewInterval, nameof(DurableFlowOptions), n |
| | 9605 | 218 | | if (options.LedgerSizeWarningBytes is { } ledgerWarning && ledgerWarning <= 0) |
| | | 219 | | { |
| | 4 | 220 | | throw new InvalidOperationException( |
| | 4 | 221 | | $"{nameof(DurableFlowOptions)}.{nameof(options.LedgerSizeWarningBytes)} must be positive, or null to dis |
| | | 222 | | } |
| | 9601 | 223 | | if (options.MaxRetainedSteps is <= 0) |
| | 4 | 224 | | throw new InvalidOperationException($"{nameof(DurableFlowOptions)}.{nameof(options.MaxRetainedSteps)} must b |
| | 9597 | 225 | | if (options.ExecutionLeaseRenewInterval >= options.ExecutionLeaseDuration) |
| | | 226 | | { |
| | 2 | 227 | | throw new InvalidOperationException( |
| | 2 | 228 | | $"{nameof(DurableFlowOptions)}.{nameof(options.ExecutionLeaseRenewInterval)} must be shorter than " + |
| | 2 | 229 | | $"{nameof(DurableFlowOptions.ExecutionLeaseDuration)}."); |
| | | 230 | | } |
| | | 231 | | // Compared against elapsed time only (the contention poll arms pollDelay-sized timers), so |
| | | 232 | | // any positive value is representable; zero or less would cap the store-driven wait below |
| | | 233 | | // "no wait at all", which is a misconfiguration rather than a way to disable the feature. |
| | 9595 | 234 | | if (options.MaxLeaseContentionWait <= TimeSpan.Zero) |
| | 4 | 235 | | throw new InvalidOperationException($"{nameof(DurableFlowOptions)}.{nameof(options.MaxLeaseContentionWait)} |
| | 9591 | 236 | | if (options.ProgressPersistenceInterval < TimeSpan.Zero) |
| | 2 | 237 | | throw new InvalidOperationException($"{nameof(DurableFlowOptions)}.{nameof(options.ProgressPersistenceInterv |
| | | 238 | | // Timer remainders at or under the threshold arm an in-process Task.Delay, so the knob is |
| | | 239 | | // timer-backed; zero legitimately means "always suspend". |
| | 9589 | 240 | | AsyncResponseChannelOptions.EnsureTimerBackedAllowZero(options.TimerInProcessThreshold, nameof(DurableFlowOption |
| | 9589 | 241 | | } |
| | | 242 | | } |
| | | 243 | | |
| | | 244 | | /// <summary>One distributed durable-flow execution lease.</summary> |
| | | 245 | | internal sealed class FlowExecutionLease : IAsyncDisposable |
| | | 246 | | { |
| | | 247 | | private readonly IFlowStateStore _store; |
| | | 248 | | private readonly string _flowId; |
| | | 249 | | private readonly string _leaseId; |
| | | 250 | | private readonly DurableFlowOptions _options; |
| | | 251 | | private readonly ILogger _logger; |
| | | 252 | | private readonly TimeProvider _timeProvider; |
| | | 253 | | private readonly CancellationTokenSource _stop = new(); |
| | | 254 | | private readonly CancellationTokenSource _lost = new(); |
| | | 255 | | private readonly Task _renewal; |
| | | 256 | | private readonly Task _deadline; |
| | | 257 | | // DateTime ticks so the renewal loop's writes and the execution path's reads tear-free on |
| | | 258 | | // 32-bit runtimes and order via Volatile. |
| | | 259 | | private long _validUntilUtcTicks; |
| | | 260 | | private int _disposed; |
| | | 261 | | |
| | | 262 | | /// <summary> |
| | | 263 | | /// Longest single wait the deadline watcher arms. ExecutionLeaseDuration is validated as a |
| | | 264 | | /// PERSISTENCE bound, not a timer bound (see <see cref="FlowStateConcurrency.ValidateOptions"/>) — |
| | | 265 | | /// a 60-day lease is a legal configuration — so the watcher sleeps in chunks and re-reads the |
| | | 266 | | /// deadline rather than handing an out-of-range delay to a BCL timer. |
| | | 267 | | /// </summary> |
| | | 268 | | private static readonly TimeSpan MaxDeadlineChunk = TimeSpan.FromDays(1); |
| | | 269 | | |
| | | 270 | | /// <summary> |
| | | 271 | | /// Budget for joining the renewal and deadline loops on disposal. The renewal loop can be |
| | | 272 | | /// stuck inside a store call that ignores its cancellation token (a wedged connection, a |
| | | 273 | | /// database that accepts the request and never answers — the exact case the deadline watcher |
| | | 274 | | /// exists for); an unbounded join there wedged the whole worker job. Past the budget the |
| | | 275 | | /// loops are abandoned: the deadline watcher has already marked the lease lost and the |
| | | 276 | | /// server-side lease expires on its own. |
| | | 277 | | /// </summary> |
| | | 278 | | private static readonly TimeSpan DisposeJoinLimit = TimeSpan.FromSeconds(30); |
| | | 279 | | |
| | | 280 | | /// <summary> |
| | | 281 | | /// Budget for the final lease release on disposal. The release is one conditional write, so |
| | | 282 | | /// ten seconds is generous; past it the call is abandoned (cancelled, its outcome observed) |
| | | 283 | | /// and the server-side lease expires on its own — the same recovery the abandoned renewal |
| | | 284 | | /// loops rely on. Separate from <see cref="DisposeJoinLimit"/> because the two hang for |
| | | 285 | | /// different reasons: the loops are joined first and are usually idle, while the release is |
| | | 286 | | /// a fresh store call that a wedged connection can hold indefinitely even after a clean join. |
| | | 287 | | /// </summary> |
| | | 288 | | private static readonly TimeSpan ReleaseLimit = TimeSpan.FromSeconds(10); |
| | | 289 | | |
| | | 290 | | /// <param name="store">The flow state store the lease was acquired through.</param> |
| | | 291 | | /// <param name="flowId">The flow the lease protects.</param> |
| | | 292 | | /// <param name="leaseId">The identity of this lease within the flow's row.</param> |
| | | 293 | | /// <param name="options">Validated durable-flow options.</param> |
| | | 294 | | /// <param name="logger">Sink for renewal and deadline watcher events.</param> |
| | | 295 | | /// <param name="timeProvider">Clock used for deadline computation; <see cref="TimeProvider.System"/> when omitted.< |
| | | 296 | | /// <param name="acquiredDeadlineUtcTicks"> |
| | | 297 | | /// The conservative deadline for the lease this instance was handed, captured BEFORE the |
| | | 298 | | /// acquire call went out. Omitted only by callers that construct a lease without an acquire |
| | | 299 | | /// round trip (tests), where "now + duration" is exact. |
| | | 300 | | /// </param> |
| | | 301 | | public FlowExecutionLease( |
| | | 302 | | IFlowStateStore store, |
| | | 303 | | string flowId, |
| | | 304 | | string leaseId, |
| | | 305 | | DurableFlowOptions options, |
| | | 306 | | ILogger logger, |
| | | 307 | | TimeProvider? timeProvider = null, |
| | | 308 | | long? acquiredDeadlineUtcTicks = null) |
| | | 309 | | { |
| | | 310 | | _store = store; |
| | | 311 | | _flowId = flowId; |
| | | 312 | | _leaseId = leaseId; |
| | | 313 | | _options = options; |
| | | 314 | | _logger = logger; |
| | | 315 | | _timeProvider = timeProvider ?? TimeProvider.System; |
| | | 316 | | Volatile.Write( |
| | | 317 | | ref _validUntilUtcTicks, |
| | | 318 | | acquiredDeadlineUtcTicks ?? DeadlineFrom(_timeProvider, options.ExecutionLeaseDuration)); |
| | | 319 | | _renewal = RenewLoopAsync(); |
| | | 320 | | _deadline = DeadlineLoopAsync(); |
| | | 321 | | } |
| | | 322 | | |
| | | 323 | | /// <summary> |
| | | 324 | | /// "Now + duration" in UTC ticks, saturating instead of overflowing: <c>ExecutionLeaseDuration</c> |
| | | 325 | | /// is bounded as a persistence TTL, not a timer, so a 60-day lease near <see cref="DateTime.MaxValue"/> |
| | | 326 | | /// is a legal configuration that must not throw here. |
| | | 327 | | /// </summary> |
| | | 328 | | internal static long DeadlineFrom(TimeProvider timeProvider, TimeSpan duration) |
| | | 329 | | { |
| | | 330 | | var now = timeProvider.GetUtcNow().UtcDateTime; |
| | | 331 | | return duration > DateTime.MaxValue - now ? DateTime.MaxValue.Ticks : now.Add(duration).Ticks; |
| | | 332 | | } |
| | | 333 | | |
| | | 334 | | public CancellationToken LostToken => _lost.Token; |
| | | 335 | | |
| | | 336 | | /// <summary> |
| | | 337 | | /// Whether this lease can still fence a write. Callers holding a claimed, unrecorded result |
| | | 338 | | /// use it to choose the lease-less persistence path BEFORE surfacing the takeover signal. |
| | | 339 | | /// </summary> |
| | | 340 | | public bool IsLost => _lost.IsCancellationRequested |
| | | 341 | | || _timeProvider.GetUtcNow().UtcDateTime.Ticks >= Volatile.Read(ref _validUntilUtcTicks); |
| | | 342 | | |
| | | 343 | | /// <summary> |
| | | 344 | | /// Throws when the lease is lost. <paramref name="cause"/> (e.g. the exception that made the |
| | | 345 | | /// caller check) is attached as the inner exception so the real failure is not discarded. |
| | | 346 | | /// <para> |
| | | 347 | | /// A passed deadline counts as lost even before any renewal fails: the renewal loop only |
| | | 348 | | /// observes loss on a store round-trip, so a stop-the-world pause (GC, VM freeze, debugger) |
| | | 349 | | /// longer than the lease lets another worker take over while this side has seen nothing — |
| | | 350 | | /// its next step body would then run concurrently with the new holder's. Checkpoints are |
| | | 351 | | /// lease-fenced; side effects are fenced only by this guard, so it is conservative near the |
| | | 352 | | /// boundary by design: retrying from the checkpoint is always safe, a concurrent step is not. |
| | | 353 | | /// </para> |
| | | 354 | | /// </summary> |
| | | 355 | | public void ThrowIfLost(Exception? cause = null) |
| | | 356 | | { |
| | | 357 | | if (!_lost.IsCancellationRequested |
| | | 358 | | && _timeProvider.GetUtcNow().UtcDateTime.Ticks < Volatile.Read(ref _validUntilUtcTicks)) |
| | | 359 | | return; |
| | | 360 | | |
| | | 361 | | MarkLost(); |
| | | 362 | | throw new InvalidOperationException($"Durable flow '{_flowId}' lost its execution lease; the worker will retry f |
| | | 363 | | } |
| | | 364 | | |
| | | 365 | | public async Task SaveAsync(FlowState state, TimeSpan ttl, CancellationToken cancellationToken = default, Exception? |
| | | 366 | | { |
| | | 367 | | ThrowIfLost(cause); |
| | | 368 | | var expectedRevision = state.Revision; |
| | | 369 | | state.Revision = checked(expectedRevision + 1); |
| | | 370 | | var nowUtc = _timeProvider.GetUtcNow().UtcDateTime; |
| | | 371 | | state.UpdatedAtUtc = nowUtc; |
| | | 372 | | |
| | | 373 | | try |
| | | 374 | | { |
| | | 375 | | if (await _store.TryUpdateAsync( |
| | | 376 | | _flowId, |
| | | 377 | | state, |
| | | 378 | | expectedRevision, |
| | | 379 | | // Every checkpoint carries the ledger's retention floor forward (see |
| | | 380 | | // FlowStateRetention): the executor's per-attempt save and an ancestor's |
| | | 381 | | // re-park stamp the plain StateExpiry, and used to shrink a ledger a |
| | | 382 | | // descendant had extended for a wait still in progress. |
| | | 383 | | FlowStateRetention.EffectiveTtl(state, ttl, nowUtc), |
| | | 384 | | _leaseId, |
| | | 385 | | cancellationToken).ConfigureAwait(false)) |
| | | 386 | | return; |
| | | 387 | | } |
| | | 388 | | catch |
| | | 389 | | { |
| | | 390 | | state.Revision = expectedRevision; |
| | | 391 | | MarkLost(); |
| | | 392 | | |
| | | 393 | | // The store exception propagates; keep the failure this save was recording from |
| | | 394 | | // vanishing with it. |
| | | 395 | | if (cause is not null) |
| | | 396 | | _logger.LogWarning(cause, "Durable flow '{FlowId}' failed to checkpoint; the failure it was recording is |
| | | 397 | | throw; |
| | | 398 | | } |
| | | 399 | | |
| | | 400 | | state.Revision = expectedRevision; |
| | | 401 | | MarkLost(); |
| | | 402 | | throw await CreateSaveRejectedExceptionAsync(expectedRevision, cause, cancellationToken).ConfigureAwait(false); |
| | | 403 | | } |
| | | 404 | | |
| | | 405 | | /// <summary> |
| | | 406 | | /// Builds the exception for a rejected checkpoint write. The store's compare-and-swap only |
| | | 407 | | /// returns <c>false</c>, so the reason is diagnosed with a best-effort re-read: a revision |
| | | 408 | | /// conflict — a concurrent lease-bypassing writer such as <c>RecoverAsync</c>, <c>FailAsync</c>, |
| | | 409 | | /// or an operator parking the run — is reported as such instead of as a lost lease, which sent |
| | | 410 | | /// operators hunting phantom lease problems. Behavior is unchanged either way: the lease is |
| | | 411 | | /// abandoned (<see cref="MarkLost"/> already ran) and the delivery retries from the last |
| | | 412 | | /// checkpoint; <paramref name="cause"/> rides along as the inner exception so the failure that |
| | | 413 | | /// triggered the save is not discarded. |
| | | 414 | | /// </summary> |
| | | 415 | | private async Task<InvalidOperationException> CreateSaveRejectedExceptionAsync( |
| | | 416 | | long expectedRevision, |
| | | 417 | | Exception? cause, |
| | | 418 | | CancellationToken cancellationToken) |
| | | 419 | | { |
| | | 420 | | var reason = "its execution lease was no longer held (expired or taken over)"; |
| | | 421 | | try |
| | | 422 | | { |
| | | 423 | | var current = await _store.LoadAsync(_flowId, cancellationToken).ConfigureAwait(false); |
| | | 424 | | if (current is null) |
| | | 425 | | reason = "its ledger entry is gone (expired or deleted)"; |
| | | 426 | | else if (current.Revision != expectedRevision) |
| | | 427 | | reason = $"a concurrent write advanced the ledger (revision {expectedRevision} -> {current.Revision}: a |
| | | 428 | | } |
| | | 429 | | catch |
| | | 430 | | { |
| | | 431 | | // Best-effort diagnosis only — the rejection itself is what matters. |
| | | 432 | | } |
| | | 433 | | |
| | | 434 | | return new InvalidOperationException( |
| | | 435 | | $"Durable flow '{_flowId}' could not checkpoint because {reason}; the worker abandons this execution and the |
| | | 436 | | cause); |
| | | 437 | | } |
| | | 438 | | |
| | | 439 | | private async Task RenewLoopAsync() |
| | | 440 | | { |
| | | 441 | | while (!_stop.IsCancellationRequested) |
| | | 442 | | { |
| | | 443 | | try |
| | | 444 | | { |
| | | 445 | | await Task.Delay(_options.ExecutionLeaseRenewInterval, _timeProvider, _stop.Token).ConfigureAwait(false) |
| | | 446 | | |
| | | 447 | | // Same anchoring rule as acquisition: the renewed lease starts when the store runs |
| | | 448 | | // the command, so the deadline is measured from before the call, not from whenever |
| | | 449 | | // the answer gets back here. Published only on success, so a failed renewal never |
| | | 450 | | // extends anything. |
| | | 451 | | var renewedDeadline = DeadlineFrom(_timeProvider, _options.ExecutionLeaseDuration); |
| | | 452 | | |
| | | 453 | | if (!await _store.TryRenewLeaseAsync( |
| | | 454 | | _flowId, |
| | | 455 | | _leaseId, |
| | | 456 | | _options.ExecutionLeaseDuration, |
| | | 457 | | _stop.Token).ConfigureAwait(false)) |
| | | 458 | | { |
| | | 459 | | MarkLost(); |
| | | 460 | | return; |
| | | 461 | | } |
| | | 462 | | |
| | | 463 | | Volatile.Write(ref _validUntilUtcTicks, renewedDeadline); |
| | | 464 | | } |
| | | 465 | | catch (OperationCanceledException) when (_stop.IsCancellationRequested) |
| | | 466 | | { |
| | | 467 | | return; |
| | | 468 | | } |
| | | 469 | | catch (Exception ex) |
| | | 470 | | { |
| | | 471 | | _logger.LogWarning(ex, "Failed to renew durable flow {FlowId} execution lease; retrying before expiry.", |
| | | 472 | | if (_timeProvider.GetUtcNow().UtcDateTime.Ticks >= Volatile.Read(ref _validUntilUtcTicks)) |
| | | 473 | | { |
| | | 474 | | MarkLost(); |
| | | 475 | | return; |
| | | 476 | | } |
| | | 477 | | } |
| | | 478 | | } |
| | | 479 | | } |
| | | 480 | | |
| | | 481 | | /// <summary> |
| | | 482 | | /// Cancels <see cref="LostToken"/> when the lease deadline passes, on a clock of its own. |
| | | 483 | | /// <para> |
| | | 484 | | /// <see cref="RenewLoopAsync"/> cannot be trusted to do this: it only learns the lease is gone |
| | | 485 | | /// by completing a store round-trip, so a renewal call that hangs — a wedged connection, a |
| | | 486 | | /// database that accepts the request and never answers — leaves the token live indefinitely |
| | | 487 | | /// while the server-side lease expires and another replica takes the flow over. Checkpoints |
| | | 488 | | /// stay fenced regardless (<see cref="ThrowIfLost"/> and the lease-fenced CAS both check the |
| | | 489 | | /// clock), but anything watching the TOKEN — a step body, a linked operation — saw nothing. |
| | | 490 | | /// This loop closes that gap: it re-reads the deadline each pass, so a successful renewal |
| | | 491 | | /// simply pushes it out, and it fires whether or not the renewal path is responsive. |
| | | 492 | | /// </para> |
| | | 493 | | /// </summary> |
| | | 494 | | private async Task DeadlineLoopAsync() |
| | | 495 | | { |
| | | 496 | | try |
| | | 497 | | { |
| | | 498 | | while (!_stop.IsCancellationRequested && !_lost.IsCancellationRequested) |
| | | 499 | | { |
| | | 500 | | var remaining = new DateTime(Volatile.Read(ref _validUntilUtcTicks), DateTimeKind.Utc) |
| | | 501 | | - _timeProvider.GetUtcNow().UtcDateTime; |
| | | 502 | | |
| | | 503 | | if (remaining <= TimeSpan.Zero) |
| | | 504 | | { |
| | | 505 | | _logger.LogWarning( |
| | | 506 | | "Durable flow {FlowId} execution lease reached its deadline without a successful renewal; abando |
| | | 507 | | _flowId); |
| | | 508 | | MarkLost(); |
| | | 509 | | return; |
| | | 510 | | } |
| | | 511 | | |
| | | 512 | | await Task.Delay( |
| | | 513 | | remaining < MaxDeadlineChunk ? remaining : MaxDeadlineChunk, |
| | | 514 | | _timeProvider, |
| | | 515 | | _stop.Token).ConfigureAwait(false); |
| | | 516 | | } |
| | | 517 | | } |
| | | 518 | | catch (OperationCanceledException) when (_stop.IsCancellationRequested) |
| | | 519 | | { |
| | | 520 | | // Normal completion: the execution finished and disposal stopped the watcher. |
| | | 521 | | } |
| | | 522 | | } |
| | | 523 | | |
| | | 524 | | private void MarkLost() |
| | | 525 | | { |
| | | 526 | | try |
| | | 527 | | { |
| | | 528 | | _lost.Cancel(); |
| | | 529 | | } |
| | | 530 | | catch (ObjectDisposedException) |
| | | 531 | | { |
| | | 532 | | // Disposal won the race. |
| | | 533 | | } |
| | | 534 | | } |
| | | 535 | | |
| | | 536 | | public async ValueTask DisposeAsync() |
| | | 537 | | { |
| | | 538 | | if (Interlocked.Exchange(ref _disposed, 1) != 0) |
| | | 539 | | return; |
| | | 540 | | |
| | | 541 | | _stop.Cancel(); |
| | | 542 | | try |
| | | 543 | | { |
| | | 544 | | // Bounded join (see DisposeJoinLimit): both loops swallow their own exceptions, so an |
| | | 545 | | // abandoned task cannot fault unobserved. |
| | | 546 | | await Task.WhenAll(_renewal, _deadline).WaitAsync(DisposeJoinLimit, _timeProvider).ConfigureAwait(false); |
| | | 547 | | } |
| | | 548 | | catch (TimeoutException) |
| | | 549 | | { |
| | | 550 | | // The store call the renewal loop is stuck in ignores cancellation, so releasing the |
| | | 551 | | // lease through the same store would hang this disposal all over again. Skip the |
| | | 552 | | // release (the server-side lease expires) and leave the cancellation sources |
| | | 553 | | // undisposed for the abandoned loops. |
| | | 554 | | _logger.LogWarning( |
| | | 555 | | "Durable flow {FlowId} execution lease loops did not stop within {DisposeJoinLimit}; abandoning them (th |
| | | 556 | | _flowId, |
| | | 557 | | DisposeJoinLimit); |
| | | 558 | | return; |
| | | 559 | | } |
| | | 560 | | |
| | | 561 | | // Bounded release (see ReleaseLimit), with a token the store can honor. An unbounded, |
| | | 562 | | // uncancelable release kept a FINISHED execution's disposal — and with it the executor's |
| | | 563 | | // `await using`, the job's DI scope, the worker slot, and the transport acknowledgement — |
| | | 564 | | // pending for as long as a wedged store took to answer, which can be forever. |
| | | 565 | | var releaseCancellation = new CancellationTokenSource(); |
| | | 566 | | Task? release = null; |
| | | 567 | | try |
| | | 568 | | { |
| | | 569 | | release = _store.ReleaseLeaseAsync(_flowId, _leaseId, releaseCancellation.Token); |
| | | 570 | | await release.WaitAsync(ReleaseLimit, _timeProvider).ConfigureAwait(false); |
| | | 571 | | releaseCancellation.Dispose(); |
| | | 572 | | } |
| | | 573 | | catch (TimeoutException) when (release is { IsCompleted: false }) |
| | | 574 | | { |
| | | 575 | | // The budget lapsed with the store still silent (a TimeoutException thrown BY the |
| | | 576 | | // store completes the task first and takes the branch below). Cancel what can be |
| | | 577 | | // cancelled, observe whatever the abandoned call eventually does, and move on: the |
| | | 578 | | // server-side lease expires on its own, exactly as when the renewal loops are abandoned. |
| | | 579 | | releaseCancellation.Cancel(); |
| | | 580 | | _logger.LogWarning( |
| | | 581 | | "Durable flow {FlowId} execution lease release did not complete within {ReleaseLimit}; abandoning it (th |
| | | 582 | | _flowId, |
| | | 583 | | ReleaseLimit); |
| | | 584 | | ObserveAbandonedRelease(release, releaseCancellation); |
| | | 585 | | } |
| | | 586 | | catch (Exception ex) |
| | | 587 | | { |
| | | 588 | | releaseCancellation.Dispose(); |
| | | 589 | | _logger.LogWarning(ex, "Failed to release durable flow {FlowId} execution lease; it will expire.", _flowId); |
| | | 590 | | } |
| | | 591 | | |
| | | 592 | | _stop.Dispose(); |
| | | 593 | | _lost.Dispose(); |
| | | 594 | | } |
| | | 595 | | |
| | | 596 | | /// <summary> |
| | | 597 | | /// Attaches the one continuation an abandoned release needs: its eventual fault is observed |
| | | 598 | | /// (and logged, so a store that finally answers with an error is not an unobserved-task |
| | | 599 | | /// event) and the cancellation source it still holds is disposed only once it can no longer |
| | | 600 | | /// be touched. |
| | | 601 | | /// </summary> |
| | | 602 | | private void ObserveAbandonedRelease(Task release, CancellationTokenSource releaseCancellation) |
| | | 603 | | => _ = release.ContinueWith( |
| | | 604 | | (task, state) => |
| | | 605 | | { |
| | | 606 | | var (lease, cancellation) = ((FlowExecutionLease, CancellationTokenSource))state!; |
| | | 607 | | if (task.IsFaulted) |
| | | 608 | | { |
| | | 609 | | lease._logger.LogWarning( |
| | | 610 | | task.Exception?.GetBaseException(), |
| | | 611 | | "The abandoned release of durable flow {FlowId}'s execution lease eventually failed; the lease e |
| | | 612 | | lease._flowId); |
| | | 613 | | } |
| | | 614 | | else |
| | | 615 | | { |
| | | 616 | | lease._logger.LogDebug( |
| | | 617 | | "The abandoned release of durable flow {FlowId}'s execution lease eventually completed ({Status} |
| | | 618 | | lease._flowId, |
| | | 619 | | task.Status); |
| | | 620 | | } |
| | | 621 | | |
| | | 622 | | cancellation.Dispose(); |
| | | 623 | | }, |
| | | 624 | | (this, releaseCancellation), |
| | | 625 | | CancellationToken.None, |
| | | 626 | | TaskContinuationOptions.ExecuteSynchronously, |
| | | 627 | | TaskScheduler.Default); |
| | | 628 | | } |