| | | 1 | | namespace AsyncResponse; |
| | | 2 | | |
| | | 3 | | /// <summary> |
| | | 4 | | /// Shared options for an async-response channel (the response/recovery substrate). Concrete channels |
| | | 5 | | /// extend this with their own transport-specific settings (key/subject prefixes, bucket names, …). |
| | | 6 | | /// </summary> |
| | | 7 | | public abstract class AsyncResponseChannelOptions |
| | | 8 | | { |
| | | 9 | | /// <summary> |
| | | 10 | | /// How long persisted <see cref="RecoveryState"/> entries live, bounding how long after a |
| | | 11 | | /// crash/redeploy a late response can still trigger the lost-subscriber callbacks. Set it |
| | | 12 | | /// comfortably above your longest-running flow. (For the in-memory channel this is process-local |
| | | 13 | | /// and lost on exit.) |
| | | 14 | | /// </summary> |
| | 27919 | 15 | | public TimeSpan RecoveryStateExpiry { get; set; } = TimeSpan.FromDays(7); |
| | | 16 | | |
| | | 17 | | /// <summary> |
| | | 18 | | /// Default timeout applied to waiters that do not specify <c>WithTimeout</c>. When <c>null</c> |
| | | 19 | | /// (the default), <see cref="RecoveryStateExpiry"/> is used — waits are never infinite, so a |
| | | 20 | | /// response that never arrives faults the waiter with a <see cref="TimeoutException"/> instead of |
| | | 21 | | /// hanging forever. |
| | | 22 | | /// </summary> |
| | 19316 | 23 | | public TimeSpan? DefaultTimeout { get; set; } |
| | | 24 | | |
| | | 25 | | /// <summary> |
| | | 26 | | /// How long disposing a waiter may DRAIN an in-flight delivery before abandoning it. Disposal |
| | | 27 | | /// settles the response task only after a dispatch that already claimed a message — an |
| | | 28 | | /// <c>Until</c> predicate running user code — has finished, so a delivered response is never |
| | | 29 | | /// reported as canceled. If that dispatch is still running when this budget lapses, the |
| | | 30 | | /// response task is faulted with <see cref="AsyncResponseIndeterminateDeliveryException"/> |
| | | 31 | | /// rather than canceled: the delivery outcome is unknown, and a cancellation would invite |
| | | 32 | | /// re-attaching to a correlation id whose response may already be consumed (durable flows |
| | | 33 | | /// instead restart the idempotent step fresh). Default: 30 seconds — comfortably above any |
| | | 34 | | /// healthy predicate, well below a stuck one holding host shutdown hostage. |
| | | 35 | | /// </summary> |
| | 16999 | 36 | | public TimeSpan DisposalDrainTimeout { get; set; } = TimeSpan.FromSeconds(30); |
| | | 37 | | |
| | | 38 | | /// <summary> |
| | | 39 | | /// The largest delay the BCL's timer plumbing accepts (<c>uint.MaxValue - 1</c> ms, ~49.7 |
| | | 40 | | /// days): CancellationTokenSource timers, <see cref="Task.Delay(TimeSpan)"/>, and |
| | | 41 | | /// <c>Task.WaitAsync</c> all reject longer values at arming — which, for a waiter, is AFTER |
| | | 42 | | /// the subscription and recovery state already exist. Timer-armed values are therefore |
| | | 43 | | /// bounded here and at waiter creation instead. |
| | | 44 | | /// </summary> |
| | 16 | 45 | | internal static readonly TimeSpan MaxTimerBackedTimeout = TimeSpan.FromMilliseconds(uint.MaxValue - 1); |
| | | 46 | | |
| | | 47 | | /// <summary> |
| | | 48 | | /// Upper bound for persisted-TTL knobs (10 years): far beyond any practical retention, small |
| | | 49 | | /// enough that the "now + TTL" expiry stamp every store computes can never overflow |
| | | 50 | | /// <see cref="DateTime"/>/<see cref="DateTimeOffset"/> arithmetic — a |
| | | 51 | | /// <see cref="TimeSpan.MaxValue"/> expiry used to pass validation and then throw |
| | | 52 | | /// <see cref="ArgumentOutOfRangeException"/> at the first recovery-state save. |
| | | 53 | | /// </summary> |
| | 16 | 54 | | internal static readonly TimeSpan MaxPersistenceTtl = TimeSpan.FromDays(3650); |
| | | 55 | | |
| | | 56 | | /// <summary> |
| | | 57 | | /// The portable correlation-id length in UTF-16 code units — the width of the |
| | | 58 | | /// <c>correlation_id</c> column in the SQL Server channel's tables, and the tightest bound any |
| | | 59 | | /// bundled channel imposes. Validated where a correlation id enters the library so an |
| | | 60 | | /// over-long id is rejected at the call site instead of truncating or failing at its first |
| | | 61 | | /// database write, halfway through a conversation. |
| | | 62 | | /// </summary> |
| | | 63 | | public const int MaxCorrelationIdLength = 400; |
| | | 64 | | |
| | | 65 | | /// <summary> |
| | | 66 | | /// Applies the portable correlation-id contract, returning the rejection message or |
| | | 67 | | /// <c>null</c>. Length is bounded by the relational column; surrounding spaces are rejected |
| | | 68 | | /// because SQL Server pads the shorter operand of an equality comparison — even under a binary |
| | | 69 | | /// collation — so <c>"abc "</c> and <c>"abc"</c> are ONE key to the database while the library |
| | | 70 | | /// compares them ordinally and would route their responses to different waiters; control |
| | | 71 | | /// characters are rejected because the stores disagree about them (see |
| | | 72 | | /// <see cref="PortableText.IndexOfControlCharacter"/>). |
| | | 73 | | /// <para> |
| | | 74 | | /// The id is quoted back through <see cref="DiagnosticText.EscapedExcerpt"/>, never raw: the |
| | | 75 | | /// rejection is logged at Error and copied onto the activity status, and the two rejections that |
| | | 76 | | /// quote the id are the two whose offending unit — a CR/LF, an unpaired surrogate — may sit |
| | | 77 | | /// inside the quoted 40 units. Quoted raw, the inbound id wrote its own log lines. |
| | | 78 | | /// </para> |
| | | 79 | | /// <para> |
| | | 80 | | /// The rules kept in <see cref="PortableText"/> are the ones this contract genuinely SHARES |
| | | 81 | | /// with <c>FlowStateConcurrency.FlowIdNotPortable</c>. Two flow-id rules are deliberately not |
| | | 82 | | /// mirrored here because correlation ids have no such sink: the UTF-8 byte cap exists for |
| | | 83 | | /// Cosmos DB item ids, and the <c>/ \ ? #</c> character set is Cosmos's id syntax — no bundled |
| | | 84 | | /// channel persists a correlation id to Cosmos. Adding either here would reject ids that every |
| | | 85 | | /// shipped channel handles correctly. |
| | | 86 | | /// </para> |
| | | 87 | | /// </summary> |
| | | 88 | | internal static string? CorrelationIdNotPortable(string correlationId) |
| | | 89 | | { |
| | | 90 | | // Length-guarded before the [0]/[^1] probe below: this method's contract is to RETURN a |
| | | 91 | | // rejection for an unusable id, and an empty one indexing off the end would throw |
| | | 92 | | // IndexOutOfRangeException out of the very method whose job is to explain bad ids. |
| | 15274 | 93 | | if (correlationId.Length == 0) |
| | 2 | 94 | | return "CorrelationId is empty. An id must carry enough information to route a response back to its waiter." |
| | | 95 | | |
| | 15272 | 96 | | if (correlationId.Length > MaxCorrelationIdLength) |
| | | 97 | | { |
| | 22 | 98 | | return $"CorrelationId is {correlationId.Length} UTF-16 code units; the portable maximum is {MaxCorrelationI |
| | 22 | 99 | | $"({nameof(AsyncResponseChannelOptions)}.{nameof(MaxCorrelationIdLength)} — the correlation_id column wi |
| | 22 | 100 | | "SQL Server channel). A longer id is truncated or rejected at its first database write."; |
| | | 101 | | } |
| | | 102 | | |
| | 15250 | 103 | | if (correlationId[0] == ' ' || correlationId[^1] == ' ') |
| | | 104 | | { |
| | 53 | 105 | | return "CorrelationId begins or ends with a space. SQL Server pads the shorter operand of an equality compar |
| | 53 | 106 | | "(binary collations included), so an id with surrounding spaces is the SAME key as one without to the da |
| | 53 | 107 | | "while the library compares ids ordinally — a response could reach the wrong waiter. Trim the id."; |
| | | 108 | | } |
| | | 109 | | |
| | 15197 | 110 | | if (PortableText.IndexOfIllFormedUtf16(correlationId) is var illFormed and >= 0) |
| | | 111 | | { |
| | 24 | 112 | | return PortableText.IllFormedUtf16Rejection( |
| | 24 | 113 | | "CorrelationId", |
| | 24 | 114 | | DiagnosticText.EscapedExcerpt(correlationId), |
| | 24 | 115 | | correlationId[illFormed], |
| | 24 | 116 | | illFormed); |
| | | 117 | | } |
| | | 118 | | |
| | 15173 | 119 | | if (PortableText.IndexOfControlCharacter(correlationId) is var control and >= 0) |
| | | 120 | | { |
| | 14 | 121 | | return PortableText.ControlCharacterRejection( |
| | 14 | 122 | | "CorrelationId", |
| | 14 | 123 | | DiagnosticText.EscapedExcerpt(correlationId), |
| | 14 | 124 | | correlationId[control], |
| | 14 | 125 | | control); |
| | | 126 | | } |
| | | 127 | | |
| | 15159 | 128 | | return null; |
| | | 129 | | } |
| | | 130 | | |
| | | 131 | | /// <summary> |
| | | 132 | | /// Guards a RESOLVED per-waiter timeout (explicit, <see cref="DefaultTimeout"/>, or the |
| | | 133 | | /// <see cref="RecoveryStateExpiry"/> fallback) before any subscribe/persist side effect: |
| | | 134 | | /// positive (one rule for every channel — zero and the never-firing -1 ms sentinel included) |
| | | 135 | | /// and under the timer ceiling. |
| | | 136 | | /// </summary> |
| | | 137 | | internal static void EnsureWaiterTimeoutSupported(TimeSpan timeout) |
| | | 138 | | { |
| | 5890 | 139 | | if (timeout <= TimeSpan.Zero || timeout > MaxTimerBackedTimeout) |
| | 20 | 140 | | throw new ArgumentOutOfRangeException( |
| | 20 | 141 | | nameof(timeout), |
| | 20 | 142 | | timeout, |
| | 20 | 143 | | $"Waiter timeout must be positive and at most {MaxTimerBackedTimeout.TotalDays:0.#} days (the .NET timer |
| | 5870 | 144 | | } |
| | | 145 | | |
| | | 146 | | /// <summary> |
| | | 147 | | /// Guards a timer-armed interval knob (<see cref="Task.Delay(TimeSpan)"/>, CTS timers, |
| | | 148 | | /// <c>WaitAsync</c>): positive and under the .NET timer ceiling. An over-ceiling interval |
| | | 149 | | /// otherwise throws at the FIRST arming inside a background loop, where the loop's own |
| | | 150 | | /// retry-delay throws again — killing dispatch instead of surfacing a config error. |
| | | 151 | | /// </summary> |
| | | 152 | | internal static void EnsureTimerBacked(TimeSpan value, string optionsName, string knob) |
| | | 153 | | { |
| | 94116 | 154 | | if (value <= TimeSpan.Zero || value > MaxTimerBackedTimeout) |
| | 184 | 155 | | throw new InvalidOperationException( |
| | 184 | 156 | | $"{optionsName}.{knob} must be positive and at most {MaxTimerBackedTimeout.TotalDays:0.#} days (the .NET |
| | 93932 | 157 | | } |
| | | 158 | | |
| | | 159 | | /// <summary> |
| | | 160 | | /// <see cref="EnsureTimerBacked"/> for knobs where zero legitimately means "skip the wait" |
| | | 161 | | /// (e.g. a startup delay): non-negative and under the .NET timer ceiling. |
| | | 162 | | /// </summary> |
| | | 163 | | internal static void EnsureTimerBackedAllowZero(TimeSpan value, string optionsName, string knob) |
| | | 164 | | { |
| | 24091 | 165 | | if (value < TimeSpan.Zero || value > MaxTimerBackedTimeout) |
| | 14 | 166 | | throw new InvalidOperationException( |
| | 14 | 167 | | $"{optionsName}.{knob} must be non-negative and at most {MaxTimerBackedTimeout.TotalDays:0.#} days (the |
| | 24077 | 168 | | } |
| | | 169 | | |
| | | 170 | | /// <summary> |
| | | 171 | | /// Guards a persisted-TTL/deadline-stamp knob: positive and small enough that the |
| | | 172 | | /// "now + value" stamp every consumer computes can never overflow date arithmetic — a |
| | | 173 | | /// <see cref="TimeSpan.MaxValue"/> here otherwise passes registration and throws only at the |
| | | 174 | | /// first stamp, after side effects (a publisher's overflow lands AFTER its insert, reporting |
| | | 175 | | /// failure for a response that may have been delivered). |
| | | 176 | | /// </summary> |
| | | 177 | | internal static void EnsurePersistedTtl(TimeSpan value, string optionsName, string knob) |
| | | 178 | | { |
| | 32388 | 179 | | if (value <= TimeSpan.Zero || value > MaxPersistenceTtl) |
| | 56 | 180 | | throw new InvalidOperationException( |
| | 56 | 181 | | $"{optionsName}.{knob} must be positive and at most {MaxPersistenceTtl.TotalDays:0} days — " + |
| | 56 | 182 | | "expiry/deadline stamps are computed as \"now + value\", and larger values overflow."); |
| | 32332 | 183 | | } |
| | | 184 | | |
| | | 185 | | /// <summary> |
| | | 186 | | /// Validates the shared channel settings, throwing an actionable |
| | | 187 | | /// <see cref="InvalidOperationException"/> on misconfiguration. Concrete channels call this from |
| | | 188 | | /// their own <c>Validate()</c> so every channel fails fast at registration/startup instead of |
| | | 189 | | /// misbehaving at the first wait (a non-positive expiry silently disables recovery; a bad |
| | | 190 | | /// timer-armed timeout otherwise surfaces only after waiter-registration side effects). |
| | | 191 | | /// </summary> |
| | | 192 | | internal void ValidateShared(string optionsName) |
| | | 193 | | { |
| | | 194 | | static string CeilingRule(string optionsName, string knob) |
| | 24 | 195 | | => $"{optionsName}.{knob} must be positive and at most {MaxTimerBackedTimeout.TotalDays:0.#} days (the .NET |
| | | 196 | | |
| | 5771 | 197 | | if (RecoveryStateExpiry <= TimeSpan.Zero) |
| | 4 | 198 | | throw new InvalidOperationException($"{optionsName}.{nameof(RecoveryStateExpiry)} must be positive."); |
| | 5767 | 199 | | if (RecoveryStateExpiry > MaxPersistenceTtl) |
| | 2 | 200 | | throw new InvalidOperationException( |
| | 2 | 201 | | $"{optionsName}.{nameof(RecoveryStateExpiry)} must be at most {MaxPersistenceTtl.TotalDays:0} days — " + |
| | 2 | 202 | | "expiry stamps are computed as \"now + expiry\", and larger values overflow at the first save."); |
| | | 203 | | |
| | | 204 | | // The ceiling applies to the expiry only in its TIMER-ARMED role — the waiter-timeout |
| | | 205 | | // fallback when DefaultTimeout is not configured. As a pure persistence TTL (with a |
| | | 206 | | // DefaultTimeout set) it may legitimately exceed it: e.g. 90-day recovery retention with |
| | | 207 | | // a 12-hour default timeout. |
| | 5765 | 208 | | if (DefaultTimeout is null && RecoveryStateExpiry > MaxTimerBackedTimeout) |
| | 2 | 209 | | throw new InvalidOperationException( |
| | 2 | 210 | | $"{optionsName}.{nameof(RecoveryStateExpiry)} is the waiter-timeout fallback while {nameof(DefaultTimeou |
| | 2 | 211 | | $"not configured, and must then be at most {MaxTimerBackedTimeout.TotalDays:0.#} days (the .NET timer ce |
| | 2 | 212 | | $"Configure {nameof(DefaultTimeout)} to keep a longer recovery retention."); |
| | | 213 | | |
| | 5763 | 214 | | if (DefaultTimeout is { } defaultTimeout && (defaultTimeout <= TimeSpan.Zero || defaultTimeout > MaxTimerBackedT |
| | 14 | 215 | | throw new InvalidOperationException(CeilingRule(optionsName, nameof(DefaultTimeout))); |
| | 5749 | 216 | | if (DisposalDrainTimeout <= TimeSpan.Zero || DisposalDrainTimeout > MaxTimerBackedTimeout) |
| | 10 | 217 | | throw new InvalidOperationException(CeilingRule(optionsName, nameof(DisposalDrainTimeout))); |
| | 5739 | 218 | | } |
| | | 219 | | } |
| | | 220 | | |
| | | 221 | | /// <summary> |
| | | 222 | | /// Shared options for a <em>durable</em> async-response channel that serializes failures onto the |
| | | 223 | | /// wire (Redis, NATS). Adds the remote stack-trace policy on top of |
| | | 224 | | /// <see cref="AsyncResponseChannelOptions"/>. |
| | | 225 | | /// </summary> |
| | | 226 | | public abstract class DurableAsyncResponseChannelOptions : AsyncResponseChannelOptions |
| | | 227 | | { |
| | | 228 | | /// <summary> |
| | | 229 | | /// Whether a failed response envelope carries the remote exception's stack trace on the wire |
| | | 230 | | /// (surfaced to the waiter via <c>Exception.Data["RemoteStackTrace"]</c>). Stack traces aid |
| | | 231 | | /// debugging but can carry file paths; set to <c>false</c> to omit them. Default: <c>true</c>. |
| | | 232 | | /// </summary> |
| | | 233 | | public bool IncludeRemoteStackTrace { get; set; } = true; |
| | | 234 | | |
| | | 235 | | /// <summary> |
| | | 236 | | /// Maximum length, in characters, of a remote stack trace placed on the wire and restored on the |
| | | 237 | | /// waiter side; longer traces are truncated with a marker. Bounds what a buggy or hostile remote |
| | | 238 | | /// can push into logs (a multi-megabyte trace). Applied on both publish and receive. Default: 16384. |
| | | 239 | | /// </summary> |
| | | 240 | | public int MaxRemoteStackTraceLength { get; set; } = 16 * 1024; |
| | | 241 | | } |