diff --git a/src/MeshWeaver.Connection.Orleans/OrleansRoutingService.cs b/src/MeshWeaver.Connection.Orleans/OrleansRoutingService.cs index 6979f34398..7c48791c8b 100644 --- a/src/MeshWeaver.Connection.Orleans/OrleansRoutingService.cs +++ b/src/MeshWeaver.Connection.Orleans/OrleansRoutingService.cs @@ -116,6 +116,27 @@ public class OrleansRoutingService : IRoutingService, IDisposable /// private bool IsHostStopping => hostStopping.IsCancellationRequested; + /// + /// Can THIS process host an Orleans grain? A silo can; a cluster CLIENT cannot, and neither + /// can a host with no Orleans at all. + /// + /// Derived from Orleans' own composition, not from a flag of ours. + /// is registered by DefaultSiloServices and by nothing + /// else โ€” a client's container has none โ€” so its presence IS the question, answered by the + /// framework that decides it. + /// + /// ๐Ÿšจ What it governs, and what it must never govern. It selects the pod-hub + /// claim's TERMINAL and the level of the line that reports one โ€” see + /// . It does NOT gate whether the claim is attempted: a routing + /// service built on a bare container (several fixtures do exactly that) must still make the + /// call, or the gate that stops a SHUTTING-DOWN silo from claiming would be indistinguishable + /// from a gate that stopped claiming altogether. + /// + /// Settable as a test seam, exactly like : instance state, + /// never static, so a unit test can pin the silo policy without standing up a silo. + /// + internal bool CanHostGrains { get; set; } + /// /// ๐Ÿšจ THE ONLY DOOR to an Orleans grain from this class โ€” and therefore the one place /// where "this host has begun stopping" turns into "do not ask Orleans for an activation". @@ -194,6 +215,10 @@ public OrleansRoutingService( // stays uncancelled and every routing decision below behaves exactly as before. hostStopping = serviceProvider.GetService()?.ApplicationStopping ?? CancellationToken.None; + // Orleans registers ILocalSiloDetails in DefaultSiloServices only, so resolving it here is + // the framework's own answer to "is this process a silo" โ€” see CanHostGrains. Optional for + // the same reason as the two above: a bare mesh in a unit test is neither silo nor client. + CanHostGrains = serviceProvider.GetService() is not null; } /// @@ -825,23 +850,72 @@ private void EnqueueStreamTeardown( } /// - /// How many times an that landed on the WRONG silo is retried. - /// This is not a poll and not a health retry: false means one specific, transient thing โ€” - /// the previous owner's activation has not gone yet, which happens exactly while an address - /// MOVES between pods. Bounded, because "the owner is not a silo at all" (an Orleans client - /// process, which cannot host a grain) must terminate rather than spin: that hub keeps the - /// stream, permanently and correctly. + /// The pod-hub claim's INITIAL budget โ€” the point at which a claim that has not landed stops + /// being "the address is still moving between pods" and becomes reportable. + /// + /// ๐Ÿšจ It is not a give-up. On a process that can host + /// grains the claim keeps retrying with the same capped + /// backoff past this point; exhausting the budget only + /// buys the one Warning line that names the address. See for + /// the terminals, and Doc/Architecture/DurableStreamsViaMeshNodes for why a bounded + /// claim was #1742's open residual. + /// + internal const int PodHubAttachRetries = 5; + + /// + /// Backoff between pod-hub claim attempts: 100 ms doubling to a 2 s ceiling. The attempt index + /// is CLAMPED at by the caller, so a claim that keeps + /// retrying settles at the ceiling instead of growing without bound. /// - private const int PodHubAttachRetries = 5; + /// Zero-based index of the attempt that just failed. + /// How long to wait before the next attempt. + internal static TimeSpan PodHubClaimBackoff(int attempt) => + TimeSpan.FromMilliseconds(Math.Min(100 * Math.Pow(2, attempt), 2_000)); + + /// + /// Test seam (InternalsVisibleTo): the backoff the pod-hub claim retry uses. Instance + /// state, never static โ€” the POLICY (indefinite on a silo, bounded where a grain can never be + /// hosted) is what a test pins, without a wall clock. Production keeps + /// . + /// + internal Func ClaimBackoff { get; set; } = PodHubClaimBackoff; /// /// Claims for THIS process, so the rest of the cluster can deliver to /// it with a directed grain call instead of a stream publish (#1742). /// /// Synchronous to the caller and best-effort by construction: RegisterStream's local - /// route is already live, and a claim that never lands simply leaves this hub on the stream โ€” - /// the transport it has always used. So a failure here degrades, it never blocks; the returned - /// disposable releases the claim. + /// route is already live, and a claim that has not landed yet simply leaves this hub on the + /// stream โ€” the transport it has always used. So a failure here degrades, it never blocks; the + /// returned disposable releases the claim. + /// + /// ๐Ÿšจ The claim's lifetime is DERIVED, never a counter โ€” the #2426 rule, applied + /// here because a bounded claim was #1742's stated open residual: six attempts over โ‰ˆ3 s and + /// then a Debug give-up, after which a SILO-hosted hub kept the stream transport for the + /// whole life of the process, invisibly. On a process that can host + /// grains the claim now retries with its capped backoff until one of exactly two + /// terminals, both of which are real events rather than budgets: + /// + /// the hub's registration is disposed โ€” the returned disposable, which is also what + /// reaches through ; + /// fires โ€” expressed by + /// inside the Defer, so every re-subscribe + /// re-asks and a claim still bouncing when shutdown begins stops asking rather than placing + /// an activation on the silo that is leaving. + /// + /// + /// ๐Ÿšจ The third terminal is IMPOSSIBILITY, and it is derived too. A process that + /// cannot host a grain can never win this claim โ€” PodHubGrain.Attach is + /// [PreferLocalPlacement], so from a client it lands on some silo which has no local + /// route and answers false, for ever. There the initial budget IS the end, exactly as + /// before, and the give-up stays at Debug because it is the expected permanent outcome. + /// Retrying it indefinitely would not be a derived lifetime, it would be a poll that cannot + /// converge โ€” and a measurable one: each attempt makes the SILO log + /// [POD-HUB] Attach โ€ฆ landed on a silo that has no local route at + /// Information, i.e. one line per hub per backoff interval, which is the log-storm shape + /// #2426/#2546 exist to remove. + /// + /// See Doc/Architecture/DurableStreamsViaMeshNodes. /// private IDisposable AttachPodHub(Address address) { @@ -855,6 +929,8 @@ private IDisposable AttachPodHub(Address address) return Disposable.Empty; var addressPath = address.ToString(); + var attempt = 0; + var reported = false; var attach = new SingleAssignmentDisposable(); inFlight.Add(attach); attach.Disposable = Observable @@ -882,25 +958,67 @@ private IDisposable AttachPodHub(Address address) .SelectMany(claimed => claimed ? Observable.Return(true) : Observable.Throw(new PodHubNotHereException(addressPath))) - .RetryWhen(errors => errors - .Select((ex, i) => (Exception: ex, Attempt: i)) - .SelectMany(t => t.Attempt >= PodHubAttachRetries - ? Observable.Throw(t.Exception) - : Observable.Timer( - TimeSpan.FromMilliseconds(Math.Min(100 * Math.Pow(2, t.Attempt), 2_000)), - Scheduler.Default))) + // The claim's own state, per registration: how many attempts have failed (CLAMPED at + // the initial budget, so an indefinite claim settles at the backoff ceiling and the + // counter can never overflow) and whether the budget-exhausted line has been reported. + // Locals in this closure, never fields and never static โ€” one claim, one lifetime. The + // error sequence RetryWhen hands us is serialised (one error, then its signal, then + // the next), so plain reads and writes are correct here. + .RetryWhen(errors => errors.SelectMany(ex => + { + if (attempt >= PodHubAttachRetries) + { + // ๐Ÿšจ THE TERMINAL, and the only one that is a decision rather than an event: a + // process that cannot host a grain can never win this claim, so the budget IS + // the end there. Everywhere else the budget only buys the line below. + if (!CanHostGrains) + return Observable.Throw(ex); + if (!reported) + { + reported = true; + OrleansRouteTrace.Write( + $"OrleansRoutingService.AttachPodHub BUDGET_EXHAUSTED addr={addressPath} ex={ex.Message}"); + // ๐Ÿšจ Warning, ONCE, naming the hub โ€” the signal #1742 was missing. A silo + // that cannot claim its own hub is abnormal: until the claim lands that hub + // is reachable only over the stream, which is the transport this design + // retires. The claim keeps trying, so this line is "still trying", not + // "gave up" โ€” and its resolution is logged below. + logger.LogWarning(ex, + "Pod-hub claim for {Address} did not land within its initial budget of " + + "{Attempts} attempts. This process CAN host grains, so a hub that cannot " + + "claim its own address is abnormal โ€” until the claim lands, the cluster " + + "can only reach it over the Orleans stream. The claim keeps retrying with " + + "a capped backoff until this hub's registration is disposed or the host " + + "stops; there is no give-up.", + addressPath, PodHubAttachRetries); + } + } + + var wait = ClaimBackoff(attempt); + attempt = Math.Min(attempt + 1, PodHubAttachRetries); + return Observable.Timer(wait, Scheduler.Default); + })) .Subscribe( _ => { OrleansRouteTrace.Write($"OrleansRoutingService.AttachPodHub OK addr={addressPath}"); - logger.LogDebug("Pod-hub claim for {Address} landed on this process", addressPath); + if (reported) + // The close of the Warning above โ€” without it a Loki reader cannot tell an + // address that recovered from one that is still stranded. + logger.LogInformation( + "Pod-hub claim for {Address} landed after its initial budget was exhausted โ€” " + + "the cluster reaches this hub by directed grain call again.", + addressPath); + else + logger.LogDebug("Pod-hub claim for {Address} landed on this process", addressPath); }, ex => { - // Debug, not Warning: on an Orleans CLIENT this is the expected, permanent - // outcome for every hub, and a per-hub warning there would be pure noise. The - // signal that matters lives on the ROUTER side โ€” the fallback line โ€” because - // that is where "a silo-hosted hub was not reachable by call" is observable. + // Reached only on the impossibility terminal above (or if the retry signal + // itself faults). Debug, not Warning: on an Orleans CLIENT this is the expected, + // permanent outcome for every hub, and a per-hub warning there would be pure + // noise โ€” which is exactly why the level is derived from what this process can + // host rather than from how many attempts were spent. OrleansRouteTrace.Write($"OrleansRoutingService.AttachPodHub GAVE_UP addr={addressPath} ex={ex.Message}"); logger.LogDebug(ex, "Pod-hub claim for {Address} did not land in this process โ€” it keeps the Orleans " diff --git a/src/MeshWeaver.Data/DataExtensions.cs b/src/MeshWeaver.Data/DataExtensions.cs index e681253c4a..6d26e31a47 100644 --- a/src/MeshWeaver.Data/DataExtensions.cs +++ b/src/MeshWeaver.Data/DataExtensions.cs @@ -1830,13 +1830,25 @@ private static IMessageDelivery HandleGetDomainTypesRequest(IMessageHub hub, IMe /// names why nothing else ever ends it: "only an UnsubscribeRequest disposes a server-side /// stream", and a dead process sends none. /// - /// The verdict gate is the STAMP, never the alone: a LIVE hub - /// also answers NotFound (an unhandled request), and evicting on that would tear down a - /// healthy subscriber's stream. Only the router โ€” the one component that asked the - /// cluster-wide subscription registry โ€” stamps TargetUnserved, so absence of the stamp - /// safely means "not ours to act on" and the delivery passes through unchanged for whoever - /// else handles it (the request/response callback already ran; it is first in the rule - /// chain). + /// The verdict gate is the STAMP, and ONLY the stamp โ€” never the . + /// A LIVE hub also answers NotFound (an unhandled request), which is why the ErrorType alone + /// could never be the gate; but only the router โ€” the one component that asked the cluster, + /// through the stream's subscription registry or through Orleans' own grain directory โ€” stamps + /// TargetUnserved, so absence of the stamp safely means "not ours to act on" and the + /// delivery passes through unchanged for whoever else handles it (the request/response callback + /// already ran; it is first in the rule chain). + /// + /// ๐Ÿšจ The ErrorType == NotFound half of this gate was removed deliberately. + /// It was redundant belt-and-braces from the era when the router's ONLY unserved verdict came + /// from the stream leg's subscriber probe. Since + /// RoutingGrain.AnswerPodHubNotHere the router also stamps the verdict on the + /// TRANSIENT โ€” "no silo serves this hub right now" reached + /// in one hop through the grain directory, the answer that replaced the stream publish + /// (Doc/Architecture/DurableStreamsViaMeshNodes). Keeping the NotFound test would have + /// made that answer inert here and silently re-opened this very leak for every dead circuit in + /// the fleet. The two verdicts are complementary, not contradictory: the SUBSCRIBER rides + /// ShuttingDown out and re-asks, while the OWNER drops the server-side half it can no + /// longer push to โ€” which is exactly the recovery described below. /// /// Evict-only, and terminal per NACK: nothing here retries, re-probes or resubscribes. A /// subscriber that was in fact alive but unreachable loses only its server-side stream, and @@ -1847,7 +1859,7 @@ private static IMessageDelivery HandleTargetUnservedFailure( IMessageHub hub, IMessageDelivery delivery) { var failure = delivery.Message; - if (!failure.TargetUnserved || failure.ErrorType != ErrorType.NotFound) + if (!failure.TargetUnserved) return delivery; var deadSubscriber = failure.Delivery?.Target; if (deadSubscriber is null || hub.GetWorkspace() is not Workspace workspace) diff --git a/src/MeshWeaver.Documentation/Data/Architecture/DurableStreamsViaMeshNodes.md b/src/MeshWeaver.Documentation/Data/Architecture/DurableStreamsViaMeshNodes.md index b17b41c731..f5a6a99ed6 100644 --- a/src/MeshWeaver.Documentation/Data/Architecture/DurableStreamsViaMeshNodes.md +++ b/src/MeshWeaver.Documentation/Data/Architecture/DurableStreamsViaMeshNodes.md @@ -93,31 +93,85 @@ The design is the roll plan's **release N+2**, made precise: probe already produces for "no silo serves this hub" (#1742), now reached in one hop. `SynchronizationStream` and `MeshNodeStreamCache` ride it out; a requester gets its answer inside the directed call's own budget instead of the stream's. -- **The owner's claim gets a DERIVED lifetime.** `OrleansRoutingService.AttachPodHub` today makes - six attempts over โ‰ˆ3 s and then gives up at `Debug` โ€” after which a silo-hosted hub keeps the - stream *forever*, invisibly (the only signal is the router-side fallback line). The claim - instead retries with its capped backoff **until one of two real terminals**: the hub's +- **The owner's claim gets a DERIVED lifetime.** `OrleansRoutingService.AttachPodHub` used to make + six attempts over โ‰ˆ3 s and then give up at `Debug` โ€” after which a silo-hosted hub kept the + stream *forever*, invisibly (the only signal was the router-side fallback line). The claim + now retries with its capped backoff **until one of two real terminals**: the hub's registration is disposed, or `IHostApplicationLifetime.ApplicationStopping` fires (the gate - `GrainWhileRunning` already expresses). Once the initial budget is exhausted **on a silo** the - give-up line is `Warning` โ€” a silo that cannot claim its own hub is abnormal, and the fleet has + `GrainWhileRunning` already expresses). Once the initial budget is exhausted **on a process that + can host grains** the line is `Warning` naming the hub โ€” abnormal, and the fleet has no clients for which it would be noise. This is the #2426 rule applied to the claim: no cleanup message a restarting process would never send, only lifetimes derived from the hub and the host. + - **A third terminal, and it is derived too: IMPOSSIBILITY.** A process that cannot host a grain + can never win the claim โ€” `PodHubGrain` is `[PreferLocalPlacement]`, so from a cluster client + the activation lands on some silo with no local route and answers `false`, for ever. There the + initial budget *is* the end and the give-up stays at `Debug`, because that is the expected + permanent outcome. Retrying it would not be a lifetime, it would be a poll that cannot + converge โ€” and a measurable one: every attempt makes the *silo* log `[POD-HUB] Attach โ€ฆ landed + on a silo that has no local route` at `Information`, i.e. one line per hub per backoff + interval, which is the storm shape #2426/#2546 exist to remove. The discriminator is Orleans' + own: `ILocalSiloDetails` is registered by `DefaultSiloServices` and by nothing else. - **The stream stays only for CLIENT-hosted address types, by declaration.** The fallback is gated - on the address type being declared client-hosted (`MeshBuilder.AddClientHostedAddressType`, - which the Orleans test fixtures call for `client`), **never** on the grain answering "not here". + on the address type being declared client-hosted (`MeshBuilder.AddClientHostedAddressType`), + **never** on the grain answering "not here". In production no address type is declared client-hosted, so the router never publishes. The - test rig keeps its stream until the rig hosts its hubs on a silo, at which point - `AddMemoryStreams` and `PubSubStore` go with it. -- **`StreamMessageSizeGuard` retargets**, as the roll plan already says: the wall is Orleans' - `MaxMessageBodySize` on the directed call, and crossing it throws, which is the outcome the guard - exists to produce. - -**The gate for shipping this is unchanged from the roll plan**: a full rolling deploy with none of -`[ROUTE] Pod-hub grain for {Address} is not attached โ€” falling back to the stream publish` in -Loki. A line *during* a roll is the plan working; a line *after* the roll has settled is a hub -whose owner never claimed โ€” which the indefinite claim above turns from "invisible at Debug" into -a `Warning` naming the hub. + Orleans test rig declares all four built-in stream-routed types โ€” it hosts a hub of each on its + cluster client (`client/{id}` from `GetClient`, the client host's own root `mesh/{guid}`, + `portal/{guid}` in the documentation/graph/markdown tests, and the client's `cache` hub) โ€” and + keeps its stream until the rig hosts its hubs on a silo, at which point `AddMemoryStreams` and + `PubSubStore` go with it. +- **The verdict is `ShuttingDown` + `TargetUnserved`, and the owner-side eviction had to be + re-gated to see it.** `DataExtensions.HandleTargetUnservedFailure` โ€” the #2426/#2546 fix that + stops an owner fanning changes out to a dead subscriber forever โ€” required + `TargetUnserved && ErrorType == NotFound`. That second test was redundant belt-and-braces from + the era when the *only* producer of the stamp was the stream leg's subscriber probe; left in + place it would have made the new one-hop verdict **inert**, silently re-opening the leak for + every dead circuit in the fleet. The gate is now the STAMP alone, which is what + `DeliveryFailure.TargetUnserved`'s own contract always said ("only the router โ€ฆ may stamp + this"). The two facts are complementary rather than contradictory: the **subscriber** rides + `ShuttingDown` out and re-asks, while the **owner** drops the server-side half it can no longer + push to. +- **`StreamMessageSizeGuard` (#1890) needs no code change, and did not get one.** The guard exists + to turn a *silent* drop into a loud, NACK'd refusal: an oversized payload on the memory stream + succeeds at the publish and dies inside `PersistentStreamPullingAgent`'s non-convergent retry + loop, naming only a queue id. On the directed call the wall is Orleans' `MaxMessageBodySize` and + crossing it **throws**, which `BuildPodHubRoute`'s `TerminalCallFailure` already turns into a + classified `DeliveryFailure` naming the address, the delivery id and the sender โ€” the outcome + the guard exists to produce. Retargeting the constant itself (plumbing `SiloMessagingOptions` + into `RoutingGrain` plus its own refusal shape) is a separate change and is *not* required for + correctness here; the guard stays on `PostToStream`, which after this slice is reachable in + production for nothing at all. + +### The N+2 gate, and why these two slices shipped without waiting for it + +**The roll plan's gate was**: a full rolling deploy with none of `[ROUTE] Pod-hub grain for +{Address} is not attached โ€” falling back to the stream publish` in Loki. That gate measured a +*risk that the two slices above jointly remove*, so it was satisfied by construction rather than +by observation: + +- The gate's worry is the window "the owner exists, but its claim has not landed" โ€” in which + removing the fallback would drop a delivery to a live hub. **Slice 2 does not drop it: it + ANSWERS it**, with a transient NACK inside the directed call's own budget. Every consumer on + that path already has recovery machinery armed for exactly this verdict + (`SynchronizationStream`'s resubscribe latch, `MeshNodeStreamCache.IsTransientOwnerFailure`, + `MeshNodeStreamExtensions`' paced retry), which is why the verdict is `ShuttingDown` and not + `NotFound`. The pre-change alternative was strictly worse for the same window: a publish that + *succeeds and discards* when nobody is subscribed, or stalls 30โ€“60 s on a wedged queue grain โ€” + the failure with **no** signal at all. +- **Slice 1 closes the window rather than surviving it.** The claim now retries until it lands, so + "owner exists, claim not landed" resolves by retry instead of persisting for the life of the + process. Before slice 1 it could persist forever, which is precisely why the gate was needed. +- And the gate's own instrument was unreliable in the direction that matters: the line it counts + is `Information`, emitted per delivery, and **its absence was never proof** (the page said so). + A `Warning` naming the hub, emitted once per claim that has not landed, is a strictly better + instrument โ€” and it is what slice 1 adds. The fallback line survives, but after slice 2 only a + DECLARED client-hosted type can reach it, so in the fleet it cannot be emitted at all. + +Residual risk, stated plainly: an address type that is genuinely client-hosted in some deployment +and was never declared would go from "works over the stream" to "NACK'd transiently". No such +deployment exists โ€” `UseOrleansMeshClient` has only test-fixture callers โ€” and the symptom would +be loud (a windowed `Warning` naming the address) rather than silent. ## 3 ยท Cross-silo change notifications โ€” the storage layer's feed is the durable channel @@ -202,22 +256,30 @@ checked against it: | classification of lifecycle faults as transient | **landed** โ€” #2518, #2645, #2647 | row 1 needs nothing else | | this design | **this page** | records the direction on #1742, #2320, #2322, #2406 | | stale "listener never started" comments corrected | **landed with this page** | see below | -| pod-hub claim: indefinite, derived lifetime, `Warning` on a silo | next | core only; closes the #1742 residual *"a claim that fails to land degrades silently"* | +| pod-hub claim: indefinite, derived lifetime, `Warning` where grains can be hosted | **landed** โ€” #2745 | closed the #1742 residual *"a claim that fails to land degrades silently"*; core only | +| routing: transient NACK on `PodHubNotHere`; fallback gated on declared client-hosted types | **landed** โ€” #2745 | closed #2320, #2322, #2406 as *made unreachable*. Shipped with slice 1 rather than after a clean roll โ€” see *The N+2 gate* above for why the two together satisfy it by construction | +| owner-side eviction re-gated on the `TargetUnserved` STAMP alone | **landed with the slice above** | required by it: gating on `NotFound` would have made the new verdict inert and re-opened #2426/#2546 | | `DataChangeNotification.NodeType/Version` + `StorageChangeFeedRelay` | next | **core first**; the relay must tolerate a payload without the new fields | | `notify_mesh_node_changes()` emits `nodeType`, `version` | after the core slice | PostgreSql adapter (MeshWeaver.Plugins); a schema-initializer revision, re-applied by the existing DROP-then-CREATE | | delete `OrleansMeshChangeFeed` broadcast + `PathCacheInvalidatorGrain` | after both | the memory stream then carries routing fallback only | -| routing: transient NACK on `PodHubNotHere`; fallback gated on declared client-hosted types | after a clean roll (the Loki gate above) | closes #2320, #2322, #2406 as *made unreachable* | +| `StreamMessageSizeGuard` retarget onto `MaxMessageBodySize` | optional, not blocking | the directed call already THROWS at that wall, which is the outcome the guard produces โ€” see the bullet above | | test rig hosts hubs on a silo โ†’ `AddMemoryStreams` deleted | last | the only remaining user | ## How to check a live cluster ``` -# the N+2 gate โ€” must be EMPTY across a full rolling deploy once the fleet has settled +# ๐Ÿšจ Must be EMPTY, always, not merely after a roll: only an address type DECLARED client-hosted +# can reach this line, and production declares none. A hit means somebody added a declaration. {namespace="memex-cloud"} |= "falling back to the stream publish" -# a silo-hosted hub whose claim never landed (after the claim slice lands: Warning, one per hub) +# THE instrument now โ€” a hub whose claim has not landed, once per claim, naming the address. +# The claim keeps retrying, so a matching "landed after its initial budget was exhausted" line +# for the same address is the resolution; one without it is a hub still on no transport at all. {namespace="memex-cloud"} |= "Pod-hub claim for" |= "did not land" +# the transient NACK that replaced the publish โ€” windowed to one Warning per address per 60 s +{namespace="memex-cloud"} |= "was refused: no silo in this cluster is currently serving that hub" + # the database feed is up on every pod โ€” one line per pod per LISTEN (re)connect {namespace="memex-cloud"} |= "PostgreSQL LISTEN started on mesh_node_changes" diff --git a/src/MeshWeaver.Documentation/Data/Architecture/OrleansTestRoutingPattern.md b/src/MeshWeaver.Documentation/Data/Architecture/OrleansTestRoutingPattern.md index 6a45a9cb7a..feb8835433 100644 --- a/src/MeshWeaver.Documentation/Data/Architecture/OrleansTestRoutingPattern.md +++ b/src/MeshWeaver.Documentation/Data/Architecture/OrleansTestRoutingPattern.md @@ -85,6 +85,8 @@ A hub in the test process is a *hosted hub*, not a grain on the silo. Routing re `MeshConfiguration.DefaultStreamRoutedAddressTypes` is `{ "portal", "client", "cache", "mesh" }`, declared at static-init time so no configurator ordering can make a built-in type "go missing". Modules add their own with `MeshBuilder.AddStreamRoutedAddressType(...)` โ€” Graph registers `import`, the gRPC hosting registers its Python and Node types. +> **๐Ÿšจ Stream-ROUTED is not the same as CLIENT-HOSTED, and since [Durable Streams Are Mesh Nodes](/Doc/Architecture/DurableStreamsViaMeshNodes) the difference decides what a delivery gets.** A stream-routed address is first attempted as a *directed* `IPodHubGrain.Deliver` call to the pod that claimed it; the memory stream is reached only when the grain answers "not here", **and only for an address type separately declared `MeshBuilder.AddClientHostedAddressType(...)`** โ€” i.e. one whose hubs live in an Orleans CLIENT process, which cannot host a grain. Production declares none; the Orleans test rig declares all four built-in types in `OrleansTestMeshExtensions.ConfigurePortalMesh`, because it genuinely hosts a hub of each on its cluster client. For anything else "not here" is answered with a transient `ShuttingDown` + `TargetUnserved` NACK instead of a publish. **If you add a test hub on the CLUSTER CLIENT at a new address type, declare it in both places** โ€” stream-routed *and* client-hosted โ€” or its cross-silo deliveries will be NACK'd rather than carried. + The registration step looks like this: ```csharp diff --git a/src/MeshWeaver.Documentation/Data/WhatsNew/2026-08-30-a-request-to-a-restarting-pod-no-longer-hangs.md b/src/MeshWeaver.Documentation/Data/WhatsNew/2026-08-30-a-request-to-a-restarting-pod-no-longer-hangs.md new file mode 100644 index 0000000000..38131df237 --- /dev/null +++ b/src/MeshWeaver.Documentation/Data/WhatsNew/2026-08-30-a-request-to-a-restarting-pod-no-longer-hangs.md @@ -0,0 +1,36 @@ +--- +Name: A page waiting on a restarting pod recovers in milliseconds instead of a minute +Category: Fix +Description: Requests aimed at a hub whose pod is mid-restart now get an immediate "try again" and recover on their own, instead of waiting out a 30โ€“60 second silence. +Icon: Sparkle +Order: -20260830 +--- + +# A page waiting on a restarting pod recovers in milliseconds instead of a minute + +When the portal runs on more than one pod, every message has to find the pod that owns the hub it is +addressed to. During a rolling update โ€” and briefly whenever a hub moves between pods โ€” a message +could arrive while the owning pod had not yet announced itself. The message was then handed to a +shared queue as a fallback, and that queue had two bad days in it: if nobody was listening the +message was accepted and silently thrown away, and if the queue itself was stuck the sender waited +out its entire budget, thirty to sixty seconds, for an answer that was never coming. What you saw +was a view stuck on "loading", a chat reply that never appeared, or a save that seemed to do +nothing โ€” usually clearing up only when you reloaded the page. + +Two changes remove that window rather than shortening it. + +**A pod now keeps claiming its hubs until the claim sticks.** Previously it tried for about three +seconds and then quietly stopped, after which that hub stayed on the slow fallback for as long as +the pod lived, with nothing said anywhere. The claim now keeps trying โ€” with a backoff, so it costs +nothing โ€” and only stops for a real reason: the hub going away, or the pod shutting down. If a claim +has not landed quickly, that is now reported once, naming the hub, so it can be investigated instead +of going unnoticed for days. + +**And a message with nowhere to go is answered immediately.** Instead of being pushed into the +fallback queue, it comes straight back as "that hub is not reachable right now โ€” ask again". Every +part of the portal that waits on data already knows how to handle that answer: live views keep their +subscription and resume, editors retry, and nothing is torn down. So the roll window that used to +cost you a minute of silence now costs a round trip you will not notice. + +The old fallback is kept only where it is genuinely the only route available, which in this +deployment is nowhere at all. diff --git a/src/MeshWeaver.Hosting.Orleans.TestBase/OrleansTestMeshExtensions.cs b/src/MeshWeaver.Hosting.Orleans.TestBase/OrleansTestMeshExtensions.cs index 3c7cd97b0c..d558c62d53 100644 --- a/src/MeshWeaver.Hosting.Orleans.TestBase/OrleansTestMeshExtensions.cs +++ b/src/MeshWeaver.Hosting.Orleans.TestBase/OrleansTestMeshExtensions.cs @@ -29,6 +29,25 @@ public static MeshBuilder ConfigurePortalMesh( { var assemblyLocation = typeof(OrleansTestMeshExtensions).Assembly.Location; var configured = builder + // ๐Ÿšจ THE RIG HOSTS ITS HUBS ON AN ORLEANS CLIENT, and this is where it says so. An + // Orleans client cannot host a grain, so IPodHubGrain.Attach can never land for a hub + // that lives there and the memory stream is not a fallback for it โ€” it is the only + // transport there is. The rig puts a hub of EVERY built-in stream-routed type on its + // cluster client: + // client โ€” OrleansTestBase.GetClient / SharedOrleansFixture.GetClient + // mesh โ€” the client host's own root hub, via RootMeshHubReplyStreamService + // portal โ€” OrleansDocumentationTest / OrleansGraphDataTest / OrleansInteractiveMarkdownTest + // cache โ€” the client host's MeshNodeStreamCache hub + // so all four are declared. PRODUCTION declares NONE: no production process hosts mesh + // hubs as an Orleans client, and there a PodHubNotHere therefore means "the owner is a + // silo whose claim has not landed", which now gets a transient NACK instead of a stream + // publish that succeeds and discards (#2320/#2322/#2406). These declarations go away + // when the rig hosts its hubs on a silo โ€” at which point AddMemoryStreams goes with + // them. See Doc/Architecture/DurableStreamsViaMeshNodes. + .AddClientHostedAddressType("client") + .AddClientHostedAddressType("mesh") + .AddClientHostedAddressType("portal") + .AddClientHostedAddressType("cache") .InstallAssemblies(assemblyLocation) .AddMeshNodes(MeshNode.FromPath($"{AddressExtensions.AppType}/HubFactory") with { diff --git a/src/MeshWeaver.Hosting.Orleans/RoutingGrain.cs b/src/MeshWeaver.Hosting.Orleans/RoutingGrain.cs index 34549f9e53..316035e36c 100644 --- a/src/MeshWeaver.Hosting.Orleans/RoutingGrain.cs +++ b/src/MeshWeaver.Hosting.Orleans/RoutingGrain.cs @@ -101,6 +101,14 @@ internal class RoutingGrain( /// private readonly DeadTargetRefusalLog nackLog = new(DeadTargetRefusalLog.DefaultWindow); + /// + /// The same window again for 's refusal, keyed by the + /// unreachable ADDRESS. Deliberately its OWN instance rather than a share of + /// : that one is cleared the moment a live stream subscriber is found, + /// and a pod-hub refusal has nothing to do with stream subscribers. + /// + private readonly DeadTargetRefusalLog podHubRefusalLog = new(DeadTargetRefusalLog.DefaultWindow); + /// /// Probes whether this process's DI container still resolves โ€” the positive half of /// (issue #2638). On a live container this is a cheap lookup of an @@ -377,21 +385,27 @@ private void ReportDrained(int inFlight) /// requester spends its full budget on silence, and nothing anywhere logs a thing. A directed /// call has an OUTCOME, which is the entire point. /// - /// The fallback is deliberate and is not scaffolding. - /// means "no silo serves this address through the grain - /// transport", and there are two very different reasons for that: + /// ๐Ÿšจ The stream fallback is taken by DECLARATION, never because the grain answered + /// "not here" โ€” Doc/Architecture/DurableStreamsViaMeshNodes, release N+2 of the roll + /// plan. means "no silo serves this address through the + /// grain transport", and that has two very different causes which the exception cannot tell + /// apart: /// - /// the owner is still on the PREVIOUS RELEASE and never claimed the address โ€” true for - /// the whole overlap window of every rolling deploy, which is precisely the window in which - /// the last change in this area stranded 39 addresses by not having a fallback (#1770); /// the owner is an Orleans CLIENT process, which cannot host a grain at all. For those - /// hubs the stream is not a legacy path, it is the only path โ€” so this branch is - /// PERMANENT. + /// hubs the stream is not a legacy path, it is the only path โ€” and that is now + /// declared by the host that + /// knows it (the Orleans test rig declares the built-in stream-routed types; production + /// declares nothing); + /// the owner is a SILO whose claim has not landed yet โ€” the overlap window of a rolling + /// deploy, and the window in which not having a fallback stranded 39 addresses (#1770). /// - /// Either way the fallback is safe rather than a return to silence, because the stream leg - /// now checks for a subscriber before it publishes and NACKs when there is none. Full plan, - /// including the one log line that says when the overlap window has closed: - /// Doc/Architecture/PodHubDeliveryRollPlan. + /// Publishing on the second reading is what kept #2320 / #2322 / #2406 reachable: a + /// publish into a stream with no live subscriber SUCCEEDS and discards (the subscriber probe + /// narrows this but fails open by design), and a publish into a stream whose queue grain is + /// wedged or whose producer never registered stalls for 30โ€“60 s. So that reading now gets a + /// fast TRANSIENT NACK instead โ€” see โ€” and the overlap window + /// is covered by the sender's own recovery plus the owner's now-indefinite claim + /// (OrleansRoutingService.AttachPodHub), not by a publish that succeeds and discards. /// /// ๐Ÿšจ Issue #2299: a TRANSIENT rejection used to be treated exactly like a terminal /// one. Prod evidence names two distinct transports-level rejections for the same underlying @@ -435,6 +449,12 @@ private IObservable BuildPodHubRoute( void PostFailureToSender(string failureMessage, ErrorType errorType) => PostFailure(delivery, address, streamProvider, grainFactory, failureMessage, errorType); + // The three-argument form, so the ROUTER's authoritative "no silo serves this hub" stamp + // travels with the verdict โ€” see AnswerPodHubNotHere. + void PostVerdictToSender(string failureMessage, ErrorType errorType, bool targetUnserved) => + PostFailure(delivery, address, streamProvider, grainFactory, failureMessage, errorType, + targetUnserved: targetUnserved); + return DeliverToGrainObservable( () => grainFactory.GetGrain(addressPath).Deliver(delivery), addressPath, delivery.Id, logger) @@ -444,7 +464,9 @@ void PostFailureToSender(string failureMessage, ErrorType errorType) => return Unit.Default; }) .Catch(ex => IsPodHubNotHere(ex) - ? FallBackToStream() + ? AnswerPodHubNotHere( + delivery, addressPath, address.Type, meshConfig, + FallBackToStream, PostVerdictToSender, logger, podHubRefusalLog) // A REAL failure of the call, transient retries exhausted (or a non-transient // fault) โ€” the owning silo threw, went away mid-call, or the placement could not // be made. This is the whole gain over a publish: it is OBSERVABLE, so it becomes @@ -464,15 +486,16 @@ void PostFailureToSender(string failureMessage, ErrorType errorType) => IObservable FallBackToStream() { RoutingGrainTrace.Write($"RoutingGrain.RouteMessage POD_HUB_NOT_HERE addr={addressPath} id={delivery.Id}"); - // ๐Ÿšจ THE LINE THAT DECIDES WHEN THE OVERLAP WINDOW HAS CLOSED. Information, once per - // delivery that takes the fallback: during a roll it is the plan working; after a roll - // has fully completed, for a SILO-hosted address, it is a hub that never claimed itself - // and wants investigating. Never read its absence as proof โ€” read a full rolling deploy - // with none of these as the signal that the grain transport carries everything it can. + // ๐Ÿšจ Reached ONLY for an address type DECLARED client-hosted. Kept at Information at + // its original level and cost: production declares no such type, so in the fleet this + // line cannot be emitted at all โ€” and if one is ever declared, this is the line that + // says a stream publish (which succeeds and discards when nobody is subscribed) is + // still carrying traffic. It no longer measures a roll window; that job moved to the + // owner-side "Pod-hub claim โ€ฆ did not land" Warning. logger.LogInformation( - "[ROUTE] Pod-hub grain for {Address} is not attached โ€” falling back to the stream publish. " - + "Expected while a previous-release pod still owns that hub, and permanently for a hub " - + "owned by an Orleans client process (a client cannot host a grain).", + "[ROUTE] Pod-hub grain for {Address} is not attached โ€” falling back to the stream publish, " + + "because this address type is DECLARED client-hosted and an Orleans client cannot host " + + "a grain. No other condition reaches this line.", addressPath); return BuildStreamRoute(delivery, address, addressPath, streamProvider, grainFactory); } @@ -497,6 +520,94 @@ IObservable TerminalCallFailure(Exception ex) } } + /// + /// What a means for THIS address type โ€” the whole of + /// release N+2 in Doc/Architecture/DurableStreamsViaMeshNodes, in one decision. + /// + /// + /// Declared client-hosted () + /// โ†’ . The owner is a process that cannot host a grain, + /// so the stream is not a fallback, it is the only transport there is. Nothing in production + /// declares one; the Orleans test rig declares the built-in stream-routed types, because it + /// hosts a hub of each on its cluster client. + /// Anything else โ†’ a TRANSIENT NACK, right here, in one hop. No silo currently + /// serves that hub; the sender is told so inside the directed call's own budget instead of + /// the stream's, and its own recovery runs. + /// + /// + /// ๐Ÿšจ , never the terminal + /// and never . The consumers + /// with recovery machinery of their own โ€” SynchronizationStream's resubscribe latch, + /// MeshNodeStreamCache.IsTransientOwnerFailure โ€” RIDE OUT ShuttingDown and TEAR + /// DOWN on a terminal verdict, and "no silo serves this hub right now" is a lifecycle + /// transition by construction: it is the rolling deploy's overlap window, and the owner's claim + /// (OrleansRoutingService.AttachPodHub) keeps retrying until it lands. That pairing is + /// what makes this safe WITHOUT the stream publish: the answer is a NACK the caller retries, + /// not a publish that succeeds and discards. + /// + /// ๐Ÿšจ Stamped โ€” the same authoritative + /// shape produces, because it is the same statement made by + /// the same authority: the ROUTER asked the cluster (here Orleans' grain directory, there the + /// stream subscription registry) and was told nobody serves that address. ๐Ÿšจ That stamp is the + /// OWNER-side eviction gate (DataExtensions.HandleTargetUnservedFailure, #2426/#2546), + /// which is why it had to be re-gated on the stamp ALONE in the same change: it used to also + /// require , so leaving that in place would have made this + /// verdict inert and re-opened the fan-out-to-a-corpse leak for every dead circuit. The two + /// halves are complementary: the SUBSCRIBER rides the transient verdict out and re-asks, while + /// the OWNER drops the server-side stream it can no longer push to. + /// + /// Windowed like : the first refusal of an address in the + /// window earns the full Warning line, repeats log at Debug and are counted into + /// the next full one. Warning rather than Error because the verdict is transient by + /// construction; every delivery is still refused and still NACKed, window or no window. + /// + /// The delivery the pod-hub grain refused. + /// The destination address, as a path. + /// The destination's address TYPE โ€” the key the declaration is on. + /// The mesh configuration carrying the declarations. + /// The stream leg, invoked ONLY for a declared client-hosted type. + /// Message, error type, and the TargetUnserved stamp. + /// Logger for the refusal line. + /// Window for the full line; null logs every refusal in full. + internal static IObservable AnswerPodHubNotHere( + IMessageDelivery delivery, + string addressPath, + string addressType, + MeshConfiguration meshConfig, + Func> fallBackToStream, + Action postFailureToSender, + ILogger logger, + DeadTargetRefusalLog? refusalLog = null) + { + if (meshConfig.ClientHostedAddressTypes.Contains(addressType)) + return fallBackToStream(); + + var reason = + $"Directed delivery to pod hub '{addressPath}' was refused: no silo in this cluster is " + + "currently serving that hub. Transient โ€” the owner claims its address for as long as it " + + "is registered, so a retry is the correct response."; + var suppressedSincePriorReport = 0; + if (refusalLog is null || refusalLog.ShouldReport(addressPath, out suppressedSincePriorReport)) + logger.LogWarning( + "[ROUTE] {Reason} Message {MessageType} ({DeliveryId}) from {Sender} was NOT posted to the " + + "Orleans stream โ€” a publish to a subscriber-less stream succeeds and discards, and a " + + "publish into a wedged queue grain stalls for the sender's whole budget (#2320/#2322/" + + "#2406). Surfacing a transient DeliveryFailure to the sender instead. {Suppressed} " + + "earlier refusal(s) of this address since the last such line were logged at Debug.", + reason, delivery.Message?.GetType().Name ?? "(null)", delivery.Id, delivery.Sender, + suppressedSincePriorReport); + else + logger.LogDebug( + "[ROUTE] {Reason} Message {MessageType} ({DeliveryId}) from {Sender} was NOT posted; " + + "refusal windowed (see the Warning line for this address). Surfacing a transient " + + "DeliveryFailure to the sender.", + reason, delivery.Message?.GetType().Name ?? "(null)", delivery.Id, delivery.Sender); + RoutingGrainTrace.Write( + $"RoutingGrain.RouteMessage POD_HUB_NOT_HERE_REFUSED addr={addressPath} id={delivery.Id} sender={delivery.Sender}"); + postFailureToSender(reason, ErrorType.ShuttingDown, true); + return Observable.Return(Unit.Default); + } + /// /// Is this the grain saying "not through this transport, not here"? Orleans wraps a thrown grain /// exception on its way back to the caller, so the test walks the inner chain โ€” matching by type @@ -815,10 +926,13 @@ private void PostFailure( new DeliveryFailure(echoedDelivery, failureMessage) { ErrorType = errorType, - // Stamped ONLY by the no-live-subscriber refusal (see PostRefusalToSender): the - // authoritative routing verdict that lets the owner-side client-subscription - // eviction distinguish "that subscriber's process is gone" from an - // application-level NotFound a live hub answered. + // Stamped ONLY by the router's two AUTHORITATIVE "nobody serves that address" + // verdicts โ€” the no-live-subscriber refusal (RefuseNoSubscriber, via + // PostRefusalToSender) and the pod-hub refusal (AnswerPodHubNotHere) โ€” which is + // what lets the owner-side client-subscription eviction distinguish "that + // subscriber's process is gone" from an application-level NotFound a live hub + // answered. The stamp is the eviction gate; the ErrorType beside it says whether + // the SENDER should keep its own recovery armed, and the two are independent. TargetUnserved = targetUnserved, }, new PostOptions(address) diff --git a/src/MeshWeaver.Mesh.Contract/MeshBuilder.cs b/src/MeshWeaver.Mesh.Contract/MeshBuilder.cs index f6d01d9574..30a6089674 100644 --- a/src/MeshWeaver.Mesh.Contract/MeshBuilder.cs +++ b/src/MeshWeaver.Mesh.Contract/MeshBuilder.cs @@ -479,6 +479,7 @@ private void Register() var accessConfig = NodeTypeAccessConfig; var routingRules = QueryRoutingRules; var streamRoutedTypes = StreamRoutedAddressTypes; + var clientHostedTypes = ClientHostedAddressTypes; ConfigureServices(services => services .AddSingleton(_ => @@ -498,7 +499,13 @@ private void Register() autocompleteExcludedNodeTypes: excludedTypes.Count > 0 ? excludedTypes : null, queryRoutingRules: routingRules, streamRoutedAddressTypes: streamRoutedTypes, - nodeTypeGates: accessConfig.BuildGates()); + nodeTypeGates: accessConfig.BuildGates()) + { + // An init property, not a ctor argument โ€” see MeshConfiguration for why a + // trailing optional parameter would still be a binary break for a module that + // is already published. + ClientHostedAddressTypes = clientHostedTypes, + }; }) // Static nodes registered via AddMeshNodes(...) flow as an // IStaticNodeProvider. Application code reads them via @@ -721,4 +728,27 @@ public MeshBuilder AddStreamRoutedAddressType(string addressType) internal HashSet StreamRoutedAddressTypes { get; } = new(global::MeshWeaver.Mesh.MeshConfiguration.DefaultStreamRoutedAddressTypes, StringComparer.Ordinal); + + /// + /// Declares that hubs at this address-type prefix are hosted in an Orleans CLIENT process, + /// which cannot host a grain. For those โ€” and ONLY those โ€” the Orleans memory stream stays the + /// transport when the pod-hub grain answers "not here": there is no directed call that could + /// ever reach them. See for why this + /// is a declaration rather than an inference from the grain's answer, and + /// Doc/Architecture/DurableStreamsViaMeshNodes for the design. + /// + /// ๐Ÿšจ Nothing in production declares one. This exists for the Orleans test rig, + /// which hosts hubs on a cluster client. Declaring a type here opts it + /// OUT of the transient NACK and back into a stream publish that succeeds-and-discards when + /// nobody is subscribed โ€” do not add one to make a routing symptom go away. + /// + /// The address-type prefix (e.g. client). + /// The builder for method chaining. + public MeshBuilder AddClientHostedAddressType(string addressType) + { + ClientHostedAddressTypes.Add(addressType); + return this; + } + + internal HashSet ClientHostedAddressTypes { get; } = new(StringComparer.Ordinal); } diff --git a/src/MeshWeaver.Mesh.Contract/MeshConfiguration.cs b/src/MeshWeaver.Mesh.Contract/MeshConfiguration.cs index 8ee60d2544..04e498f58f 100644 --- a/src/MeshWeaver.Mesh.Contract/MeshConfiguration.cs +++ b/src/MeshWeaver.Mesh.Contract/MeshConfiguration.cs @@ -1,3 +1,4 @@ +using System.Collections.Immutable; using System.Runtime.CompilerServices; using MeshWeaver.Mesh.Security; using MeshWeaver.Messaging; @@ -68,6 +69,53 @@ public class MeshConfiguration( public static readonly IReadOnlySet DefaultStreamRoutedAddressTypes = new HashSet(StringComparer.Ordinal) { "portal", "client", "cache", "mesh" }; + /// + /// Address-type prefixes whose hubs are hosted in an Orleans CLIENT process โ€” a process that + /// cannot host a grain, so the cluster can never reach those hubs by a directed + /// IPodHubGrain.Deliver call and the Orleans memory stream is not a fallback but the + /// only transport there is. Declared via MeshBuilder.AddClientHostedAddressType. + /// + /// ๐Ÿšจ EMPTY by default, and empty in production. No production process hosts mesh + /// hubs as an Orleans client โ€” UseOrleansMeshClient has only test-fixture callers, the + /// distributed portal is a co-hosted silo, and the monolith / LocalMesh / bake host run + /// no Orleans at all. The Orleans TEST rig does host hubs on a client โ€” one of every built-in + /// stream-routed type โ€” and declares all of them here. + /// + /// Why a DECLARATION and not an inference. RoutingGrain.BuildPodHubRoute + /// used to take the stream fallback whenever the pod-hub grain answered + /// PodHubNotHereException โ€” a condition that means BOTH "the owner is an Orleans client" + /// and "the owner is a silo whose claim has not landed yet". Publishing on the second reading + /// is what kept #2320 / #2322 / #2406 reachable: a publish into a stream with no live + /// subscriber SUCCEEDS and discards, and a publish into a wedged queue grain stalls for + /// 30โ€“60 s. Asking the DECLARATION instead separates the two: a client-hosted type keeps its + /// only transport, and everything else gets a fast, transient NACK the sender rides out. See + /// Doc/Architecture/DurableStreamsViaMeshNodes. + /// + /// An init property rather than a primary-constructor parameter, deliberately: + /// adding a parameter โ€” even a trailing optional one โ€” changes the constructor's SIGNATURE, and + /// an already-published module binds that signature in its IL. Additive here, breaking there. + /// + /// ๐Ÿšจ SNAPSHOTTED into an immutable set on assignment, not stored by reference. + /// MeshBuilder hands over its own live HashSet, and an IReadOnlySet that is + /// really a HashSet can be cast back and written to โ€” either way a routing decision would + /// depend on when it was asked. This set decides whether a delivery is carried or NACK'd, so it + /// is fixed at construction and cannot move afterwards. + /// + public IReadOnlySet ClientHostedAddressTypes + { + get => clientHostedAddressTypes; + init => clientHostedAddressTypes = value is null or { Count: 0 } + ? EmptyAddressTypes + : ImmutableHashSet.CreateRange(StringComparer.Ordinal, value); + } + + private readonly IReadOnlySet clientHostedAddressTypes = EmptyAddressTypes; + + /// Immutable, read-only, initialised once and never written โ€” the sanctioned + /// static readonly shape (see NoStaticState.md). + private static readonly ImmutableHashSet EmptyAddressTypes = + ImmutableHashSet.Empty.WithComparer(StringComparer.Ordinal); + // No public Nodes / MeshNodes property. Static nodes registered via // MeshBuilder.AddMeshNodes(...) flow through StaticMeshNodeListProvider // (registered as an IStaticNodeProvider). Application code reads them via diff --git a/test/MeshWeaver.Data.Test/UnservedVerdictEvictionTest.cs b/test/MeshWeaver.Data.Test/UnservedVerdictEvictionTest.cs new file mode 100644 index 0000000000..67ea6e1afa --- /dev/null +++ b/test/MeshWeaver.Data.Test/UnservedVerdictEvictionTest.cs @@ -0,0 +1,164 @@ +using System; +using System.Reactive.Linq; +using System.Reactive.Threading.Tasks; +using System.Text.Json; +using System.Threading.Tasks; +using MeshWeaver.Data.TestDomain; +using MeshWeaver.Fixture; +using MeshWeaver.Messaging; +using Microsoft.Extensions.DependencyInjection; +using Xunit; + +namespace MeshWeaver.Data.Test; + +/// +/// ๐Ÿšจ The owner-side eviction keys on the router's STAMP, not on the ErrorType it happens to +/// arrive with โ€” the coupling that would otherwise have silently re-opened issues #2426/#2546 +/// when the routing leg stopped publishing to the Orleans memory stream. +/// +/// What the eviction is for. An owner hub serving a data stream to a subscriber whose +/// PROCESS died keeps that server-side stream forever โ€” only an UnsubscribeRequest disposes +/// one, and a corpse sends none. So the owner fans every change out to the dead address and the +/// router refuses each one. The one signal that can end the loop is the router's authoritative +/// verdict reaching the owner, which disposes that +/// subscriber's server-side stream (20,718 error lines in 3 h on memex-cloud before it did). +/// +/// ๐Ÿšจ Why this test exists NOW. That verdict used to have exactly one producer โ€” the +/// stream leg's no-live-subscriber refusal, which stamps it beside +/// โ€” so the handler could afford a redundant +/// ErrorType == NotFound test beside the stamp. Since +/// RoutingGrain.AnswerPodHubNotHere the router reaches the same conclusion ONE HOP EARLIER, +/// through Orleans' grain directory, and reports it as the TRANSIENT +/// โ€” because a hub whose owner is mid-roll must not have its +/// subscriber's mirrors torn down. Had the ErrorType test survived, that verdict would have been +/// inert here and every dead circuit in the fleet would have leaked its server-side streams again. +/// The two facts are complementary, not contradictory: the SUBSCRIBER rides the transient verdict +/// out and re-asks; the OWNER drops the half it can no longer push to. +/// +/// Fails on unfixed code: +/// โ€” with the +/// ErrorType == NotFound test in place the eviction counter never leaves zero. +/// +/// See Doc/Architecture/DurableStreamsViaMeshNodes and +/// Doc/Architecture/ErrorPropagationAndWedges. +/// +public class UnservedVerdictEvictionTest(ITestOutputHelper output) : HubTestBase(output) +{ + protected override MessageHubConfiguration ConfigureHost(MessageHubConfiguration configuration) => + base.ConfigureHost(configuration) + .AddData(data => data.AddSource(source => source + .WithType(t => t.WithInitialData(TestData.BusinessUnits)))); + + protected override MessageHubConfiguration ConfigureClient(MessageHubConfiguration configuration) => + base.ConfigureClient(configuration) + .AddData(data => data.AddHubSource( + CreateHostAddress(), + source => source.WithType())); + + /// + /// Drives a real subscription so the host genuinely holds a server-side stream for the client, + /// then hands the host the router's verdict about it. + /// + private async Task<(IMessageHub Host, IMessageHub Client, Workspace HostWorkspace)> ServingClientAsync() + { + var client = GetClient(); + var clientWorkspace = client.ServiceProvider.GetRequiredService(); + // Data arrived โ‡’ the owner built and is serving a stream for this subscriber. Waiting on + // the DATA rather than on a delay is what makes "the owner had something to evict" a fact. + var units = await clientWorkspace.GetObservable() + .Should().Within(10.Seconds()).Emit(); + units.Should().NotBeEmpty("the subscription must be live before the verdict arrives"); + + var host = GetHost(); + var hostWorkspace = (Workspace)host.GetWorkspace(); + hostWorkspace.ClientSubscriptionsEvicted.Should().Be(0, + "nothing has told the owner that this subscriber is unreachable yet"); + return (host, client, hostWorkspace); + } + + /// + /// The router's verdict, in exactly the shape RoutingGrain.PostFailure stamps it: the + /// fanned-out delivery it could not carry, the TargetUnserved stamp, and the transient + /// ErrorType that keeps the SUBSCRIBER's own recovery armed. + /// + private static DeliveryFailure RouterVerdict(IMessageHub host, Address subscriber, ErrorType errorType) => + new(new MessageDelivery( + host.Address, subscriber, + new RawJson("{\"$type\":\"DataChangedEvent\"}"), + JsonSerializerOptions.Default), + $"Directed delivery to pod hub '{subscriber}' was refused: no silo in this cluster is " + + "currently serving that hub.") + { + ErrorType = errorType, + TargetUnserved = true, + }; + + private static Task EvictedAsync(Workspace workspace) => + Observable.Interval(50.Milliseconds()) + .StartWith(0L) + .Select(_ => workspace.ClientSubscriptionsEvicted) + .Where(n => n >= 1) + .FirstAsync() + .Timeout(20.Seconds()) + .ToTask(TestContext.Current.CancellationToken); + + /// + /// ๐Ÿšจ THE PIN. The transient verdict the retired stream leg's replacement produces must still + /// end the fan-out at its source. + /// + [HubFact] + public async Task ATransientUnservedVerdict_EvictsTheOwnersServerSideStream() + { + var (host, client, hostWorkspace) = await ServingClientAsync(); + + host.Post(RouterVerdict(host, client.Address, ErrorType.ShuttingDown), + o => o.WithTarget(host.Address)); + + (await EvictedAsync(hostWorkspace)).Should().BeGreaterThanOrEqualTo(1, + "the eviction gate is the router's TargetUnserved STAMP โ€” the only component that asks " + + "the cluster โ€” and never the ErrorType beside it. Gating on NotFound made the pod-hub " + + "refusal inert and left every dead circuit fanning out forever (#2426/#2546)"); + } + + /// + /// The verdict that has always worked, unchanged: the stream leg's no-live-subscriber refusal + /// stamps the same verdict beside , and it must keep evicting. + /// + [HubFact] + public async Task ATerminalUnservedVerdict_StillEvicts() + { + var (host, client, hostWorkspace) = await ServingClientAsync(); + + host.Post(RouterVerdict(host, client.Address, ErrorType.NotFound), + o => o.WithTarget(host.Address)); + + (await EvictedAsync(hostWorkspace)).Should().BeGreaterThanOrEqualTo(1, + "widening the gate must not narrow it anywhere โ€” the original producer's verdict is " + + "untouched"); + } + + /// + /// ๐Ÿšจ NEGATIVE CONTROL, and the reason the gate is the stamp at all. A LIVE hub also answers + /// NotFound โ€” an unhandled request โ€” and evicting on that would tear down a healthy + /// subscriber's stream. Only the router stamps TargetUnserved; without the stamp the + /// failure is not ours to act on, whatever its ErrorType says. + /// + [HubFact] + public async Task AnUnstampedFailure_NeverEvicts_WhateverItsErrorType() + { + var (host, client, hostWorkspace) = await ServingClientAsync(); + + foreach (var errorType in new[] { ErrorType.NotFound, ErrorType.ShuttingDown, ErrorType.Failed }) + host.Post( + RouterVerdict(host, client.Address, errorType) with { TargetUnserved = false }, + o => o.WithTarget(host.Address)); + + // A negative assertion with no positive signal to filter for: give the hub's action block + // time to process all three, then assert nothing was evicted. + await Task.Delay(1_000, TestContext.Current.CancellationToken); + + hostWorkspace.ClientSubscriptionsEvicted.Should().Be(0, + "an application-level NACK from a LIVE hub must never cost a healthy subscriber its " + + "server-side stream โ€” absence of the router's stamp means 'do not evict'"); + } +} diff --git a/test/MeshWeaver.Hosting.Orleans.Test/PodHubClaimLifetimeTest.cs b/test/MeshWeaver.Hosting.Orleans.Test/PodHubClaimLifetimeTest.cs new file mode 100644 index 0000000000..4b1dc36adf --- /dev/null +++ b/test/MeshWeaver.Hosting.Orleans.Test/PodHubClaimLifetimeTest.cs @@ -0,0 +1,361 @@ +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Linq; +using System.Reactive.Linq; +using System.Threading; +using System.Threading.Tasks; +using MeshWeaver.Connection.Orleans; +using MeshWeaver.Messaging; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using Orleans; +using Orleans.Runtime; +using Xunit; + +namespace MeshWeaver.Hosting.Orleans.Test; + +/// +/// ๐Ÿšจ Issue #1742's stated open residual โ€” "a claim that fails to land degrades SILENTLY back to +/// the stream" โ€” at the seam that produced it. +/// +/// The defect. OrleansRoutingService.AttachPodHub claims an address for THIS +/// process so the rest of the cluster can reach its hub with a directed IPodHubGrain.Deliver +/// call instead of an Orleans memory-stream publish. The claim carried a BUDGET โ€” five retries over +/// โ‰ˆ3 s โ€” and on exhausting it logged a Debug line and stopped. After that a SILO-hosted hub +/// kept the stream transport for the whole life of the process, with no signal anywhere on the +/// owning side; the only trace was a router-side fallback line in a different pod's log. +/// +/// The rule. #2426: no lifetime that depends on a message a restarting process would +/// never send, and none that expires on a counter either. The claim's terminals are now DERIVED โ€” +/// the hub's registration being disposed, or +/// โ€” plus one that is genuine IMPOSSIBILITY rather than a budget: a process that cannot host a grain +/// can never win the claim, so there the initial budget IS the end and the give-up stays at +/// Debug because it is the expected permanent outcome. +/// +/// No cluster, no clock, no mocks of anything of ours. The subject is the real +/// ; is a RECORDER whose +/// pod-hub grain answers false โ€” the exact answer PodHubGrain.Attach gives when the +/// activation lands on a silo with no local route for the address โ€” and the backoff is collapsed +/// through the instance seam so the POLICY is what is measured. The stopping signal is a real +/// , cancelled through the same +/// the runtime calls on SIGTERM. +/// +/// Fails on unfixed code: +/// +/// observes six attempts and zero Warning lines. +/// +/// See Doc/Architecture/DurableStreamsViaMeshNodes. +/// +public class PodHubClaimLifetimeTest +{ + private static readonly Address Hub = new("portal", "claim-lifetime"); + + /// The gap that separates "still retrying" from "gave up", with margin. + private const int BeyondTheBudget = OrleansRoutingService.PodHubAttachRetries + 5; + + private static IObservable Ignore(IMessageDelivery d, CancellationToken _) => + Observable.Return(d); + + private static Microsoft.Extensions.DependencyInjection.ServiceProvider Services( + IHostApplicationLifetime? lifetime = null) + { + var services = new ServiceCollection(); + // Registered but never fired: the stream attach then stays parked on the #1129 readiness + // gate, so this test needs no stream provider and touches none. + services.AddSingleton(new OrleansStreamingReadiness()); + if (lifetime is not null) + services.AddSingleton(lifetime); + return services.BuildServiceProvider(); + } + + private static OrleansRoutingService Router( + RefusingGrainFactory factory, IServiceProvider sp, RecordingLogger logger, bool canHostGrains) => + new(factory, sp, logger) + { + // Instance seams, never static state: the POLICY is under test, not the clock, and not + // whether a silo happens to be running underneath. + CanHostGrains = canHostGrains, + ClaimBackoff = _ => TimeSpan.FromMilliseconds(1), + }; + + /// Bounded wait on a POSITIVE signal: the claim reached at least this many attempts. + private static async Task WaitForAttempts(RefusingGrainFactory factory, int target) + { + for (var i = 0; i < 400; i++) + { + var now = factory.AttachCalls; + if (now >= target) + return now; + await Task.Delay(25); + } + throw new TimeoutException( + $"the claim stopped at {factory.AttachCalls} attempt(s), short of {target} โ€” it gave up"); + } + + /// + /// Bounded wait for the attempt count to STOP moving. There is no completion signal to await + /// for a claim that ends, so two consecutive equal readings after the first attempt landed is + /// the settled state โ€” the same technique as StreamAttachTransientRetryTest. + /// + private static async Task WaitUntilSettled(RefusingGrainFactory factory) + { + var last = -1; + for (var i = 0; i < 400; i++) + { + await Task.Delay(25); + var now = factory.AttachCalls; + if (now > 0 && now == last) + return now; + last = now; + } + throw new TimeoutException( + $"the claim never settled โ€” last observed attempt count {factory.AttachCalls}"); + } + + /// + /// ๐Ÿšจ THE REGRESSION. On unfixed code the claim makes exactly + /// PodHubAttachRetries + 1 attempts and then leaves a silo-hosted hub on the Orleans + /// stream forever, saying so only at Debug. + /// + [Fact] + public async Task AClaimThatCannotLand_WhereGrainsCanBeHosted_KeepsRetrying_AndWarnsExactlyOnce() + { + var factory = new RefusingGrainFactory(); + var logger = new RecordingLogger(); + await using var sp = Services(); + using var routing = Router(factory, sp, logger, canHostGrains: true); + + using var registration = routing.RegisterStream(Hub, Ignore); + var attempts = await WaitForAttempts(factory, BeyondTheBudget); + + attempts.Should().BeGreaterThan(OrleansRoutingService.PodHubAttachRetries + 1, + "the claim's lifetime is the HUB's, not a counter โ€” a bounded claim that gave up left a " + + "silo-hosted hub on the stream transport for the life of the process, invisibly, which " + + "is #1742's stated open residual"); + + var warnings = logger.Records.Where(r => r.Level == LogLevel.Warning).ToList(); + warnings.Should().ContainSingle( + "a silo that cannot claim its own hub is abnormal and must be named ONCE โ€” a line per " + + "attempt would be the log storm #2426/#2546 exist to remove"); + warnings[0].Message.Should().Contain(Hub.ToString()) + .And.Contain("did not land", + "the Loki gate in Doc/Architecture/DurableStreamsViaMeshNodes matches on " + + "\"Pod-hub claim for\" and \"did not land\""); + logger.Records.Should().NotContain( + r => r.Level == LogLevel.Debug && r.Message.Contains("did not land in this process"), + "the give-up line belongs to a process that cannot host a grain โ€” this one can, and it " + + "has not given up"); + } + + /// + /// The other direction, and what keeps the rule from becoming an unbounded poll: a process that + /// cannot host a grain can NEVER win this claim (PodHubGrain is + /// [PreferLocalPlacement], so from a cluster client the activation lands on some silo + /// that has no local route and answers false, for ever). That is impossibility, not a + /// budget โ€” and retrying it would cost the SILO one Information line per hub per backoff + /// interval, which is the storm shape this codebase removes rather than adds. + /// + [Fact] + public async Task AClaimThatCannotLand_WhereNoGrainCanBeHosted_StopsAtTheInitialBudget_AtDebug() + { + var factory = new RefusingGrainFactory(); + var logger = new RecordingLogger(); + await using var sp = Services(); + using var routing = Router(factory, sp, logger, canHostGrains: false); + + using var registration = routing.RegisterStream(Hub, Ignore); + var attempts = await WaitUntilSettled(factory); + + attempts.Should().Be(OrleansRoutingService.PodHubAttachRetries + 1, + "an Orleans client cannot host a grain, so the claim is impossible rather than slow โ€” " + + "spinning on it converges on nothing and costs the silo a log line per attempt"); + logger.Records.Should().NotContain(r => r.Level == LogLevel.Warning, + "there it is the EXPECTED permanent outcome โ€” a per-hub warning would be pure noise, " + + "which is why the level is derived from what this process can host and never from the " + + "attempt count"); + logger.Records.Should().Contain( + r => r.Level == LogLevel.Debug && r.Message.Contains("did not land"), + "the outcome is still on the record, at the level it deserves"); + } + + /// + /// TERMINAL 1 โ€” the hub's registration. The claim is alive exactly as long as the hub it claims + /// for; disposing the registration ends it, which is also the path Dispose takes through + /// the in-flight composite. + /// + [Fact] + public async Task DisposingTheRegistration_EndsTheClaim() + { + var factory = new RefusingGrainFactory(); + await using var sp = Services(); + using var routing = Router(factory, sp, new RecordingLogger(), canHostGrains: true); + + var registration = routing.RegisterStream(Hub, Ignore); + await WaitForAttempts(factory, BeyondTheBudget); + + registration.Dispose(); + + // Let any attempt already in flight land, then assert the count is STILL โ€” a negative + // "nothing more happened" with no positive signal to filter for. + await Task.Delay(250); + var afterDisposal = factory.AttachCalls; + await Task.Delay(400); + factory.AttachCalls.Should().Be(afterDisposal, + "an indefinite claim whose terminal is not wired is an unbounded poll โ€” the whole point " + + "of a DERIVED lifetime is that the hub going away ends it"); + } + + /// + /// TERMINAL 2 โ€” the host. fires + /// strictly before the Orleans silo hosted service stops, and the claim re-reads it on every + /// re-subscribe (the gate lives inside the Defer), so a claim still bouncing when + /// shutdown begins stops asking instead of placing an activation on the silo that is leaving. + /// + [Fact] + public async Task TheHostBeginningToStop_EndsTheClaim() + { + // A real host purely for its real ApplicationLifetime โ€” nothing is started; StopApplication + // fires ApplicationStopping regardless, and that token is the signal under test. + var lifetime = new HostBuilder().Build().Services.GetRequiredService(); + var factory = new RefusingGrainFactory(); + await using var sp = Services(lifetime); + using var routing = Router(factory, sp, new RecordingLogger(), canHostGrains: true); + + using var registration = routing.RegisterStream(Hub, Ignore); + await WaitForAttempts(factory, BeyondTheBudget); + + lifetime.StopApplication(); + + await Task.Delay(250); + var afterStopping = factory.AttachCalls; + await Task.Delay(400); + factory.AttachCalls.Should().Be(afterStopping, + "claiming an address for a process that is going away only places an activation on the " + + "silo that is leaving โ€” the indefinite retry must not re-open that window"); + } + + /// + /// The policy itself, with no host at all: the backoff grows, is capped, and the initial budget + /// is small enough to be a REPORTING threshold rather than a wait anybody notices. + /// + [Fact] + public void TheClaimBackoff_GrowsAndIsCapped() + { + OrleansRoutingService.PodHubAttachRetries.Should().BeInRange(1, 10, + "this is the point at which a claim becomes reportable, not a budget anything waits out"); + + var waits = Enumerable.Range(0, OrleansRoutingService.PodHubAttachRetries + 1) + .Select(OrleansRoutingService.PodHubClaimBackoff) + .ToArray(); + + waits.Should().BeInAscendingOrder("backoff must grow, or it is a busy loop with a sleep in it"); + waits.Should().OnlyContain(w => w <= TimeSpan.FromSeconds(2), + "an indefinite retry MUST be capped โ€” the ceiling is what bounds its steady-state cost"); + waits[^1].Should().Be(TimeSpan.FromSeconds(2), + "the attempt index is clamped at the budget, so a claim that keeps retrying settles " + + "exactly at the ceiling instead of growing without bound"); + } + + /// + /// The pod-hub grain a claim that cannot land meets: Attach answers false, which + /// is verbatim what PodHubGrain.Attach returns when the activation lands on a silo that + /// has no local route for the address. Counting happens on the GRAIN, so a Detach during + /// teardown can never be mistaken for another attempt. + /// + private sealed class RefusingPodHubGrain : IPodHubGrain + { + private int attachCalls; + public int AttachCalls => Volatile.Read(ref attachCalls); + + public Task Attach() + { + Interlocked.Increment(ref attachCalls); + return Task.FromResult(false); + } + + public Task Detach() => Task.CompletedTask; + + public Task Deliver(IMessageDelivery delivery) => Task.FromResult(delivery); + } + + /// + /// Hands out the one refusing pod-hub grain. Only the string-key overload is implemented + /// because it is the only shape the mesh uses โ€” every other member throws, so a new call shape + /// fails loudly here instead of passing silently. Same recorder technique as + /// SiloShutdownActivationGateTest.RecordingGrainFactory. + /// + private sealed class RefusingGrainFactory : IGrainFactory + { + private readonly RefusingPodHubGrain podHub = new(); + private readonly ConcurrentQueue keys = new(); + + public int AttachCalls => podHub.AttachCalls; + public IReadOnlyList Keys => keys.ToArray(); + + public TGrainInterface GetGrain(string primaryKey, string? grainClassNamePrefix = null) + where TGrainInterface : IGrainWithStringKey + { + keys.Enqueue(primaryKey); + if (typeof(TGrainInterface) == typeof(IPodHubGrain)) + return (TGrainInterface)(object)podHub; + throw new NotSupportedException($"Unexpected grain interface {typeof(TGrainInterface)}"); + } + + public TGrainInterface GetGrain(Guid primaryKey, string? grainClassNamePrefix = null) + where TGrainInterface : IGrainWithGuidKey => throw new NotSupportedException(); + + public TGrainInterface GetGrain(long primaryKey, string? grainClassNamePrefix = null) + where TGrainInterface : IGrainWithIntegerKey => throw new NotSupportedException(); + + public TGrainInterface GetGrain(Guid primaryKey, string keyExtension, string? grainClassNamePrefix = null) + where TGrainInterface : IGrainWithGuidCompoundKey => throw new NotSupportedException(); + + public TGrainInterface GetGrain(long primaryKey, string keyExtension, string? grainClassNamePrefix = null) + where TGrainInterface : IGrainWithIntegerCompoundKey => throw new NotSupportedException(); + + public TGrainObserverInterface CreateObjectReference(IGrainObserver obj) + where TGrainObserverInterface : IGrainObserver => throw new NotSupportedException(); + + public void DeleteObjectReference(IGrainObserver obj) + where TGrainObserverInterface : IGrainObserver => throw new NotSupportedException(); + + public IGrain GetGrain(Type grainInterfaceType, Guid grainPrimaryKey) => throw new NotSupportedException(); + public IGrain GetGrain(Type grainInterfaceType, long grainPrimaryKey) => throw new NotSupportedException(); + public IGrain GetGrain(Type grainInterfaceType, string grainPrimaryKey) => throw new NotSupportedException(); + public IGrain GetGrain(Type grainInterfaceType, Guid grainPrimaryKey, string keyExtension) => throw new NotSupportedException(); + public IGrain GetGrain(Type grainInterfaceType, long grainPrimaryKey, string keyExtension) => throw new NotSupportedException(); + public TGrainInterface GetGrain(GrainId grainId) + where TGrainInterface : IAddressable => throw new NotSupportedException(); + public IAddressable GetGrain(GrainId grainId) => throw new NotSupportedException(); + public IAddressable GetGrain(GrainId grainId, GrainInterfaceType interfaceType) => throw new NotSupportedException(); + public IAddressable GetGrain(Type interfaceType, IdSpan grainKey, string grainClassNamePrefix) => throw new NotSupportedException(); + public IAddressable GetGrain(Type interfaceType, IdSpan grainKey) => throw new NotSupportedException(); + } + + /// + /// Captures what the router logged, at which level โ€” the assertion here is over the LEVEL, so a + /// null logger would make every claim in this file vacuous. + /// + private sealed class RecordingLogger : ILogger + { + private readonly ConcurrentQueue<(LogLevel Level, string Message)> records = new(); + + public IReadOnlyList<(LogLevel Level, string Message)> Records => records.ToArray(); + + public IDisposable BeginScope(TState state) where TState : notnull => NullScope.Instance; + + public bool IsEnabled(LogLevel logLevel) => true; + + public void Log(LogLevel logLevel, EventId eventId, TState state, Exception? exception, + Func formatter) + => records.Enqueue((logLevel, formatter(state, exception))); + + private sealed class NullScope : IDisposable + { + public static readonly NullScope Instance = new(); + public void Dispose() { } + } + } +} diff --git a/test/MeshWeaver.Hosting.Orleans.Test/PodHubNotHereRefusalTest.cs b/test/MeshWeaver.Hosting.Orleans.Test/PodHubNotHereRefusalTest.cs new file mode 100644 index 0000000000..fac5a9e8fe --- /dev/null +++ b/test/MeshWeaver.Hosting.Orleans.Test/PodHubNotHereRefusalTest.cs @@ -0,0 +1,215 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Reactive; +using System.Reactive.Linq; +using System.Reactive.Threading.Tasks; +using System.Text.Json; +using System.Threading.Tasks; +using MeshWeaver.Mesh; +using MeshWeaver.Messaging; +using Microsoft.Extensions.Logging; +using Xunit; + +namespace MeshWeaver.Hosting.Orleans.Test; + +/// +/// ๐Ÿšจ The Orleans memory-stream publish leaves the routing leg โ€” issues #2320 / #2322 / #2406, +/// made UNREACHABLE rather than fixed. +/// +/// The defect those three tickets name is Orleans-internal, and there is no MeshWeaver +/// line to change on that path. A publish into a stream with no live subscriber SUCCEEDS and +/// discards (the subscriber probe narrows this, but fails open by design); a publish into a stream +/// whose queue grain is wedged, or whose producer never registered, STALLS for the sender's whole +/// 30โ€“60 s budget. Durability cannot fix a request that should not have been queued. +/// +/// So the leg goes. RoutingGrain.BuildPodHubRoute used to take the stream +/// fallback whenever the pod-hub grain answered โ€” a condition +/// that covers BOTH "the owner is an Orleans client, which cannot host a grain" and "the owner is a +/// silo whose claim has not landed yet", and the exception cannot tell them apart. The fallback is +/// now taken ONLY for an address type DECLARED client-hosted +/// (); production declares none, so the +/// publish is unreachable there. Everything else is answered in ONE hop with a transient NACK the +/// requester's own recovery rides out, inside the directed call's budget instead of the stream's. +/// See Doc/Architecture/DurableStreamsViaMeshNodes. +/// +/// The decision is the subject. is +/// internal static and takes the stream leg as a thunk, so "no stream publish happened" is +/// asserted directly โ€” the thunk was never invoked โ€” rather than inferred from the absence of a +/// side effect. Pure: no cluster, no host, no clock. +/// +/// Fails on unfixed code: +/// โ€” the method does not +/// exist, and the arm it replaces published unconditionally. +/// +public class PodHubNotHereRefusalTest +{ + private static readonly Address Sender = new("client", "pod-hub-sender"); + + private static MeshConfiguration Declaring(params string[] clientHostedTypes) => + new(Array.Empty()) + { + ClientHostedAddressTypes = new HashSet(clientHostedTypes, StringComparer.Ordinal), + }; + + private static IMessageDelivery Delivery(string id, string targetType = "portal") => + new MessageDelivery( + Sender, new Address(targetType, "unclaimed"), + new RawJson("{\"$type\":\"DataChangedEvent\"}"), + JsonSerializerOptions.Default) with + { Id = id }; + + private sealed record Nack(string Message, ErrorType Type, bool TargetUnserved); + + /// + /// ๐Ÿšจ THE PIN. An address type nobody declared client-hosted gets a TRANSIENT verdict and NO + /// publish โ€” a publish that would have succeeded and discarded, or stalled for the sender's + /// whole budget. + /// + [Fact] + public async Task AnUndeclaredType_IsNackedTransiently_AndNeverReachesTheStream() + { + var nacks = new List(); + var published = 0; + + await RoutingGrain.AnswerPodHubNotHere( + Delivery("d-2320"), "portal/unclaimed", "portal", + Declaring(), + fallBackToStream: () => { published++; return Observable.Return(Unit.Default); }, + postFailureToSender: (m, t, u) => nacks.Add(new Nack(m, t, u)), + logger: new RecordingLogger()) + .ToTask(); + + published.Should().Be(0, + "the stream publish is the leg #2320/#2322/#2406 live on โ€” it must be UNREACHABLE for a " + + "type nobody declared client-hosted, not merely unlikely"); + + var nack = nacks.Should().ContainSingle().Subject; + nack.Type.Should().Be(ErrorType.ShuttingDown, + "SynchronizationStream's resubscribe latch and MeshNodeStreamCache.IsTransientOwnerFailure " + + "RIDE OUT ShuttingDown and TEAR DOWN on a terminal verdict โ€” and 'no silo serves this hub " + + "right now' is a lifecycle transition, resolved by the owner's own indefinite claim"); + nack.TargetUnserved.Should().BeTrue( + "this is the router's authoritative 'nobody serves that address' verdict, the same stamp " + + "the subscriber probe produces โ€” it is what lets an OWNER drop the server-side stream it " + + "can no longer push to, ending the #2426 fan-out-to-a-corpse loop"); + nack.Message.Should().Contain("portal/unclaimed"); + } + + /// + /// The other direction, and the reason the gate is a DECLARATION: for a type whose hubs live in + /// an Orleans client the stream is not a fallback, it is the only transport there is. NACKing + /// those would make every such hub permanently unreachable. + /// + [Fact] + public async Task ADeclaredClientHostedType_StillFallsBackToTheStream() + { + var nacks = new List(); + var published = 0; + + await RoutingGrain.AnswerPodHubNotHere( + Delivery("d-client", targetType: "client"), "client/unclaimed", "client", + Declaring("client"), + fallBackToStream: () => { published++; return Observable.Return(Unit.Default); }, + postFailureToSender: (m, t, u) => nacks.Add(new Nack(m, t, u)), + logger: new RecordingLogger()) + .ToTask(); + + published.Should().Be(1, + "an Orleans client cannot host a grain, so a directed call can never reach it โ€” refusing " + + "here would take the hub's only transport away"); + nacks.Should().BeEmpty("the delivery was carried, so there is nothing to report"); + } + + /// + /// The declaration is OPT-IN and EMPTY by default โ€” which is what makes "production declares + /// none, so the publish is unreachable in the fleet" a property of the code rather than of a + /// deployment's configuration. + /// + [Fact] + public void NothingIsClientHostedUnlessItSaysSo() + { + new MeshConfiguration(Array.Empty()).ClientHostedAddressTypes.Should().BeEmpty( + "no production process hosts mesh hubs as an Orleans client โ€” UseOrleansMeshClient has " + + "only test-fixture callers, the distributed portal is a co-hosted silo, and the " + + "monolith / LocalMesh / bake host run no Orleans at all"); + + // ...and the built-in stream-routed types are NOT client-hosted by association: being + // reachable over the stream says nothing about who hosts the hub. + var config = new MeshConfiguration(Array.Empty()); + foreach (var streamRouted in MeshConfiguration.DefaultStreamRoutedAddressTypes) + config.ClientHostedAddressTypes.Should().NotContain(streamRouted, + $"'{streamRouted}' is stream-ROUTED, which is a statement about the transport, not " + + "about whether a grain can be hosted for it"); + } + + /// + /// Bounded per delivery: one refusal is one emission that COMPLETES. No retry, no timer, no + /// watchdog anywhere on this path, so the refusal can never become its own storm. + /// + [Fact] + public async Task ARefusalIsTerminalPerDelivery() + { + var notifications = await RoutingGrain.AnswerPodHubNotHere( + Delivery("t1"), "portal/unclaimed", "portal", Declaring(), + fallBackToStream: () => throw new InvalidOperationException("must not publish"), + postFailureToSender: (_, _, _) => { }, + logger: new RecordingLogger(), + refusalLog: new DeadTargetRefusalLog(TimeSpan.FromSeconds(60))) + .Materialize() + .ToList() + .Timeout(TimeSpan.FromSeconds(5)) + .ToTask(); + + notifications.Select(n => n.Kind).Should().Equal(NotificationKind.OnNext, NotificationKind.OnCompleted); + } + + /// + /// The log window, same contract as RefuseNoSubscriber: a KNOWN-unreachable address must + /// not buy a full line per delivery (#2426/#2546 โ€” 20,718 lines in 3 h), while EVERY delivery is + /// still refused and still NACKed, because the NACK is each sender's terminal answer AND the + /// owner-side eviction signal. + /// + [Fact] + public async Task AHundredRefusalsOfOneAddress_ShipOneFullLine_AndAHundredNacks() + { + var logger = new RecordingLogger(); + var refusalLog = new DeadTargetRefusalLog(TimeSpan.FromSeconds(60)); + var nacks = new List(); + var config = Declaring(); + + for (var i = 0; i < 100; i++) + await RoutingGrain.AnswerPodHubNotHere( + Delivery($"d{i}"), "portal/unclaimed", "portal", config, + fallBackToStream: () => throw new InvalidOperationException("must not publish"), + postFailureToSender: (m, t, u) => nacks.Add(new Nack(m, t, u)), + logger: logger, refusalLog: refusalLog) + .ToTask(); + + nacks.Should().HaveCount(100, "the window bounds the LOG, never the answer"); + nacks.Should().OnlyContain(n => n.Type == ErrorType.ShuttingDown && n.TargetUnserved); + + logger.Records.Count(r => r.Level == LogLevel.Warning).Should().Be(1, + "one full line per address per window โ€” Warning rather than Error because the verdict is " + + "transient by construction and the owner's claim is still landing"); + logger.Records.Count(r => r.Level == LogLevel.Debug).Should().Be(99, + "the suppressed volume still leaves per-delivery evidence"); + } + + private sealed class RecordingLogger : ILogger + { + private readonly List<(LogLevel Level, string Message)> records = []; + public IReadOnlyList<(LogLevel Level, string Message)> Records => records; + public IDisposable BeginScope(TState state) where TState : notnull => NullScope.Instance; + public bool IsEnabled(LogLevel logLevel) => true; + public void Log(LogLevel logLevel, EventId eventId, TState state, Exception? exception, + Func formatter) + => records.Add((logLevel, formatter(state, exception))); + + private sealed class NullScope : IDisposable + { + public static readonly NullScope Instance = new(); + public void Dispose() { } + } + } +}