From 611e321ea9e4df84fa82ba786841f1b631d43c99 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Roland=20B=C3=BCrgi?= Date: Mon, 31 Aug 2026 09:23:19 +0200 Subject: [PATCH] fix(#2882): the write registers its response callback BEFORE posting MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit MeshNodeStreamHandle.UpdateRemote posted the PatchDataRequest and only then registered the response subject (Post -> arm late watch -> Observe(delivery)). The hub DROPS a response whose requestId has no registered subject yet ("No subject found for response ... treating as processed", HandleCallbacks), and a WARM owning per-node hub acks in sub-millisecond time — so a thread-pool preemption between Post and Observe lost the verdict, the caller waited out the full 31 s WriteVerdictBound, and failed OwnerUnreachable for a write that had APPLIED. Reproduced 1-in-6 bulk runs of MeshWeaver.AI.Test; same signature as Plugins#1014 and the recorded occurrence behind Plugins#971. The READ path had the identical defect and the identical fix (GetMeshNode's Issue(), pinned by GetMeshNode_WarmOwner_DropsResponse_WhenSubjectRegisteredAfterPost). The write could not take the same route because Observe(object, options) mints its own id AFTER the caller's options run, and the write needs the id UP FRONT to arm LatePatchResponseRegistry before anything can answer. ── the seam (#2882's ask, verbatim) ── Observe(object request, Func options, string messageId) on IMessageHub + MessageHub + the public extension: registers the subject under the SUPPLIED id, then posts with WithMessageId(id). Returns null when Post returned null (unresolvable address), removing the just-registered subject so nothing ripens into a leaked pending callback. ── the write path, reordered ── mint requestId -> arm LatePatchResponseRegistry -> Observe(request, options, requestId) [registers subject, THEN posts] Arming the late watch before the post also closes the sibling window where a DeliveryFailure could arrive before Register ran. The unresolvable-address branch keeps its immediate OwnerUnreachable error and now also disarms the late watch it armed. As a side effect the write trail loses its permanent REGISTERED_AFTER_POST (earlier stages not recorded) marker — DescribeRequestFate can finally say where a write's delivery went, which is the diagnostic gap the issue led with. ── the deterministic twin test ── WriteVerdictRegisteredBeforePostTest mirrors the read path's regression pin, with the preemption made explicit in both directions: post-then-observe with a warm owner provably NEVER emits (the ack is dropped); the new seam under the same preemption provably emits the Success ack (the AsyncSubject buffers and replays it). Monolith.Test write-path + cache suites (OwnerDisposalNack, LateNackReenqueue, WriteWaitsForCommitVerdict, QueuedWriteAdvancesOnHandoff, SupersededActivationPatchNack, NackReachesTheWaiterDuringTeardown, WorkspaceCacheEviction, and the new twin): 15/15. Messaging.Hub / Mesh.Contract / Monolith.Test: -c Release -warnaserror, 0/0. This is the strongest root-cause candidate yet for the ambient ADVANCE_WITHOUT_HANDOFF -> 31 s family (#2865/#2776): the queue advances because the owner's ack was dropped, not because the owner never answered. Co-Authored-By: Claude Opus 5 (1M context) --- ...-no-longer-goes-missing-on-a-fast-reply.md | 26 +++++++ .../MeshNodeStreamExtensions.cs | 70 ++++++++++++------ src/MeshWeaver.Messaging.Hub/IMessageHub.cs | 14 ++++ src/MeshWeaver.Messaging.Hub/MessageHub.cs | 45 +++++++++++ .../MessageHubExtensions.cs | 20 +++++ .../WriteVerdictRegisteredBeforePostTest.cs | 74 +++++++++++++++++++ 6 files changed, 226 insertions(+), 23 deletions(-) create mode 100644 src/MeshWeaver.Documentation/Data/WhatsNew/2026-08-31-a-saved-answer-no-longer-goes-missing-on-a-fast-reply.md create mode 100644 test/MeshWeaver.Hosting.Monolith.Test/WriteVerdictRegisteredBeforePostTest.cs diff --git a/src/MeshWeaver.Documentation/Data/WhatsNew/2026-08-31-a-saved-answer-no-longer-goes-missing-on-a-fast-reply.md b/src/MeshWeaver.Documentation/Data/WhatsNew/2026-08-31-a-saved-answer-no-longer-goes-missing-on-a-fast-reply.md new file mode 100644 index 0000000000..4ecfb4c8d3 --- /dev/null +++ b/src/MeshWeaver.Documentation/Data/WhatsNew/2026-08-31-a-saved-answer-no-longer-goes-missing-on-a-fast-reply.md @@ -0,0 +1,26 @@ +--- +Name: Saving no longer times out when the answer arrives too fast +Category: Fix +Description: Occasionally a save would wait half a minute and then report the owner unreachable, even though the write had gone through. The confirmation was arriving faster than the sender had prepared to receive it, and was thrown away; the listener is now in place before the request is sent. +Icon: Sparkle +Order: -20260831 +--- + +# Saving no longer times out when the answer arrives too fast + +When part of the mesh saves a change to data owned elsewhere, it sends the change across and waits +for the owner to confirm. Rarely — about once in several full test runs, and by the same mechanism +in production — that wait ran a full thirty-one seconds and then failed with "owner unreachable", +even though the owner had applied the change and confirmed it immediately. + +The confirmation was being thrown away. The sender posted the request first and only then set up the +listener for the answer — and an owner that is already warm answers in under a millisecond, so on a +busy machine the confirmation could arrive in the gap and find nobody listening. Unclaimed answers +are discarded by design; the sender then waited out its entire budget for a reply that had already +come and gone. + +The reading half of the platform had this exact fault, found and fixed earlier with a test that +forces the bad timing on purpose. The writing half had the same shape and now has the same fix: the +listener is registered before the request is posted, so however fast the answer comes back, someone +is there to receive it. The same forced-timing test now covers writes too, in both directions — the +old order provably loses the answer, the new order provably keeps it. diff --git a/src/MeshWeaver.Mesh.Contract/MeshNodeStreamExtensions.cs b/src/MeshWeaver.Mesh.Contract/MeshNodeStreamExtensions.cs index 31e4d79822..9f0db726cd 100644 --- a/src/MeshWeaver.Mesh.Contract/MeshNodeStreamExtensions.cs +++ b/src/MeshWeaver.Mesh.Contract/MeshNodeStreamExtensions.cs @@ -7,6 +7,7 @@ using MeshWeaver.Data.Serialization; using MeshWeaver.Mesh.Services; using MeshWeaver.Messaging; +using MeshWeaver.ShortGuid; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; @@ -1524,28 +1525,33 @@ private IObservable UpdateRemote( var baseValues = MeshNodePatchMerge.ExtractBaseValues(currentNode, patch); var capturedContext = capturedContextAtEntry; - var delivery = _workspace.Hub.Post( - new PatchDataRequest(new MeshNodeReference(), new RawJson(patchJson)) - { - BaseValues = baseValues is null - ? null - : new RawJson(baseValues.ToJsonString(jsonOpts)) - }, - o => - { - o = o.WithTarget(new Address(_path!)); - return capturedContext is null - ? o - : o.WithAccessContext(capturedContext); - }); - if (delivery == null) + // 🚨 NOT posted here any more (#2882). The old shape was + // Post(...) → arm the late watch → Observe(delivery), which registers + // the response subject only AFTER the delivery is on its way. The hub + // DROPS a response whose id has no registered subject yet ("No subject + // found for response … treating as processed", HandleCallbacks), and a + // WARM owning per-node hub answers in sub-millisecond time — so a + // thread-pool preemption between Post and Observe lost the verdict and + // the caller waited out the full 31 s WriteVerdictBound with a trail + // that could only say REGISTERED_AFTER_POST. The read path had the + // identical defect and the identical fix (GetMeshNode's Issue(), pinned + // by GetMeshNode_WarmOwner_DropsResponse_WhenSubjectRegisteredAfterPost). + // The request and options are built here; the post happens inside + // Observe(request, options, requestId) below, AFTER the late watch is + // armed and the subject is registered. + var patchRequest = new PatchDataRequest(new MeshNodeReference(), new RawJson(patchJson)) { - observer.OnError(new MeshNodeStreamException(new MeshNodeError( - MeshNodeErrorCode.OwnerUnreachable, - _path!, - "Post of PatchDataRequest returned null — owner address could not be resolved"))); - return; - } + BaseValues = baseValues is null + ? null + : new RawJson(baseValues.ToJsonString(jsonOpts)) + }; + Func patchPostOptions = o => + { + o = o.WithTarget(new Address(_path!)); + return capturedContext is null + ? o + : o.WithAccessContext(capturedContext); + }; // 🚨 EXACTLY-ONCE caller terminal (#2661). The verdict can now reach // the caller on FOUR seams — the bounded response wait, the late @@ -1652,7 +1658,10 @@ void ChainTerminal(IObservable reattempt, bool attachToCaller) // code are never retried. See LatePatchResponseRegistry. var lateRegistry = _workspace.Hub.ServiceProvider .GetService(); - var requestId = delivery.Id; + // 🚨 Minted BEFORE the post — the whole point of the #2882 seam. The + // late watch below and the response subject are both keyed by this id, + // and both must exist before anything can answer. + var requestId = Guid.NewGuid().AsString(); lateRegistry?.Register(requestId, _path!, resp => { // 🚨 EVERY branch below now settles the CALLER too (#2661). Before, @@ -1802,7 +1811,22 @@ or MeshNodeErrorCode.Conflict // Every terminal below disarms the late watch (Complete); ONLY the // TimeoutException branch leaves it armed — that is the one case where // the owner's verdict is still outstanding. - var responseSub = _workspace.Hub.Observe(delivery) + // Registers the subject under requestId, THEN posts (#2882). Null = + // the owner address could not be resolved — nothing was posted and + // nothing will answer, so disarm the watch armed above and fail the + // caller exactly as the old Post-returned-null branch did. + var responseObservable = _workspace.Hub.Observe( + patchRequest, patchPostOptions, requestId); + if (responseObservable is null) + { + lateRegistry?.TryComplete(requestId); + FailTerminal(new MeshNodeStreamException(new MeshNodeError( + MeshNodeErrorCode.OwnerUnreachable, + _path!, + "Post of PatchDataRequest returned null — owner address could not be resolved"))); + return; + } + var responseSub = responseObservable .Timeout(UpdateResponseWaitBound) .Take(1) .Subscribe( diff --git a/src/MeshWeaver.Messaging.Hub/IMessageHub.cs b/src/MeshWeaver.Messaging.Hub/IMessageHub.cs index a3d96b9257..1dbc4f44de 100644 --- a/src/MeshWeaver.Messaging.Hub/IMessageHub.cs +++ b/src/MeshWeaver.Messaging.Hub/IMessageHub.cs @@ -78,6 +78,20 @@ public interface IMessageHub : IMessageHandlerRegistry, IDisposable /// in place. Application code calls hub.Observe(request, options?). /// internal IObservable Observe(object request, Func options); + + /// + /// Like but with a + /// CALLER-SUPPLIED message id — the seam #2882 asked for. A caller that must arm something + /// keyed by the request id BEFORE the post (the cross-hub write arms + /// LatePatchResponseRegistry) cannot use the self-minting overload: it needs the id + /// first, and forcing one through options is overwritten by WithMessageId after + /// the caller's options run. Registers the response subject before posting, exactly like the + /// self-minting overload. + /// + /// The response observable, or null when the post itself returned null (the + /// target address could not be resolved) — the just-registered subject is removed again, so + /// a caller keeping its own "unresolvable address" error path leaks nothing. + internal IObservable? Observe(object request, Func options, string messageId); /// /// Schedules an action to run on the hub's action block (so it executes serially with message /// handling, on the hub's thread). The work is posted as an execution request; its diff --git a/src/MeshWeaver.Messaging.Hub/MessageHub.cs b/src/MeshWeaver.Messaging.Hub/MessageHub.cs index de04ce6237..44664a1161 100644 --- a/src/MeshWeaver.Messaging.Hub/MessageHub.cs +++ b/src/MeshWeaver.Messaging.Hub/MessageHub.cs @@ -1021,6 +1021,51 @@ public IObservable Observe(object r, Func + /// with a CALLER-SUPPLIED id — + /// the #2882 seam. Identical register-subject-then-post ordering; the only difference is who + /// mints the id, which is what lets the caller arm id-keyed state (the cross-hub write's + /// LatePatchResponseRegistry entry) BEFORE anything can answer. + /// + public IObservable? Observe(object r, Func options, string messageId) + { + // Same identity capture as the self-minting overload — see its comment. + var capturedCtx = accessService.Context; + var probeOptions = options(new PostOptions(Address)); + var requestType = r?.GetType().Name ?? ""; + var subject = GetOrAddResponseSubject(messageId, requestType, probeOptions.Target, + (r as IDiagnosticKeyed)?.DiagnosticKey); + IMessageDelivery? posted; + try + { + posted = Post(r, opts => options(opts).WithMessageId(messageId)); + } + catch (Exception postEx) + { + requestFates.Find(messageId)?.Add($"POST_THREW {postEx.GetType().Name}: {postEx.Message}", Address); + throw; + } + if (posted is null) + { + // The address could not be resolved, so nothing will ever answer this id. Remove the + // subject registered three lines up rather than letting it ripen into a leaked + // pending callback the quiescing budget reports, and hand the caller back its + // existing "unresolvable address" error path via null. + lock (responseSubjects) + { + if (responseSubjects.TryGetValue(messageId, out var entry) + && ReferenceEquals(entry.Subject, subject)) + responseSubjects.Remove(messageId); + } + return null; + } + return RestoreUserContextOnEmission( + WrapWithCancelOnDispose( + ApplyTimeout(subject, requestType, probeOptions.Target, messageId), + messageId, subject), + capturedCtx); + } + private IObservable ObserveById(string messageId, string requestType = "", Address? target = null) diff --git a/src/MeshWeaver.Messaging.Hub/MessageHubExtensions.cs b/src/MeshWeaver.Messaging.Hub/MessageHubExtensions.cs index f093637044..729320e0e9 100644 --- a/src/MeshWeaver.Messaging.Hub/MessageHubExtensions.cs +++ b/src/MeshWeaver.Messaging.Hub/MessageHubExtensions.cs @@ -80,6 +80,26 @@ public static IMessageHub CreateMessageHub(this IServiceProvider serviceProvider public static IObservable Observe(this IMessageHub hub, IMessageDelivery delivery) => hub.Observe(delivery); + /// + /// Posts under a CALLER-SUPPLIED message id and observes the + /// response, registering the subject BEFORE the post — the #2882 seam. + /// + /// 🚨 Why the id must come from the caller. The self-minting overload applies + /// AFTER the caller's options, so an id cannot be + /// forced through it — and the cross-hub write needs the id up front to arm + /// LatePatchResponseRegistry before the post. The previous Post(...) + + /// Observe(delivery) shape registered the subject after the post, and the hub DROPS a + /// response whose id has no registered subject yet — against a WARM owner answering in + /// sub-millisecond time, a thread-pool preemption between the two calls lost the verdict and + /// the caller waited out the full 31 s outer bound. The read path had the identical defect, + /// fixed the identical way (see GetMeshNode's Issue()). + /// + /// The response observable, or null when the target address could not be + /// resolved (the post returned null) — the registered subject is already cleaned up. + public static IObservable? Observe( + this IMessageHub hub, object request, Func options, string messageId) + => hub.Observe(request, options, messageId); + /// /// Posts and observes the typed response. /// diff --git a/test/MeshWeaver.Hosting.Monolith.Test/WriteVerdictRegisteredBeforePostTest.cs b/test/MeshWeaver.Hosting.Monolith.Test/WriteVerdictRegisteredBeforePostTest.cs new file mode 100644 index 0000000000..5689aa0c22 --- /dev/null +++ b/test/MeshWeaver.Hosting.Monolith.Test/WriteVerdictRegisteredBeforePostTest.cs @@ -0,0 +1,74 @@ +using System; +using System.Reactive.Linq; +using System.Threading.Tasks; +using MeshWeaver.Data; +using MeshWeaver.Fixture; +using MeshWeaver.Hosting.Monolith.TestBase; +using MeshWeaver.Mesh; +using MeshWeaver.Mesh.Services; +using MeshWeaver.Messaging; +using MeshWeaver.ShortGuid; +using Xunit; + +namespace MeshWeaver.Hosting.Monolith.Test; + +/// +/// 🚨 #2882 — the WRITE path's twin of +/// . +/// +/// MeshNodeStreamHandle.UpdateRemote used to Post the +/// and only THEN register the response subject +/// (Observe(delivery)). The hub DROPS any response whose requestId has no registered +/// subject yet ("No subject found for response … treating as processed", +/// MessageHub.HandleCallbacks). A WARM owning per-node hub acks in sub-millisecond time, so +/// a thread-pool preemption between Post and Observe lost the verdict — the caller waited out the +/// full 31 s WriteVerdictBound and failed OwnerUnreachable, and the request trail +/// could only say REGISTERED_AFTER_POST … earlier stages not recorded. That is the +/// signature behind the intermittent 31 s write timeouts (1-in-6 bulk runs of +/// MeshWeaver.AI.Test; Plugins#1014's Patch_ConcurrentUpdates_NoDeadlock). +/// +/// The fix is the same as the read path's: register the subject BEFORE posting, via the +/// caller-supplied-id Observe(request, options, messageId) seam — which the write needs +/// because it must also arm LatePatchResponseRegistry under that id before the post. Here +/// the preemption is made explicit (an awaited delay) so the race is DETERMINISTIC in both +/// directions. +/// +public class WriteVerdictRegisteredBeforePostTest(ITestOutputHelper output) : MonolithMeshTestBase(output) +{ + [Fact(Timeout = 30000)] + public async Task PatchAck_WarmOwner_IsDropped_WhenSubjectRegisteredAfterPost_AndKept_WhenBefore() + { + var path = $"{TestPartition}/write-race"; + await NodeFactory.CreateNode( + new MeshNode("write-race", TestPartition) { Name = "Warm", NodeType = "Markdown" }).Should().Emit(); + + // WARM the owning per-node hub so a patch is acked in sub-ms — the precondition under + // which the post-then-observe window loses the verdict. + await Mesh.GetMeshNode(path).Should().Match(n => n is not null); + var addr = new Address(path); + + // (1) OLD shape — post, then a deliberate preemption, then register the subject. The warm + // owner's ack lands during the delay, finds no subject, and is dropped: observing + // afterwards never emits. This arm is what fails if anyone reverts to post-then-observe. + var dropped = RequestHub.Post( + new PatchDataRequest(new MeshNodeReference(), new RawJson("""{"name":"AfterPost"}""")), + o => o.WithTarget(addr)); + dropped.Should().NotBeNull(); + await Task.Delay(300); // let the warm ack complete before the subject exists + await RequestHub.Observe(dropped!).Select(d => (object?)d.Message) + .Should().NotEmit(within: 2.Seconds()); + + // (2) FIX shape — the #2882 seam: a caller-minted id, subject registered before the post. + // The same preemption is harmless: the ack is buffered in the AsyncSubject and + // replayed to the late subscriber. + var requestId = Guid.NewGuid().AsString(); + var safe = RequestHub.Observe( + new PatchDataRequest(new MeshNodeReference(), new RawJson("""{"name":"BeforePost"}""")), + o => o.WithTarget(addr), + requestId); + Assert.NotNull(safe); // the target address resolves, so the seam returns the observable + await Task.Delay(300); // same preemption — but the subject was registered before the post + await safe!.Select(d => d.Message as PatchDataResponse) + .Should().Within(5.Seconds()).Match(r => r != null && r.Success); + } +}