| | | 1 | | namespace AsyncResponse; |
| | | 2 | | |
| | | 3 | | /// <summary> |
| | | 4 | | /// The ledger retention floor (<see cref="FlowState.RetainUntilUtc"/>) at the write sites. |
| | | 5 | | /// <para> |
| | | 6 | | /// A park longer than the ordinary idle <c>StateExpiry</c> — a timer sleep, an awaited step's |
| | | 7 | | /// window, a child flow's own park — needs the parked run's ledger AND every ancestor waiting on |
| | | 8 | | /// it to outlive the wait. The TTL stamped by the park's own save covers that only until the |
| | | 9 | | /// next write: every store recomputes expiry as "now + ttl", and the writers that can race a |
| | | 10 | | /// park (an ancestor's replay re-parking on a stale child snapshot, an executor's per-attempt |
| | | 11 | | /// save, a recovery or operator mutation) know nothing about the wait and stamp the plain |
| | | 12 | | /// <c>StateExpiry</c>. The floor rides in the ledger itself, so whoever writes the ledger next |
| | | 13 | | /// carries it forward: the TTL a write stamps is raised to reach the floor. That is what makes |
| | | 14 | | /// a descendant's extension of an ancestor durable across the ancestor's own checkpoints, and |
| | | 15 | | /// what lets the extension prove — by re-reading the ancestor — that a concurrent write which |
| | | 16 | | /// beat its compare-and-swap left adequate retention behind. |
| | | 17 | | /// </para> |
| | | 18 | | /// <para> |
| | | 19 | | /// Terminal runs ignore the floor: they have no wait in progress, and a failed run should not |
| | | 20 | | /// be retained for the length of the sleep it never finished. |
| | | 21 | | /// </para> |
| | | 22 | | /// </summary> |
| | | 23 | | internal static class FlowStateRetention |
| | | 24 | | { |
| | | 25 | | /// <summary> |
| | | 26 | | /// The TTL a write must stamp for <paramref name="state"/>: <paramref name="requested"/>, |
| | | 27 | | /// raised to reach the state's retention floor when the run is live and the floor is further |
| | | 28 | | /// out. Saturated at the persistence ceiling (clock skew between replicas could otherwise push |
| | | 29 | | /// a floor stamped elsewhere a hair past it). |
| | | 30 | | /// </summary> |
| | | 31 | | public static TimeSpan EffectiveTtl(FlowState state, TimeSpan requested, DateTime nowUtc) |
| | | 32 | | { |
| | 13024 | 33 | | if (state.RetainUntilUtc is not { } floor || IsTerminal(state.Status)) |
| | 6850 | 34 | | return requested; |
| | | 35 | | |
| | 6174 | 36 | | var needed = floor - nowUtc; |
| | 6174 | 37 | | if (needed <= requested) |
| | 2491 | 38 | | return requested; |
| | | 39 | | |
| | 3683 | 40 | | return needed > AsyncResponseChannelOptions.MaxPersistenceTtl |
| | 3683 | 41 | | ? AsyncResponseChannelOptions.MaxPersistenceTtl |
| | 3683 | 42 | | : needed; |
| | | 43 | | } |
| | | 44 | | |
| | | 45 | | /// <summary> |
| | | 46 | | /// Raises the floor of <paramref name="state"/> to <paramref name="nowUtc"/> + |
| | | 47 | | /// <paramref name="ttl"/> when that is further out than the current one (never lowers it) and |
| | | 48 | | /// returns the instant the floor now sits at. |
| | | 49 | | /// </summary> |
| | | 50 | | public static DateTime RaiseFloor(FlowState state, DateTime nowUtc, TimeSpan ttl) |
| | | 51 | | { |
| | 2299 | 52 | | var until = FloorAt(nowUtc, ttl); |
| | 2299 | 53 | | if (state.RetainUntilUtc is not { } floor || until > floor) |
| | 2177 | 54 | | state.RetainUntilUtc = until; |
| | | 55 | | |
| | 2299 | 56 | | return state.RetainUntilUtc!.Value; |
| | | 57 | | } |
| | | 58 | | |
| | | 59 | | /// <summary>The instant a floor stamped now for <paramref name="ttl"/> sits at (saturating).</summary> |
| | 2951 | 60 | | public static DateTime FloorAt(DateTime nowUtc, TimeSpan ttl) => AddSaturating(nowUtc, ttl); |
| | | 61 | | |
| | | 62 | | /// <summary>Whether the floor of <paramref name="state"/> already reaches <paramref name="until"/>.</summary> |
| | | 63 | | public static bool Covers(FlowState state, DateTime until) |
| | 656 | 64 | | => state.RetainUntilUtc is { } floor && floor >= until; |
| | | 65 | | |
| | | 66 | | private static bool IsTerminal(FlowRunStatus status) |
| | 7612 | 67 | | => status is FlowRunStatus.Succeeded or FlowRunStatus.Failed; |
| | | 68 | | |
| | | 69 | | private static DateTime AddSaturating(DateTime instant, TimeSpan ttl) |
| | 2951 | 70 | | => ttl > DateTime.MaxValue - instant ? DateTime.MaxValue : instant + ttl; |
| | | 71 | | } |