Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions src/MeshWeaver.Hosting/Persistence/PersistenceExtensions.cs
Original file line number Diff line number Diff line change
Expand Up @@ -831,6 +831,15 @@ public static IServiceCollection AddMeshCatalog(this IServiceCollection services
// waiting" and would restore the bug in a form no test could see.
services.TryAddSingleton<ILatePatchVerdictSink>(sp =>
sp.GetRequiredService<LatePatchResponseRegistry>());
// 🚨 …and the SAME instance again under the seam the ROUTING layer can see. When a reply
// cannot be forwarded at all — a whole-mesh teardown where the parent is already past
// DisposeHostedHubs — HierarchicalRouting offers it here instead of dropping it with
// nobody told (the shape NackReachesTheWaiterDuringTeardownTest keeps catching:
// "cannot route PatchDataResponse to cache/…" while the caller sits armed). A FACTORY over
// the same concrete singleton, for the same reason as the line above: a second instance
// would take the reply into a registry nobody armed.
services.TryAddSingleton<MeshWeaver.Messaging.IUndeliverableReplySink>(sp =>
sp.GetRequiredService<LatePatchResponseRegistry>());
// Post-commit durable-flush hook for the PatchDataRequest handler: chains the
// owner's ack off the storage write so a cross-hub stream.Update completion
// guarantees read-after-write (see IPostCommitFlush / StoragePostCommitFlush).
Expand Down
33 changes: 32 additions & 1 deletion src/MeshWeaver.Mesh.Contract/LatePatchWriteWatch.cs
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,7 @@ namespace MeshWeaver.Mesh;
/// write the owner had refused. <see cref="DispatchFailure"/> is that missing seam, and it obeys
/// the identical exactly-once rule.</para>
/// </summary>
public sealed class LatePatchResponseRegistry : ILatePatchVerdictSink
public sealed class LatePatchResponseRegistry : ILatePatchVerdictSink, IUndeliverableReplySink
{
/// <summary>
/// 🚨 How long after the patch post a late owner response is still acted upon. A response
Expand Down Expand Up @@ -285,6 +285,37 @@ public bool IsAdmissible(string requestId)
&& entry.ExpiresAt >= clock.GetUtcNow();

/// <summary>Number of armed watches — test seam.</summary>
/// <summary>
/// <see cref="IUndeliverableReplySink"/>: take a reply the routing layer is about to DROP,
/// when a caller here is armed for that correlation.
///
/// <para>🚨 The last resort, never a second transport. Routing offers a delivery here only
/// where it cannot forward it at all — a whole-mesh teardown past <c>DisposeHostedHubs</c> —
/// so there is no post left for this to race. Serving an armed caller ALONGSIDE a healthy post
/// was tried twice at the patch-verdict seam and reverted both times: the sink dispatches
/// synchronously on the owner's turn, ahead of the state the ack is about
/// (<c>ComboGateRollTest</c>, <c>ImportTypeBeforeInstanceTest</c>).</para>
///
/// <para>The two shapes a dropped reply takes are the two this registry already knows —
/// the verdict itself and a <see cref="DeliveryFailure"/> standing in for one. Anything else
/// is declined, so the drop stands and is logged as before.</para>
/// </summary>
public bool TryDeliver(IMessageDelivery delivery)
{
if (delivery.Properties.TryGetValue(PostOptions.RequestId, out var raw)
&& raw?.ToString() is { Length: > 0 } requestId)
{
return delivery.Message switch
{
PatchDataResponse response => Dispatch(requestId, response),
DeliveryFailure failure => DispatchFailure(requestId, failure),
_ => false,
};
}

return false;
}

public int ArmedCount => entries.Count;

/// <summary>
Expand Down
46 changes: 46 additions & 0 deletions src/MeshWeaver.Messaging.Hub/HierarchicalRouting.cs
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,41 @@ internal HierarchicalRouting(IMessageHub hub, IMessageHub? parentHub)



/// <summary>
/// Offers a correlated reply this router is about to DROP to a local waiter
/// (<see cref="IUndeliverableReplySink"/>). Only a delivery carrying
/// <see cref="PostOptions.RequestId"/> qualifies — fire-and-forget traffic has nobody waiting,
/// and answering it is the storm shape every NACK guard here exists to avoid.
/// </summary>
/// <returns><c>true</c> when a waiter took it, so the sender HAS its answer.</returns>
private bool TryHandOverUndeliverableReply(IMessageDelivery delivery)
{
if (!delivery.Properties.ContainsKey(PostOptions.RequestId))
return false;
try
{
var sink = hub.ServiceProvider.GetService<IUndeliverableReplySink>();
if (sink is null || !sink.TryDeliver(delivery))
return false;
logger.LogDebug(
"Undeliverable {MessageType} (ID: {MessageId}) for request {RequestId} was handed to "
+ "a local waiter instead of being dropped in {Address}",
delivery.Message?.GetType().Name, delivery.Id,
delivery.Properties[PostOptions.RequestId], hub.Address);
return true;
}
catch (Exception ex)
{
// The hand-over is a last resort; a sink that throws must not turn a classified drop
// into an unhandled fault on the routing path.
logger.LogWarning(ex,
"The undeliverable-reply sink threw for {MessageType} (ID: {MessageId}) in {Address}; "
+ "the delivery is dropped as before",
delivery.Message?.GetType().Name, delivery.Id, hub.Address);
return false;
}
Comment on lines +41 to +62
}

/// <summary>
/// Loops through forward rules in a sequence. Each forward rule either applies and returns delivery.Forwarded() or doesn't apply and returns delivery.
/// </summary>
Expand Down Expand Up @@ -135,6 +170,9 @@ private IMessageDelivery RouteAlongHostingHierarchy(IMessageDelivery delivery)
// and UNANSWERED — MessageService owes the sender a DeliveryFailure (a bare
// Failed(...) used to be dropped on the not-on-target path and the caller waited
// forever). TRANSIENT (ShuttingDown): the address may reactivate on the next probe.
if (isDisposing && TryHandOverUndeliverableReply(delivery))
return delivery.Processed();

return isDisposing
? delivery.Failed(errorMessage, ErrorType.ShuttingDown)
: delivery.NotFound();
Expand Down Expand Up @@ -202,6 +240,14 @@ private IMessageDelivery RouteAlongHostingHierarchy(IMessageDelivery delivery)
// and SynchronizationStream's transient check — still recognise a transient hub reject by
// that substring, so a NACK phrased any other way would be filed as a PERMANENT owner
// failure and cached as one.
// 🚨 …but if this is a REPLY somebody HERE is waiting for, hand it over before dropping
// it. The owner answered correctly and its post was accepted; only the route out died,
// and the waiter is in this same process with its registry entry armed. See
// IUndeliverableReplySink — offered ONLY here, where no post can reach the caller any
// more, never alongside a healthy one.
if (TryHandOverUndeliverableReply(delivery))
return delivery.Processed();

return delivery.Failed(
$"Hub {hub.Address} cannot route {delivery.Message.GetType().Name} to {delivery.Target} — "
+ $"its parent hub {parentHub.Address} is shutting down (RunLevel={parentHub.RunLevel}). "
Expand Down
39 changes: 39 additions & 0 deletions src/MeshWeaver.Messaging.Hub/IUndeliverableReplySink.cs
Original file line number Diff line number Diff line change
@@ -0,0 +1,39 @@
namespace MeshWeaver.Messaging;

/// <summary>
/// A last-resort taker for a CORRELATED REPLY that routing cannot deliver — the answer to a request
/// somebody in THIS process is still waiting for, which the hierarchy is about to drop.
///
/// <para>🚨 <b>Why this exists.</b> During a whole-mesh teardown the owner answers its request
/// normally: it posts the reply, its own <c>Post</c> accepts it (its run level is still open), and
/// one turn later <see cref="HierarchicalRouting"/> finds the parent past
/// <c>DisposeHostedHubs</c> and drops it — classified, logged, and told to nobody:</para>
/// <code>
/// Message delivery failed for PatchDataResponse (ID: l9ret6X…) in TestData/teardown-nack-node:
/// Hub TestData/teardown-nack-node cannot route PatchDataResponse to cache/VLe3ZpQ…
/// </code>
/// <para>The waiter is in the SAME process — its registry entry is armed and could take the reply
/// directly — but the routing layer has no way to reach it, so the caller burns its whole budget in
/// silence. Measured repeatedly on <c>NackReachesTheWaiterDuringTeardownTest</c> (core CI runs
/// 33861078925 and 33865385033, and locally under load), which is precisely this shape.</para>
///
/// <para>🚨 <b>This is NOT a second transport for live traffic.</b> It is offered ONLY where the
/// delivery is being dropped — where a post can no longer reach the waiter by any route. Serving it
/// alongside a healthy post reorders the answer ahead of the state it acknowledges: tried twice on
/// the patch-verdict seam and reverted both times (<c>ComboGateRollTest</c>,
/// <c>ImportTypeBeforeInstanceTest</c>). Here there is no post left to race.</para>
///
/// <para>Implemented by the mesh's late-verdict registry and resolved from the hub's service
/// provider; a mesh without one simply drops as before.</para>
/// </summary>
public interface IUndeliverableReplySink
{
/// <summary>
/// Offer <paramref name="delivery"/> — a reply carrying <c>PostOptions.RequestId</c> — to a
/// local waiter armed for that correlation.
/// </summary>
/// <param name="delivery">The reply routing is about to drop.</param>
/// <returns><c>true</c> when a local waiter took it, so the caller HAS been answered;
/// <c>false</c> when nobody here was waiting, and the drop stands.</returns>
bool TryDeliver(IMessageDelivery delivery);
}
29 changes: 26 additions & 3 deletions test/MeshWeaver.Messaging.Hub.Test/DisposalRaceNackTest.cs
Original file line number Diff line number Diff line change
Expand Up @@ -122,11 +122,33 @@ await handlerEntered.Should().Within(20.Seconds()).Emit(
// 2. Accepted while the hub is healthy, so it lands in the MAIN queue behind the
// stall — not in the deferred queue the #672 drain answers for.
IRequest<RaceResponse> request = faulting ? new FaultingRequest() : new RaceRequest();
// 🚨 The messageId OVERLOAD, not `o.WithMessageId(...)`: `Observe<TResponse>` mints the
// correlation id itself, so an id set through the options never becomes the ledger key —
// the first version of this wait polled a key that did not exist and timed out at 36 s
// having proved nothing.
var requestId = Guid.NewGuid().ToString("N");
var response = host
.Observe<RaceResponse>(request, o => o.WithTarget(victimAddress))
.Observe((object)request, o => o.WithTarget(victimAddress), requestId)!
.FirstAsync()
.Await(TestContext.Current.CancellationToken);

// 🚨 …and "accepted" is OBSERVED, never assumed. `Observe` posts on the HOST, whose own
// turn then routes the delivery down to the victim: between the two, the victim can
// pass Quiescing and reach DisposeHostedHubs, where the intake gate rejects it —
// "Hub disposal-race/1 is shutting down (RunLevel=DisposeHostedHubs …) — cannot
// process RaceRequest … Rejecting now."
// — which is CORRECT behaviour for a request that arrived after the door closed, and
// has nothing to do with what this test is about. It failed exactly that way once the
// suite grew a class and the scheduling shifted (2026-09-04), i.e. it was a race the
// whole time, decided by whether the host's routing turn beat Dispose(). The fate
// ledger already records the stage this test needs, so wait for it.
await Observable.Interval(TimeSpan.FromMilliseconds(20)).StartWith(0L)
.Select(_ => host.DescribeRequestFate(requestId))
.Where(trail => trail.Contains($"ENQUEUED@{victimAddress}", StringComparison.Ordinal))
.FirstAsync()
.Timeout(TestTimeouts.Convergence)
.Await(TestContext.Current.CancellationToken);

// 3. Dispose. The blocker observes the shutdown and returns; the queue behind it —
// our request, then the ShutdownRequest — runs in order.
victim!.Dispose();
Expand All @@ -138,8 +160,9 @@ await handlerEntered.Should().Within(20.Seconds()).Emit(
// one would leave this task pending for the [Fact] timeout with no explanation —
// exactly the production symptom (an install that spins for 60 s).
var served = await response;
served.Should().NotBeNull("the request was accepted, so it runs and is answered "
+ "before the hub goes down — the shutdown waits its turn behind accepted work");
served.Message.Should().BeOfType<RaceResponse>(
"the request was accepted, so it runs and is answered before the hub goes down "
+ "— the shutdown waits its turn behind accepted work");
await victim.DisposalCompleted.FirstOrDefaultAsync().Await()
.WaitAsync(TestTimeouts.Convergence);
victim.RunLevel.Should().Be(MessageHubRunLevel.Dead);
Expand Down
40 changes: 32 additions & 8 deletions test/MeshWeaver.PluginTester.Test/CascadeTest.cs
Original file line number Diff line number Diff line change
Expand Up @@ -149,19 +149,43 @@ public async Task IndependentNodesRunInParallel_BoundedByTheSlotCount()
(_, _) =>
{
var now = Interlocked.Increment(ref inFlight);
int seen;
do
try
{
seen = Volatile.Read(ref peak);
} while (now > seen && Interlocked.CompareExchange(ref peak, now, seen) != seen);
Thread.Sleep(80);
Interlocked.Decrement(ref inFlight);
return (0, true);
int seen;
do
{
seen = Volatile.Read(ref peak);
} while (now > seen && Interlocked.CompareExchange(ref peak, now, seen) != seen);
// 🚨 OBSERVE the overlap; do not infer it from a sleep. This body used to
// `Thread.Sleep(80)` and trust that two of the three slots would land inside it
// at once. On a starved runner they need not: the scheduler can admit, run and
// release one node before the next is dispatched, `peak` stays 1, and
// "nodes with no edge between them run at the same time" fails having measured
// the RUNNER's load rather than the cascade's parallelism (core CI run
// 33878671896, shard 3 — green on three sibling PRs and on main's previous
// three runs, and 10/10 locally, i.e. exactly the load-sensitivity signature).
//
// Now the first node in waits until a SECOND is inside with it. If the cascade
// really does admit more than one, both see it immediately and neither waits;
// if it serialises, the wait expires and the assertion below fails for the
// reason it is written for. The bound is per node and generous, so a genuinely
// serial scheduler costs the test a few seconds rather than hanging it.
SpinWait.SpinUntil(() => Volatile.Read(ref inFlight) > 1, TimeSpan.FromSeconds(2));
return (0, true);
}
finally
{
// In the finally so a body that throws cannot strand the count and turn this
// into a different, harder question.
Interlocked.Decrement(ref inFlight);
}
},
maxParallel: slots).Await(TestContext.Current.CancellationToken);

Assert.Equal(nodes, results.Length); Assert.All(results, r => Assert.True(r.IsGreen));
Assert.True(peak > 1, "nodes with no edge between them run at the same time");
Assert.True(peak > 1,
"nodes with no edge between them run at the same time — each body waits (bounded) for a "
+ "second node to join it, so this fails only when the cascade actually serialises");
Assert.True(peak <= slots, $"peak {peak} exceeded the slot cap {slots}");
Assert.Contains(results, r => r.Queued > TimeSpan.Zero); // six nodes on three slots: somebody waited
}
Expand Down
Loading