fix(#2882): the write registers its response callback BEFORE posting

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.

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<MeshNode> 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<PostOptions, PostOptions> 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<MeshNode> reattempt, bool attachToCaller)
                             // code are never retried. See LatePatchResponseRegistry.
                             var lateRegistry = _workspace.Hub.ServiceProvider
                                 .GetService<LatePatchResponseRegistry>();
-                            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(

src/MeshWeaver.Messaging.Hub/IMessageHub.cs

@@ -78,6 +78,20 @@ public interface IMessageHub : IMessageHandlerRegistry, IDisposable
     /// in place. Application code calls <c>hub.Observe(request, options?)</c>.
     /// </summary>
     internal IObservable<IMessageDelivery> Observe(object request, Func<PostOptions, PostOptions> options);
+
+    /// <summary>
+    /// Like <see cref="Observe(object, Func{PostOptions, PostOptions})"/> 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
+    /// <c>LatePatchResponseRegistry</c>) cannot use the self-minting overload: it needs the id
+    /// first, and forcing one through <c>options</c> is overwritten by <c>WithMessageId</c> after
+    /// the caller's options run. Registers the response subject before posting, exactly like the
+    /// self-minting overload.
+    /// </summary>
+    /// <returns>The response observable, or <c>null</c> 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.</returns>
+    internal IObservable<IMessageDelivery>? Observe(object request, Func<PostOptions, PostOptions> options, string messageId);
     /// <summary>
     /// 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

src/MeshWeaver.Messaging.Hub/MessageHub.cs

@@ -1021,6 +1021,51 @@ public IObservable<IMessageDelivery> Observe(object r, Func<PostOptions, PostOpt
             capturedCtx);
     }
 
+    /// <summary>
+    /// <see cref="Observe(object, Func{PostOptions, PostOptions})"/> 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
+    /// <c>LatePatchResponseRegistry</c> entry) BEFORE anything can answer.
+    /// </summary>
+    public IObservable<IMessageDelivery>? Observe(object r, Func<PostOptions, PostOptions> 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 ?? "<null>";
+        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<IMessageDelivery> ObserveById(string messageId,
         string requestType = "<unknown>",
         Address? target = null)

src/MeshWeaver.Messaging.Hub/MessageHubExtensions.cs

@@ -80,6 +80,26 @@ public static IMessageHub CreateMessageHub(this IServiceProvider serviceProvider
     public static IObservable<IMessageDelivery> Observe(this IMessageHub hub, IMessageDelivery delivery)
         => hub.Observe(delivery);
 
+    /// <summary>
+    /// Posts <paramref name="request"/> under a CALLER-SUPPLIED message id and observes the
+    /// response, registering the subject BEFORE the post — the #2882 seam.
+    ///
+    /// <para>🚨 <b>Why the id must come from the caller.</b> The self-minting overload applies
+    /// <see cref="PostOptions.WithMessageId"/> 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
+    /// <c>LatePatchResponseRegistry</c> before the post. The previous <c>Post(...)</c> +
+    /// <c>Observe(delivery)</c> 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 <c>GetMeshNode</c>'s Issue()).</para>
+    /// </summary>
+    /// <returns>The response observable, or <c>null</c> when the target address could not be
+    /// resolved (the post returned null) — the registered subject is already cleaned up.</returns>
+    public static IObservable<IMessageDelivery>? Observe(
+        this IMessageHub hub, object request, Func<PostOptions, PostOptions> options, string messageId)
+        => hub.Observe(request, options, messageId);
+
     /// <summary>
     /// Posts <paramref name="request"/> and observes the typed response.
     /// <para>

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;
+
+/// <summary>
+/// 🚨 <b>#2882 — the WRITE path's twin of
+/// <see cref="WorkspaceCacheEvictionTest.GetMeshNode_WarmOwner_DropsResponse_WhenSubjectRegisteredAfterPost"/>.</b>
+///
+/// <para><c>MeshNodeStreamHandle.UpdateRemote</c> used to <c>Post</c> the
+/// <see cref="PatchDataRequest"/> and only THEN register the response subject
+/// (<c>Observe(delivery)</c>). The hub DROPS any response whose requestId has no registered
+/// subject yet ("No subject found for response … treating as processed",
+/// <c>MessageHub.HandleCallbacks</c>). 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 <c>WriteVerdictBound</c> and failed <c>OwnerUnreachable</c>, and the request trail
+/// could only say <c>REGISTERED_AFTER_POST … earlier stages not recorded</c>. That is the
+/// signature behind the intermittent 31 s write timeouts (1-in-6 bulk runs of
+/// <c>MeshWeaver.AI.Test</c>; Plugins#1014's <c>Patch_ConcurrentUpdates_NoDeadlock</c>).</para>
+///
+/// <para>The fix is the same as the read path's: register the subject BEFORE posting, via the
+/// caller-supplied-id <c>Observe(request, options, messageId)</c> seam — which the write needs
+/// because it must also arm <c>LatePatchResponseRegistry</c> under that id before the post. Here
+/// the preemption is made explicit (an awaited delay) so the race is DETERMINISTIC in both
+/// directions.</para>
+/// </summary>
+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);
+    }
+}