diff --git a/src/MeshWeaver.Connection.Orleans/README.md b/src/MeshWeaver.Connection.Orleans/README.md index 997de7f6b3..b7a16a7ecd 100644 --- a/src/MeshWeaver.Connection.Orleans/README.md +++ b/src/MeshWeaver.Connection.Orleans/README.md @@ -35,6 +35,9 @@ app.StartPortalApplication(); - Integrates with [Microsoft Orleans](https://learn.microsoft.com/en-us/dotnet/orleans/overview) client system - Supports Azure and other cloud providers for clustering +## Placement expiry +A delivery whose message outlives Orleans' response timeout while its target grain is still being placed is cancelled with `OperationCanceledException: Message expired before placement could complete for grain ...` (issue #5731; seen once on memex for an ephemeral `_Activity/compile-*` hub). The message reached no activation, so the sender is told a transient failure (`ErrorType.ShuttingDown`) on the first attempt and the delivery is not re-sent: `RoutingGrain.IsPlacementExpired` (in `MeshWeaver.Hosting.Orleans`) is the classifier and `PlacementExpiryClassificationTest` pins Orleans' wording. Whether the stall itself is general placement latency during cluster membership changes is not evidenced; watch log fingerprint `bb128e1afa6c6449` for recurrence. + ## See Also - [Orleans Documentation](https://learn.microsoft.com/en-us/dotnet/orleans/overview) - Learn more about Orleans clients - [Main MeshWeaver Documentation](../../Readme.md) - More about mesh connectivity options diff --git a/src/MeshWeaver.Hosting.Orleans/RoutingGrain.cs b/src/MeshWeaver.Hosting.Orleans/RoutingGrain.cs index 8b8bdca587..53b5a7f80c 100644 --- a/src/MeshWeaver.Hosting.Orleans/RoutingGrain.cs +++ b/src/MeshWeaver.Hosting.Orleans/RoutingGrain.cs @@ -1678,7 +1678,7 @@ internal static IObservable DeliverToGrainObservable( // Defer keeps grainCall COLD so every RetryWhen re-subscribe re-invokes it (fresh grain // reference → fresh activation). Never Observable.FromAsync — see AsynchronousCalls.md. - return Observable.Defer(() => grainCall().ToObservable()) + return ObserveGrainCall(grainCall) .RetryWhen(errors => errors .Select((ex, i) => (Exception: ex, Attempt: i)) .SelectMany(t => @@ -1796,6 +1796,8 @@ internal static ErrorType ClassifyDeliveryException( Exception ex, Func? scopeDisposed = null, bool activationErrorRecorded = true) => OrleansRoutingService.IsDirectoryUnstable(ex) || IsShutdownShaped(ex) + // The message expired while its target grain was still being placed (#5731): no activation ever saw it. + || IsPlacementExpired(ex) || IsDeactivatedActivation(ex, activationErrorRecorded) || IsScopeTeardown(ex, scopeDisposed) ? ErrorType.ShuttingDown @@ -2133,4 +2135,102 @@ internal static bool IsDepartedSiloRejection(Exception e) => /// A literal of Orleans.Runtime. /// internal const string SupersededSiloGenerationMarker = "The target silo is no longer active"; + + /// + /// Orleans' own wording from PlacementService.PlacementWorker.ExecutePlacementAsync: + /// "Message expired before placement could complete for grain {id}.", raised as an + /// when the message's time-to-live (Orleans' + /// ResponseTimeout) ran out while the grain was still being PLACED (issue #5731). Kept + /// as the verbatim prefix without the grain id, so it matches whichever grain the expiry names. + /// + /// A literal of Orleans.Runtime. PlacementExpiryClassificationTest pins it + /// against the shipped assembly, so a re-worded Orleans turns that test red instead of turning + /// silently inert. + /// + internal const string PlacementExpiredMarker = "Message expired before placement could complete"; + + /// + /// The delivery's message EXPIRED while Orleans was still placing its target grain - issue + /// #5731. Orleans' PlacementService gives up on a message whose time-to-live has run out, + /// and the caller receives an carrying + /// . Seen once on memex (2026-09-24 13:43:20Z): two + /// IMessageHubGrain.DeliverMessage calls to an ephemeral _Activity/compile-* hub. + /// + /// Why it is a transient verdict for the sender. The message died while being + /// ADDRESSED: no activation existed, so it reached no hub and nothing was half-applied. That is a + /// statement about the cluster (placement or the grain directory was slower than the message's + /// time-to-live, which is what a membership change produces), not about the target, and the + /// sender's recovery is bounded the same way describes. + /// Reported as it tore down consumers that ride out + /// . + /// + /// Classification only - the delivery is NOT sent again. Each re-attempt can itself + /// wait a full time-to-live inside placement while holding a routing-pool slot, so six retries + /// under a persistent stall would pin that slot for minutes - the amplification #1172 removed for + /// response timeouts. therefore does not match it, and the sender + /// is told on the first attempt. + /// + /// Narrow on purpose. A cancellation is also what a hub raises while it shuts down or + /// times out an activation, and those are different statements. Only the one Orleans sentence + /// qualifies, and only on an (an + /// quoting the words stays terminal). It is prose, so + /// PlacementExpiryClassificationTest pins it against the shipped Orleans build: if that + /// test fails after an upgrade, repair the marker and never delete the test. The exception reaches + /// this predicate with its text intact only because hands it + /// over - see there. + /// + /// The exception the delivery attempt faulted with. + /// true when the message expired before its target grain was placed. + internal static bool IsPlacementExpired(Exception ex) => + ExceptionChain.Contains(ex, static e => + e is OperationCanceledException + && e.Message.Contains(PlacementExpiredMarker, StringComparison.OrdinalIgnoreCase)); + + /// + /// The cold grain call as an observable, with a CANCELLED call's original exception kept intact - + /// issue #5731. Orleans completes a call it cannot address (a message that expired in placement) + /// with the carried , which ValueTask.AsTask turns + /// into a CANCELLED task that still stores it. Rx's task bridge then replaces that with a fresh + /// whose text is only "A task was canceled." - so every + /// classifier downstream, and the sender's NACK, lost the one sentence that says what happened. + /// + /// This hands over the stored exception instead. Anything else (a fault, a result, a + /// cancellation that stored nothing) passes through unchanged, and the retry around it is + /// unaffected: only the exception's identity changes, never whether it is retried. + /// + /// The grain call's result. + /// The grain call; re-invoked from scratch on every subscription. + /// A cold observable of the call's result. + internal static IObservable ObserveGrainCall(Func> grainCall) => + Observable.Defer(() => grainCall().ToObservable()) + .Catch(canceled => Observable.Throw(OriginalCancellation(canceled))); + + /// + /// The exception a cancelled task stored, or itself when it stored + /// none. The task is already complete, so reading it never blocks - it only re-throws what it + /// holds. + /// + /// The exception Rx made from the cancelled task. + /// The task's own cancellation, or . + internal static Exception OriginalCancellation(TaskCanceledException canceled) + { + var task = canceled.Task; + if (task is null || !task.IsCanceled) + return canceled; + + try + { + task.GetAwaiter().GetResult(); + } + catch (OperationCanceledException original) + { + return original; + } + catch (Exception) + { + // Not a cancellation after all: keep the exception Rx made. + } + + return canceled; + } } diff --git a/test/MeshWeaver.Hosting.Orleans.Test/PlacementExpiryClassificationTest.cs b/test/MeshWeaver.Hosting.Orleans.Test/PlacementExpiryClassificationTest.cs new file mode 100644 index 0000000000..a002ffa3d0 --- /dev/null +++ b/test/MeshWeaver.Hosting.Orleans.Test/PlacementExpiryClassificationTest.cs @@ -0,0 +1,209 @@ +using System; +using System.Collections.Generic; +using System.IO; +using System.Reactive.Concurrency; +using System.Text; +using System.Threading.Tasks; +using MeshWeaver.Messaging; +using Microsoft.Extensions.Logging.Abstractions; +using Xunit; +using MeshWeaver.Fixture; + +namespace MeshWeaver.Hosting.Orleans.Test; + +/// +/// Issue #5731 - a delivery whose message EXPIRED while Orleans was still placing its target grain. +/// +/// Orleans' PlacementService cancels such a message with an OperationCanceledException that +/// the caller receives as a CANCELLED task. Two things were wrong on our side: Rx's task bridge +/// replaced the stored exception with a message-less TaskCanceledException ("A task was canceled."), +/// so nothing downstream could recognise the condition, and the classifier would not have known it +/// anyway, so the sender was told a terminal Failed and consumers that ride out ShuttingDown tore +/// down. The facts here are pure and need no cluster; the exception text is quoted verbatim from the +/// production log. +/// +public class PlacementExpiryClassificationTest +{ + private static readonly Func NoBackoff = _ => TimeSpan.Zero; + + /// + /// Verbatim from the production log (#5731, 2026-09-24 13:43:20Z, Orleans.Messaging[100071]). + /// + private const string PlacementExpiredText = + "Message expired before placement could complete for grain " + + "messagehub/Signature/Desk/_Activity/compile-20260924134235046d3415968dc0b4b9cad1eddb1be5b8bb6."; + + /// + /// The SHAPE Orleans hands the caller: an async completion that throws an OperationCanceledException + /// ends as a CANCELLED task that still stores it - the same state ValueTask.AsTask produces for a + /// carried cancellation. + /// + private static async Task ExpiredBeforePlacement() + { + await Task.Yield(); + throw new OperationCanceledException(PlacementExpiredText); + } + + private static async Task CancelledForSomeOtherReason() + { + await Task.Yield(); + throw new OperationCanceledException("The operation was canceled."); + } + + [Fact] + public async Task APlacementExpiry_ReachesTheClassifierWithItsOwnText_AndIsNotResent() + { + var calls = 0; + Exception? caught = null; + + try + { + await RoutingGrain.DeliverToGrainObservable( + grainCall: () => + { + calls++; + return ExpiredBeforePlacement(); + }, + grainKey: "messagehub/Signature/Desk/_Activity/compile-1", + deliveryId: "placement-expiry-1", + logger: NullLogger.Instance, + backoff: NoBackoff, + scheduler: Scheduler.Immediate) + .Await(TestContext.Current.CancellationToken); + } + catch (Exception ex) + { + caught = ex; + } + + Assert.NotNull(caught); + calls.Should().Be(1, + "the message died while being addressed; sending it again would wait out another full " + + "time-to-live inside placement while holding a routing-pool slot, so the sender is told " + + "on the first attempt"); + RoutingGrain.IsPlacementExpired(caught).Should().BeTrue( + "Rx's task bridge turns a cancelled task into a message-less TaskCanceledException, and the " + + "grain call must hand over the exception the task stored instead - otherwise no classifier " + + "downstream can see the Orleans sentence"); + } + + [Fact] + public async Task APlacementExpiry_IsNackedAsTransient_NotTerminal() + { + var nacks = new List<(string Message, ErrorType Type)>(); + + await RoutingGrain.DeliverToGrainRoute( + grainCall: ExpiredBeforePlacement, + grainKey: "messagehub/Signature/Desk/_Activity/compile-1", + addressPath: "Signature/Desk/_Activity/compile-1", + deliveryId: "placement-expiry-2", + postFailureToSender: (message, type) => nacks.Add((message, type)), + logger: NullLogger.Instance, + backoff: NoBackoff, + scheduler: Scheduler.Immediate) + .Await(TestContext.Current.CancellationToken); + + nacks.Count.Should().Be(1, "the route answers the sender exactly once"); + nacks[0].Type.Should().Be(ErrorType.ShuttingDown, + "the message reached no activation - a statement about the cluster, not the target - and the " + + "consumers with their own recovery machinery ride out ShuttingDown and tear down on Failed"); + nacks[0].Message.Contains(RoutingGrain.PlacementExpiredMarker, StringComparison.OrdinalIgnoreCase) + .Should().BeTrue("the sender's NACK now names what happened instead of 'A task was canceled.'"); + } + + [Fact] + public void APlacementExpiry_IsClassifiedTransient_ButNeverResent() + { + var expired = new OperationCanceledException(PlacementExpiredText); + + RoutingGrain.ClassifyDeliveryException(expired).Should().Be(ErrorType.ShuttingDown); + RoutingGrain.ClassifyPodHubDeliveryException(expired).Should().Be(ErrorType.ShuttingDown, + "the pod-hub leg classifies through the same arms"); + RoutingGrain.IsResendableDeliveryFailure(expired).Should().BeFalse( + "classification only: a re-send under a persistent placement stall would hold a routing-pool " + + "slot for a full time-to-live per attempt"); + } + + [Fact] + public void ItIsFoundBesideAnotherFaultInAnAggregate() + { + var twoTransports = new AggregateException( + new InvalidOperationException("the NACK's other transport also failed"), + new OperationCanceledException(PlacementExpiredText)); + + RoutingGrain.ClassifyDeliveryException(twoTransports).Should().Be(ErrorType.ShuttingDown, + "which fault sits at index 0 of an aggregate is a race, so the walk reads the whole graph"); + } + + [Fact] + public async Task AnyOtherCancellation_StaysTerminal() + { + RoutingGrain.ClassifyDeliveryException(new OperationCanceledException("The operation was canceled.")) + .Should().Be(ErrorType.Failed, + "a hub raising a cancellation while it shuts down or times out is a different statement " + + "from an expiry in placement - only Orleans' one sentence qualifies"); + + var nacks = new List<(string Message, ErrorType Type)>(); + + await RoutingGrain.DeliverToGrainRoute( + grainCall: CancelledForSomeOtherReason, + grainKey: "messagehub/Signature/Desk/_Activity/compile-1", + addressPath: "Signature/Desk/_Activity/compile-1", + deliveryId: "placement-expiry-3", + postFailureToSender: (message, type) => nacks.Add((message, type)), + logger: NullLogger.Instance, + backoff: NoBackoff, + scheduler: Scheduler.Immediate) + .Await(TestContext.Current.CancellationToken); + + nacks.Count.Should().Be(1); + nacks[0].Type.Should().Be(ErrorType.Failed, + "anything this does not recognise stays terminal, so a genuine defect is still reported as one"); + } + + [Fact] + public void TheMarkerQuotedByAnythingButACancellation_StaysTerminal() + { + RoutingGrain.ClassifyDeliveryException(new InvalidOperationException(PlacementExpiredText)) + .Should().Be(ErrorType.Failed, + "the words alone are not the signal: the type guard is what keeps an unrelated fault that " + + "quotes them from being demoted to a transient verdict"); + } + + /// + /// THE ANTI-INERT PIN. If this fact fails after an Orleans upgrade, IsPlacementExpired has gone + /// SILENTLY INERT - repair the marker, never delete the test. The phrase is a literal of the SHIPPED + /// Orleans.Runtime (PlacementService lives there), not a copy of it. + /// + [Fact] + public void PlacementExpiredMarker_IsStillTheShippedOrleansWording() + { + var runtimeAssembly = Path.Combine(AppContext.BaseDirectory, "Orleans.Runtime.dll"); + + File.Exists(runtimeAssembly).Should().BeTrue( + $"this fact reads Orleans' own string literals out of {runtimeAssembly}; without the file it " + + "would pass having verified nothing, which is precisely the failure mode it exists to prevent"); + + ShippedLiteralPresent(runtimeAssembly, RoutingGrain.PlacementExpiredMarker).Should().BeTrue( + $"'{RoutingGrain.PlacementExpiredMarker}' is the phrase RoutingGrain.IsPlacementExpired matches " + + "an expiry in Orleans' PlacementService on. If Orleans reworded it, a delivery that expired " + + "before its target was placed is NACK'd as permanent again (#5731). REPAIR THE MARKER; do not " + + "delete this"); + } + + /// + /// Reads the #US heap at BOTH byte alignments - a managed string literal's UTF-16 payload follows a + /// compressed-integer length, so a blob can begin at an ODD file offset and decoding from offset 0 + /// alone would miss a phrase that is genuinely present (a false red on the one assertion whose value + /// is being believed when it fires). OrdinalIgnoreCase, matching the classifier. + /// + private static bool ShippedLiteralPresent(string assemblyPath, string phrase) + { + var raw = File.ReadAllBytes(assemblyPath); + + return Encoding.Unicode.GetString(raw) + .Contains(phrase, StringComparison.OrdinalIgnoreCase) + || Encoding.Unicode.GetString(raw, 1, raw.Length - 1) + .Contains(phrase, StringComparison.OrdinalIgnoreCase); + } +} \ No newline at end of file diff --git a/test/MeshWeaver.Hosting.Orleans.Test/PlacementExpiryThroughTheRouteTest.cs b/test/MeshWeaver.Hosting.Orleans.Test/PlacementExpiryThroughTheRouteTest.cs new file mode 100644 index 0000000000..60aa291c58 --- /dev/null +++ b/test/MeshWeaver.Hosting.Orleans.Test/PlacementExpiryThroughTheRouteTest.cs @@ -0,0 +1,300 @@ +using System; +using System.Linq; +using System.Reactive; +using System.Reactive.Disposables; +using System.Reactive.Linq; +using System.Reactive.Subjects; +using System.Reactive.Threading.Tasks; +using System.Threading; +using System.Threading.Tasks; +using MeshWeaver.Data; +using MeshWeaver.Fixture; +using MeshWeaver.Mesh; +using MeshWeaver.Mesh.Security; +using MeshWeaver.Mesh.Services; +using MeshWeaver.Messaging; +using Microsoft.Extensions.DependencyInjection; +using Orleans; +using Orleans.Hosting; +using Orleans.TestingHost; +using Xunit; + +namespace MeshWeaver.Hosting.Orleans.Test; + +/// +/// Issue #5731, end to end: a delivery to an ephemeral compile-activity hub +/// ({owner}/_Activity/compile-{id}) whose message EXPIRES while Orleans is still placing the +/// target grain must reach its caller as ONE transient +/// - no hang, no unhandled error, no silent loss, no second attempt. +/// +/// The rig. An Orleans outgoing grain-call filter on the silo, armed for one test's +/// grain keys, answers IMessageHubGrain.DeliverMessage by throwing Orleans' own +/// from an async completion - a CANCELLED task that stores +/// the exception, the shape PlacementService hands the caller. The call never leaves the caller, so +/// no activation exists, exactly as in production. Everything else is real: the issuing hub's +/// Observe, the silo's routing service, 's route, its classifier +/// and its NACK back to the sender. +/// +/// Negative controls. Any other cancellation through the same rig stays terminal +/// (); and the pure fact at the bottom shows the pre-fix task bridge +/// loses Orleans' sentence, so without the exception preservation and the classifier arm the first +/// test's ShuttingDown assertion fails. +/// +public class PlacementExpiryThroughTheRouteTest(ITestOutputHelper output) : OrleansMeshTestBase(output) +{ + /// + protected override Type SiloConfiguratorType => typeof(PlacementExpirySiloConfigurator); + + private const string Marker = "Message expired before placement could complete"; + + private IMessageHub SiloMesh => + ((InProcessSiloHandle)Cluster.Silos[0]).SiloHost.Services.GetRequiredService(); + + [Fact(Timeout = 120_000)] + public async Task APlacementExpiry_OnTheWayToACompileActivity_ReachesTheCallerOnce_AsShuttingDown() + { + using var deadline = CancellationTokenSource.CreateLinkedTokenSource(TestContext.Current.CancellationToken); + deadline.CancelAfter(TimeSpan.FromSeconds(110)); + + var outcome = await DeliverToACompileActivityUnder( + target => $"{Marker} for grain {target}.", deadline.Token); + + outcome.Kind.Should().Be(NotificationKind.OnError, + "a delivery that died in placement must be REPORTED to its caller - completing or answering " + + "it would be a silent loss"); + var failure = FindDeliveryFailure(outcome.Error); + Assert.True(failure is not null, + $"the caller must see a DeliveryFailure, not an unhandled error; got {outcome.Error}"); + failure!.Failure.ErrorType.Should().Be(ErrorType.ShuttingDown, + "the message reached no activation - a statement about the cluster, not the target - and " + + "consumers that ride out ShuttingDown tear down on Failed (#5731)"); + (failure.Failure.Message ?? failure.Message).Contains(Marker, StringComparison.OrdinalIgnoreCase) + .Should().BeTrue("the caller is told what happened, not 'A task was canceled.'"); + outcome.Attempts.Should().Be(1, + "classification only: a re-send under a persistent placement stall would hold a " + + "routing-pool slot for a full time-to-live per attempt"); + Assert.All(outcome.Targets, t => Assert.Contains("/_Activity/compile-", t)); + } + + [Fact(Timeout = 120_000)] + public async Task AnyOtherCancellation_OnTheSameRoute_StaysTerminal() + { + using var deadline = CancellationTokenSource.CreateLinkedTokenSource(TestContext.Current.CancellationToken); + deadline.CancelAfter(TimeSpan.FromSeconds(110)); + + var outcome = await DeliverToACompileActivityUnder(_ => "The operation was canceled.", deadline.Token); + + outcome.Kind.Should().Be(NotificationKind.OnError); + var failure = FindDeliveryFailure(outcome.Error); + Assert.True(failure is not null, + $"the caller must see a DeliveryFailure, not an unhandled error; got {outcome.Error}"); + failure!.Failure.ErrorType.Should().Be(ErrorType.Failed, + "only Orleans' placement-expiry sentence is demoted to transient; anything else stays terminal"); + outcome.Attempts.Should().Be(1); + } + + /// + /// The negative control of the exception preservation: the pre-fix bridge + /// (Observable.Defer(() => call().ToObservable())) replaces the cancelled task's stored + /// exception with a message-less TaskCanceledException, so the classifier answers Failed; the + /// route's own bridge keeps Orleans' sentence and answers ShuttingDown. + /// + [Fact] + public async Task WithoutThePreservation_TheSentenceIsLost_AndTheVerdictIsTerminal() + { + var ct = TestContext.Current.CancellationToken; + + Exception? viaOldBridge = null; + try + { + await Observable.Defer(() => ExpiredBeforePlacement().ToObservable()).Await(ct); + } + catch (Exception e) + { + viaOldBridge = e; + } + + Assert.NotNull(viaOldBridge); + RoutingGrain.IsPlacementExpired(viaOldBridge!).Should().BeFalse( + "Rx's task bridge drops the stored exception - this is what the fix repairs"); + RoutingGrain.ClassifyDeliveryException(viaOldBridge!).Should().Be(ErrorType.Failed); + + Exception? viaRoute = null; + try + { + await RoutingGrain.DeliverToGrainObservable( + grainCall: ExpiredBeforePlacement, + grainKey: "messagehub/placement-expiry/x/_Activity/compile-x", + deliveryId: "placement-expiry-control", + logger: Microsoft.Extensions.Logging.Abstractions.NullLogger.Instance, + backoff: _ => TimeSpan.Zero, + scheduler: System.Reactive.Concurrency.Scheduler.Immediate) + .Await(ct); + } + catch (Exception e) + { + viaRoute = e; + } + + Assert.NotNull(viaRoute); + RoutingGrain.IsPlacementExpired(viaRoute!).Should().BeTrue(); + RoutingGrain.ClassifyDeliveryException(viaRoute!).Should().Be(ErrorType.ShuttingDown); + } + + private static async Task ExpiredBeforePlacement() + { + await Task.Yield(); + throw new OperationCanceledException( + $"{Marker} for grain messagehub/placement-expiry/x/_Activity/compile-x."); + } + + private sealed record Outcome(NotificationKind Kind, Exception? Error, int Attempts, string[] Targets); + + /// + /// Seeds an owner node and its compile activity, arms the expiry for this test's keys only + /// (AFTER the seeding, whose own deliveries must go through), then issues a request to the + /// activity and returns the caller's whole first notification plus the attempts the rig saw. + /// + private async Task DeliverToACompileActivityUnder( + Func cancellationText, CancellationToken ct) + { + var access = SiloMesh.ServiceProvider.GetRequiredService(); + var meshService = SiloMesh.ServiceProvider.GetRequiredService(); + var issuing = SiloMesh.NodeOperationIssuingHub(); + var injector = Cluster.SiloServices().GetRequiredService(); + + var id = Guid.NewGuid().ToString("N"); + var owner = $"placement-expiry/{id}"; + var activity = $"{owner}/_Activity/compile-{id}"; + + await access.RunAsSystem(() => meshService.CreateNode(MeshNode.FromPath(owner) with + { + Name = "The owner of a compile activity", + NodeType = "Markdown", + State = MeshNodeState.Active, + })).FirstAsync().Await(ct); + await access.RunAsSystem(() => meshService.CreateNode(MeshNode.FromPath(activity) with + { + Name = "An ephemeral compile activity", + NodeType = "Activity", + MainNode = owner, + State = MeshNodeState.Active, + })).FirstAsync().Await(ct); + + using var armed = injector.Arm(id, cancellationText); + + var outcome = access.RunAsSystem(() => issuing.Observe( + new GetDataRequest(new MeshNodeReference()), o => o.WithTarget(new Address(activity)))) + .Take(1) + .Materialize() + .Replay(); + using var connected = outcome.Connect(); + + var first = await outcome.Should().Within(TestTimeouts.Convergence).Emit( + "the caller of a delivery that expired in placement must hear about it - a hang here is " + + "the silent loss #5731 asks about", ct); + var attempts = injector.Attempts; + + await injector.Attempted.Skip(1).Should().NotEmit(TimeSpan.FromSeconds(3), + "the route answers the sender once and does not forward the delivery again", ct); + + Output.WriteLine($"caller saw {first.Kind}: {first.Exception?.Message}; attempts {attempts}; " + + $"targets {string.Join(", ", injector.Targets)}"); + return new Outcome(first.Kind, first.Exception, attempts, injector.Targets); + } + + private static DeliveryFailureException? FindDeliveryFailure(Exception? error) + { + for (var e = error; e is not null; e = e.InnerException) + { + if (e is DeliveryFailureException failure) + return failure; + if (e is AggregateException aggregate) + return aggregate.Flatten().InnerExceptions.Select(FindDeliveryFailure) + .FirstOrDefault(f => f is not null); + } + return null; + } +} + +/// +/// The silo-side switch of : while armed, every +/// DeliverMessage to a grain whose key carries the armed fragment is failed the way Orleans' +/// PlacementService fails an expired message, and counted. Instance state on a silo singleton - +/// never static - so it dies with the cluster. +/// +public sealed class PlacementExpiryInjector +{ + private sealed record Arming(string Fragment, Func Text); + + private Arming? arming; + private int attempts; + private readonly ReplaySubject attempted = new(); + private readonly System.Collections.Concurrent.ConcurrentQueue targets = new(); + + /// The intercepted deliveries so far. + public int Attempts => Volatile.Read(ref attempts); + + /// Emits the target key of every intercepted delivery. + public IObservable Attempted => attempted; + + /// The target keys of the intercepted deliveries. + public string[] Targets => targets.ToArray(); + + /// Arms the expiry for grain keys containing . + /// A fragment unique to one test's grain keys. + /// The cancellation's message for a target key. + /// The disarm. + public IDisposable Arm(string fragment, Func text) + { + var armedNow = new Arming(fragment, text); + Volatile.Write(ref arming, armedNow); + return Disposable.Create(() => Interlocked.CompareExchange(ref arming, null, armedNow)); + } + + internal Exception? ExpiryFor(string targetId, string methodName) + { + var current = Volatile.Read(ref arming); + if (current is null + || methodName != "DeliverMessage" + || !targetId.Contains(current.Fragment, StringComparison.Ordinal)) + return null; + + Interlocked.Increment(ref attempts); + targets.Enqueue(targetId); + attempted.OnNext(targetId); + return new OperationCanceledException(current.Text(targetId)); + } +} + +/// +/// Fails an armed delivery from an ASYNC completion, so the caller gets a cancelled task that +/// stores the exception - the shape Orleans hands it when a message expires before placement. +/// +internal sealed class PlacementExpiryFilter(PlacementExpiryInjector injector) : IOutgoingGrainCallFilter +{ + /// + public async Task Invoke(IOutgoingGrainCallContext context) + { + if (injector.ExpiryFor(context.TargetId.ToString(), context.InterfaceMethod?.Name ?? string.Empty) + is { } expiry) + { + await Task.Yield(); + throw expiry; + } + + await context.Invoke(); + } +} + +/// The stock silo with on its outgoing grain calls. +public class PlacementExpirySiloConfigurator : SharedSiloConfigurator, ISiloConfigurator +{ + /// + void ISiloConfigurator.Configure(ISiloBuilder siloBuilder) + { + Configure(siloBuilder); + siloBuilder.Services.AddSingleton(); + siloBuilder.Services.AddSingleton(); + } +}