| | | 1 | | using System.Security.Cryptography; |
| | | 2 | | using System.Text; |
| | | 3 | | |
| | | 4 | | namespace AsyncResponse; |
| | | 5 | | |
| | | 6 | | /// <summary>What a contended wake-up may conclude from two observations of the lease in its way.</summary> |
| | | 7 | | internal enum FlowLeaseContentionVerdict |
| | | 8 | | { |
| | | 9 | | /// <summary> |
| | | 10 | | /// The lease has not changed since the baseline: no proof of a live holder. A dead holder's |
| | | 11 | | /// lease reads exactly like this, so the wake-up keeps waiting (to the persisted expiry). |
| | | 12 | | /// </summary> |
| | | 13 | | KeepWaiting, |
| | | 14 | | |
| | | 15 | | /// <summary> |
| | | 16 | | /// A live worker acquired or renewed the lease while this wake-up waited, and that execution |
| | | 17 | | /// is driven by a DIFFERENT job (or one of the two carries no job identity). The holder's own |
| | | 18 | | /// job is still unacknowledged at the broker, so this delivery is redundant and safe to ack. |
| | | 19 | | /// </summary> |
| | | 20 | | AcknowledgeDuplicate, |
| | | 21 | | |
| | | 22 | | /// <summary> |
| | | 23 | | /// A live worker holds the lease — and the job driving it is THIS delivery's job. The broker |
| | | 24 | | /// redelivered a job whose handler is still running (an in-flight ceiling lapsed), so this |
| | | 25 | | /// delivery is the only copy of the run's wake-up the broker still has: never a duplicate. |
| | | 26 | | /// </summary> |
| | | 27 | | HolderOwnJobRedelivered |
| | | 28 | | } |
| | | 29 | | |
| | | 30 | | /// <summary> |
| | | 31 | | /// The job identity recorded with a durable-flow execution lease, and the decision a contended |
| | | 32 | | /// wake-up takes from it. Pure, so the rule is testable without a store, a clock, or a transport. |
| | | 33 | | /// <para> |
| | | 34 | | /// The identity rides INSIDE the lease id — <c>{guid:N}.{tag}</c> — because that is the one value |
| | | 35 | | /// every store already writes atomically with the acquire and reports back through |
| | | 36 | | /// <see cref="IFlowStateStore.ObserveLeaseAsync"/>: no store, schema, or wire change. The tag is a |
| | | 37 | | /// fixed-width digest of <see cref="WorkerJobEnvelope.JobId"/> rather than the id itself, since a |
| | | 38 | | /// job id is a wire value a foreign producer controls and the relational stores keep the lease id |
| | | 39 | | /// in a 64-character column (32 + 1 + <see cref="JobTagLength"/> = 55). |
| | | 40 | | /// </para> |
| | | 41 | | /// </summary> |
| | | 42 | | internal static class FlowLeaseContention |
| | | 43 | | { |
| | | 44 | | /// <summary>Characters of base64url(SHA-256) kept as the tag: 132 bits.</summary> |
| | | 45 | | internal const int JobTagLength = 22; |
| | | 46 | | |
| | | 47 | | private const int GuidLength = 32; |
| | | 48 | | private const char Separator = '.'; |
| | | 49 | | |
| | | 50 | | /// <summary>The lease tag for <paramref name="jobId"/>, or <c>null</c> for a job without an identity.</summary> |
| | | 51 | | public static string? JobTag(string? jobId) |
| | | 52 | | { |
| | 2071 | 53 | | if (string.IsNullOrEmpty(jobId)) |
| | 182 | 54 | | return null; |
| | | 55 | | |
| | 1889 | 56 | | Span<byte> digest = stackalloc byte[SHA256.HashSizeInBytes]; |
| | 1889 | 57 | | SHA256.HashData(Encoding.UTF8.GetBytes(jobId), digest); |
| | | 58 | | |
| | | 59 | | // 32 bytes encode to 44 base64 characters (padding included); the tag is the first 22, |
| | | 60 | | // rewritten to the URL-safe alphabet so it is inert in every store's id column. |
| | 1889 | 61 | | Span<char> encoded = stackalloc char[44]; |
| | 1889 | 62 | | Convert.TryToBase64Chars(digest, encoded, out _); |
| | 86894 | 63 | | for (var i = 0; i < JobTagLength; i++) |
| | | 64 | | { |
| | 41558 | 65 | | encoded[i] = encoded[i] switch |
| | 41558 | 66 | | { |
| | 648 | 67 | | '+' => '-', |
| | 683 | 68 | | '/' => '_', |
| | 40227 | 69 | | var other => other |
| | 41558 | 70 | | }; |
| | | 71 | | } |
| | | 72 | | |
| | 1889 | 73 | | return new string(encoded[..JobTagLength]); |
| | | 74 | | } |
| | | 75 | | |
| | | 76 | | /// <summary> |
| | | 77 | | /// A fresh lease id recording <paramref name="jobTag"/>; the plain 32-character id when the |
| | | 78 | | /// execution is not driven by an identified job (a direct call, or a job written before |
| | | 79 | | /// <see cref="WorkerJobEnvelope.JobId"/> existed). |
| | | 80 | | /// </summary> |
| | | 81 | | public static string NewLeaseId(string? jobTag) |
| | | 82 | | { |
| | 6335 | 83 | | var unique = Guid.NewGuid().ToString("N"); |
| | 6335 | 84 | | return jobTag is null ? unique : string.Concat(unique, ".", jobTag); |
| | | 85 | | } |
| | | 86 | | |
| | | 87 | | /// <summary> |
| | | 88 | | /// The job tag recorded in <paramref name="leaseId"/>, or <c>null</c> when it carries none: a |
| | | 89 | | /// lease issued by an older build, by a direct execution, or by anything else that does not |
| | | 90 | | /// write this exact shape. |
| | | 91 | | /// </summary> |
| | | 92 | | public static string? JobTagOf(string? leaseId) |
| | 240 | 93 | | => leaseId is { Length: GuidLength + 1 + JobTagLength } && leaseId[GuidLength] == Separator |
| | 240 | 94 | | ? leaseId[(GuidLength + 1)..] |
| | 240 | 95 | | : null; |
| | | 96 | | |
| | | 97 | | /// <summary> |
| | | 98 | | /// Judges the lease in the way of a wake-up driven by the job tagged <paramref name="ownJobTag"/> |
| | | 99 | | /// (<c>null</c> when the delivery has no identity), given the first observation of that lease |
| | | 100 | | /// and the current one. |
| | | 101 | | /// </summary> |
| | | 102 | | public static FlowLeaseContentionVerdict Judge( |
| | | 103 | | FlowLeaseObservation baseline, |
| | | 104 | | FlowLeaseObservation observed, |
| | | 105 | | string? ownJobTag) |
| | | 106 | | { |
| | | 107 | | // Only a worker that acquired or renewed the lease AFTER the baseline can have written a |
| | | 108 | | // different owner or a later expiry; an unchanged pair proves nothing. |
| | 4117 | 109 | | var changed = !string.Equals(observed.LeaseId, baseline.LeaseId, StringComparison.Ordinal) |
| | 4117 | 110 | | || observed.ExpiresAtUtc > baseline.ExpiresAtUtc; |
| | 4117 | 111 | | if (!changed) |
| | 3889 | 112 | | return FlowLeaseContentionVerdict.KeepWaiting; |
| | | 113 | | |
| | 228 | 114 | | return ownJobTag is not null |
| | 228 | 115 | | && string.Equals(JobTagOf(observed.LeaseId), ownJobTag, StringComparison.Ordinal) |
| | 228 | 116 | | ? FlowLeaseContentionVerdict.HolderOwnJobRedelivered |
| | 228 | 117 | | : FlowLeaseContentionVerdict.AcknowledgeDuplicate; |
| | | 118 | | } |
| | | 119 | | } |