| | | 1 | | using System.Collections.Concurrent; |
| | | 2 | | using System.Diagnostics.CodeAnalysis; |
| | | 3 | | using System.Text.Json; |
| | | 4 | | |
| | | 5 | | namespace AsyncResponse; |
| | | 6 | | |
| | | 7 | | /// <summary> |
| | | 8 | | /// Outcome of classifying a lost-subscriber response payload. |
| | | 9 | | /// </summary> |
| | | 10 | | /// <param name="Action"> |
| | | 11 | | /// The recovery route the payload chose (<see cref="IAsyncResponsePayload.OnRecovery"/>), |
| | | 12 | | /// or <c>null</c> when it could not be classified — a <c>null</c> payload, missing/unresolvable |
| | | 13 | | /// type information, or a conversion failure. Callers must treat <c>null</c> conservatively as |
| | | 14 | | /// "do not resume", so a payload that cannot be understood never takes the happy path. |
| | | 15 | | /// </param> |
| | | 16 | | /// <param name="MaterializedPayload"> |
| | | 17 | | /// The payload as an <see cref="IAsyncResponsePayload"/> instance of the REGISTERED type, |
| | | 18 | | /// materialized from the payload's wire representation — never the publisher's live instance, so |
| | | 19 | | /// the verdict (and what callbacks see) can never depend on a serialization boundary. |
| | | 20 | | /// <c>null</c> exactly when <paramref name="Action"/> is <c>null</c>. Callbacks must receive THIS |
| | | 21 | | /// instance, never the raw JSON: an <c>object</c>-/interface-/base-typed callback parameter |
| | | 22 | | /// otherwise gets a <see cref="JsonElement"/> and every type guard in the consuming flow silently |
| | | 23 | | /// fails (the 292332 flow-deadlock incident). |
| | | 24 | | /// </param> |
| | 742 | 25 | | internal readonly record struct RecoveryClassification(RecoveryAction? Action, object? MaterializedPayload); |
| | | 26 | | |
| | | 27 | | /// <summary> |
| | | 28 | | /// Resolves the lost-subscriber recovery route for a response payload by materializing its wire |
| | | 29 | | /// representation as the registered payload type and asking |
| | | 30 | | /// <see cref="IAsyncResponsePayload.OnRecovery"/>. |
| | | 31 | | /// <para> |
| | | 32 | | /// The input is always a wire representation — raw broker JSON, or a typed publish normalized to |
| | | 33 | | /// its declared-type serialization by the dispatcher — so in-process and broker deliveries of the |
| | | 34 | | /// same response classify identically. The materialized instance is returned so the chosen |
| | | 35 | | /// callback receives it instead of the raw JSON. |
| | | 36 | | /// </para> |
| | | 37 | | /// </summary> |
| | | 38 | | internal static class PayloadRecoveryClassifier |
| | | 39 | | { |
| | | 40 | | // Entries carry the resolver-registry generation observed before the scan that produced them, |
| | | 41 | | // mirroring the negative cache's stamp: a plain clear-on-unregister has a race — an in-flight |
| | | 42 | | // resolution that got its answer from the departing resolver can insert AFTER the clear, |
| | | 43 | | // permanently re-poisoning the name with the revoked type. A stale stamp makes the entry a |
| | | 44 | | // non-hit, so the next lookup rescans against the current resolver set. |
| | | 45 | | private static readonly ConcurrentDictionary<string, (Type Type, int Generation)> PayloadTypes = new(StringComparer. |
| | | 46 | | private static int _resolvedPayloadTypeGeneration; |
| | | 47 | | |
| | | 48 | | /// <summary> |
| | | 49 | | /// Drops resolved payload types when a type resolver is unregistered. Counterpart to |
| | | 50 | | /// <see cref="ReflectionExtensions.InvalidateResolvedServiceTypes"/>: a payload name the |
| | | 51 | | /// departing resolver already answered would otherwise keep materializing into its old type, |
| | | 52 | | /// and keep that type's assembly reachable through this cache. |
| | | 53 | | /// </summary> |
| | | 54 | | internal static void InvalidateResolvedPayloadTypes() |
| | | 55 | | { |
| | | 56 | | // Bump BEFORE clearing: the bump is what fences in-flight scans (their pre-scan stamp goes |
| | | 57 | | // stale); the clear just reclaims memory. |
| | | 58 | | Interlocked.Increment(ref _resolvedPayloadTypeGeneration); |
| | | 59 | | PayloadTypes.Clear(); |
| | | 60 | | } |
| | | 61 | | |
| | | 62 | | /// <summary> |
| | | 63 | | /// Attempts to classify <paramref name="payload"/> for the lost-subscriber path: which route it |
| | | 64 | | /// takes, and the materialized instance the route's callback must receive. |
| | | 65 | | /// </summary> |
| | | 66 | | /// <param name="payload"> |
| | | 67 | | /// The payload as received by <c>SetResponse</c>: either an already-typed |
| | | 68 | | /// <see cref="IAsyncResponsePayload"/>, or raw JSON (<see cref="JsonElement"/> / JSON string) |
| | | 69 | | /// when the response came through a broker ingress. |
| | | 70 | | /// </param> |
| | | 71 | | /// <param name="payloadTypeFullName"> |
| | | 72 | | /// Full name of the payload type the waiter subscribed for, from the recovery state. |
| | | 73 | | /// </param> |
| | | 74 | | public static RecoveryClassification Classify(object? payload, string? payloadTypeFullName) |
| | | 75 | | { |
| | | 76 | | try |
| | | 77 | | { |
| | | 78 | | if (payload is null || string.IsNullOrWhiteSpace(payloadTypeFullName)) |
| | | 79 | | { |
| | | 80 | | return new RecoveryClassification(null, null); |
| | | 81 | | } |
| | | 82 | | |
| | | 83 | | // Wire-only: classification never consults a live CLR instance. A non-JSON payload |
| | | 84 | | // here means no wire representation exists for it (the dispatcher's serialization |
| | | 85 | | // failed) — conservatively unclassifiable rather than letting in-process state that |
| | | 86 | | // never crosses the wire decide the route. |
| | | 87 | | if (payload is not (JsonElement or string)) |
| | | 88 | | { |
| | | 89 | | return new RecoveryClassification(null, null); |
| | | 90 | | } |
| | | 91 | | |
| | | 92 | | // The REGISTRATION's payload type governs, never the publisher's runtime type: |
| | | 93 | | // multiple registrations may share one correlation id with different payload types |
| | | 94 | | // (shared-correlation recovery), and each must be classified as the type IT |
| | | 95 | | // registered. The payload arrives as its WIRE representation (the dispatcher |
| | | 96 | | // normalizes typed publishes to their declared-type serialization), so materializing |
| | | 97 | | // it here yields exactly what a broker delivery would have produced — polymorphic |
| | | 98 | | // discriminators included, in-process-only ([JsonIgnore]) state excluded. |
| | | 99 | | var registeredType = ResolvePayloadType(payloadTypeFullName!); |
| | | 100 | | if (registeredType is null || !typeof(IAsyncResponsePayload).IsAssignableFrom(registeredType)) |
| | | 101 | | { |
| | | 102 | | return new RecoveryClassification(null, null); |
| | | 103 | | } |
| | | 104 | | |
| | | 105 | | return payload.ConvertTo(registeredType) is IAsyncResponsePayload materialized |
| | | 106 | | ? new RecoveryClassification(materialized.OnRecovery(), materialized) |
| | | 107 | | : new RecoveryClassification(null, null); |
| | | 108 | | } |
| | | 109 | | catch |
| | | 110 | | { |
| | | 111 | | // A payload that cannot be materialized as the registered type (or whose classifier |
| | | 112 | | // throws) carries no usable domain state; treat it conservatively as "do not resume". |
| | | 113 | | // The failure route then carries the raw payload, exactly as before materialization |
| | | 114 | | // existed, and the diagnostics on the resolution path surface the cause. |
| | | 115 | | return new RecoveryClassification(null, null); |
| | | 116 | | } |
| | | 117 | | } |
| | | 118 | | |
| | | 119 | | [UnconditionalSuppressMessage("Trimming", "IL2026", |
| | | 120 | | Justification = "The persisted payload type name comes from a recoverable-waiter registration whose payload type |
| | | 121 | | "parameter is statically referenced by the registering app; an unresolvable name is answered wit |
| | | 122 | | "null — the conservative 'do not resume' route — plus a type-resolution-failure diagnostic.")] |
| | | 123 | | internal static Type? ResolvePayloadType(string payloadTypeFullName) |
| | | 124 | | { |
| | | 125 | | // Before any cache or the parser, exactly as ResolveServiceType does and for the same |
| | | 126 | | // reason — and with more at stake here: this name is resolved BEFORE a callback is chosen, |
| | | 127 | | // so no callback authorizer ever stands between a recovery row and this line. |
| | | 128 | | if (!AsyncResponseTypeResolution.IsWithinResolutionLimits(payloadTypeFullName)) |
| | | 129 | | { |
| | | 130 | | AsyncResponseDiagnostics.RecordTypeResolutionFailure("payload"); |
| | | 131 | | return null; |
| | | 132 | | } |
| | | 133 | | |
| | | 134 | | // Must precede any cache consult/populate: a miss cached without the invalidation hook |
| | | 135 | | // active could outlive a later assembly load that makes the name resolvable. |
| | | 136 | | UnresolvableTypeNames.EnsureAssemblyLoadInvalidation(); |
| | | 137 | | |
| | | 138 | | if (PayloadTypes.TryGetValue(payloadTypeFullName, out var cached) |
| | | 139 | | && cached.Generation == Volatile.Read(ref _resolvedPayloadTypeGeneration)) |
| | | 140 | | { |
| | | 141 | | return cached.Type; |
| | | 142 | | } |
| | | 143 | | |
| | | 144 | | // Fail fast on a name that already failed a full scan (shared with the callback service |
| | | 145 | | // resolver): without this, every redelivery naming an unresolvable payload type (a |
| | | 146 | | // poisoned recovery row, a renamed class) re-walks every loaded assembly. Only a |
| | | 147 | | // CURRENT-generation entry counts — a stale stamp means the miss may have raced a |
| | | 148 | | // resolver registration or assembly load, so it rescans. The diagnostic still fires per |
| | | 149 | | // attempt, so a poisoned name stays visible to operators while costing a dictionary hit. |
| | | 150 | | if (UnresolvableTypeNames.IsKnownMiss(payloadTypeFullName)) |
| | | 151 | | { |
| | | 152 | | AsyncResponseDiagnostics.RecordTypeResolutionFailure("payload"); |
| | | 153 | | return null; |
| | | 154 | | } |
| | | 155 | | |
| | | 156 | | var generationBeforeScan = UnresolvableTypeNames.GenerationBeforeScan(); |
| | | 157 | | var resolvedGenerationBeforeScan = Volatile.Read(ref _resolvedPayloadTypeGeneration); |
| | | 158 | | |
| | | 159 | | // Loaded assemblies only — every component of the name, generic arguments included (see |
| | | 160 | | // ResolveLoaded): a persisted name must never make the process load an assembly. |
| | | 161 | | var resolved = AsyncResponseTypeResolution.ResolveLoaded(payloadTypeFullName); |
| | | 162 | | |
| | | 163 | | // Opt-in fallback for payload types loaded into a non-default AssemblyLoadContext (plugins). |
| | | 164 | | resolved ??= AsyncResponseTypeResolution.Resolve(payloadTypeFullName); |
| | | 165 | | |
| | | 166 | | if (resolved is not null) |
| | | 167 | | { |
| | | 168 | | // Collectible (plugin) payload types stay resolve-per-call: a strong process-wide |
| | | 169 | | // cache entry would pin the plugin's AssemblyLoadContext after unload. Indexer, not |
| | | 170 | | // TryAdd, so a stale-stamped survivor of a raced unregister is replaced. |
| | | 171 | | if (!resolved.Assembly.IsCollectible) |
| | | 172 | | PayloadTypes[payloadTypeFullName] = (resolved, resolvedGenerationBeforeScan); |
| | | 173 | | } |
| | | 174 | | else |
| | | 175 | | { |
| | | 176 | | UnresolvableTypeNames.RecordMiss(payloadTypeFullName, generationBeforeScan); |
| | | 177 | | |
| | | 178 | | // Surface the silent "couldn't materialize the payload type" path so operators can |
| | | 179 | | // correlate a recovery that routed to failure with a missing/ALC-loaded type. |
| | | 180 | | AsyncResponseDiagnostics.RecordTypeResolutionFailure("payload"); |
| | | 181 | | } |
| | | 182 | | |
| | | 183 | | return resolved; |
| | | 184 | | } |
| | | 185 | | } |