diff --git a/src/MeshWeaver.Documentation/Data/Architecture/AsynchronousCalls.md b/src/MeshWeaver.Documentation/Data/Architecture/AsynchronousCalls.md index 6ccc21d893..a34aed3025 100644 --- a/src/MeshWeaver.Documentation/Data/Architecture/AsynchronousCalls.md +++ b/src/MeshWeaver.Documentation/Data/Architecture/AsynchronousCalls.md @@ -505,7 +505,13 @@ If you find yourself using `.Take(N)` outside a `SelectMany` that immediately pr 2. **No `*Async` extension shims on `IMeshService`.** Use `meshService.CreateNode(node)` / `UpdateNode(node)` / `DeleteNode(path)` — these return `IObservable`. Never use `.CreateNodeAsync(...)` / `.UpdateNodeAsync(...)` / `.DeleteNodeAsync(...)` — those extensions bridge to Task via `.ToTask()` and deadlock every time they are reached from a hub handler. -3. **Use `hub.Observe(...)` instead of `RegisterCallback` / `AwaitResponse`.** The Task-returning overloads are `[Obsolete]`. Production code MUST use `hub.Observe(delivery)` (already-posted) or `hub.Observe(request, options?)` (also posts) — both return `IObservable]>`. `DeliveryFailure` flows via `OnError`; no Task-await deadlock surface, no silently-skipped callback. +3. **Use `hub.Observe(...)` instead of `RegisterCallback` / `AwaitResponse`.** The Task-returning overloads are `[Obsolete]`. Production code MUST use `hub.Observe(request, options?)` — it returns `IObservable]>`, `DeliveryFailure` flows via `OnError`, and there is no Task-await deadlock surface and no silently-skipped callback. + + **🚨 3a. ALWAYS the pre-registering `hub.Observe(request, options)` — NEVER `hub.Post(...)` then `hub.Observe(delivery)`.** The two are not interchangeable. `Observe(request, options)` registers the response `AsyncSubject` **before** posting; `Observe(delivery)` registers it **after**, and `HandleCallbacks` **DROPS** any response whose correlation id has no registered subject yet ("No subject found for response … treating as processed"). A reply that lands in that window is consumed and gone — the caller's callback then sits pending forever (surfacing as a hang to its timeout, or as a leaked callback at teardown). + + The window is real, not theoretical: `PostImplGeneric` runs `ScheduleNotify` **synchronously**, so by the time `Post` returns the delivery is already enqueued and its turn scheduled onto `turnScheduler` on another thread. Preempt the posting thread — routine under CI thread-pool contention — and the handler answers first. A warm owner answers in sub-millisecond time. + + `Observe(delivery)` is safe ONLY when the response provably cannot have arrived yet, and "the delivery is already posted" is *not* that proof. If you already hold a posted delivery, that is the smell — restructure so the request is handed to `Observe(request, options)` instead. Prior occurrences: `MeshOperations` (four verbs), `MeshNodeStreamExtensions.GetMeshNode`, `HubStreamProviderFactory`, and `CreateOrUpdateNodeRequest`'s inner create (#981). Pinned by `WorkspaceCacheEvictionTest.GetMeshNode_WarmOwner_DropsResponse_WhenSubjectRegisteredAfterPost` and `UpsertInnerCreateObservationTest`. 4. **Never `.QueryAsync($"path:X").FirstOrDefaultAsync()` to read a known node.** Queries go through a lagged read-side index. For a known path: live = `workspace.GetMeshNodeStream(path).Subscribe(...)`; one-shot = the same stream with `.Where(n => n is not null).Take(1).Timeout(...)` (or the `hub.GetMeshNode(path, timeout?)` convenience, which wraps it null-on-absent). @@ -894,13 +900,26 @@ return delivery.Processed(); **Why `hub.Observe(...)` and not `RegisterCallback`:** the legacy `RegisterCallback` returns `Task` and the framework short-circuits the user callback for `DeliveryFailure` — the Task gets the exception, the callback never fires. Callers that drop the Task get silent infinite hangs. `hub.Observe(...)` exposes the same TCS-backed Task via `task.ToObservable()`, so `OnError` fires naturally. -### When you already have a delivery +### 🚨 Never post first and observe afterwards ```csharp +// ❌ WRONG — the response subject is registered AFTER the post. HandleCallbacks DROPS a +// response whose id has no subject yet, so a reply that lands in this window is gone +// and the callback hangs forever. Post returns only after ScheduleNotify has already +// enqueued the delivery and scheduled its turn on another thread — the handler can and +// does win this race under load. var delivery = hub.Post(request, o => ...); hub.Observe(delivery).Subscribe(onNext, onError); + +// ✅ RIGHT — one call; the AsyncSubject is registered BEFORE the post, so however early the +// reply lands it is buffered and replayed to the subscriber. +hub.Observe(request, o => ...).Subscribe(onNext, onError); ``` +See rule 3a. If some intermediate code needs the delivery id up front, pass an explicit +`o.WithMessageId(id)` through the same `Observe(request, options)` call — do not split it +into a post and a later observe. + ### Inside an `IObservable` chain ```csharp @@ -994,7 +1013,7 @@ The view binds to `_suggestions` directly — no callback that returns `Task(req).Subscribe(c => ApplyChange(c))` | | `await Hub.AwaitResponse(req, ...)` | `Hub.Observe(req).Subscribe(r => UpdateState(r.Message), ex => …)` | -| `Hub.RegisterCallback(delivery, r => { … })` | `Hub.Observe(delivery).Subscribe(r => …, ex => …)` | +| `var d = Hub.Post(req, o); Hub.RegisterCallback(d, r => { … })` | `Hub.Observe(req, o).Subscribe(r => …, ex => …)` — pass the REQUEST, never a delivery you already posted (rule 3a) | | `var n = await mesh.GetMeshNodeStream(p).Take(1).ToTask()` | live = `mesh.GetMeshNodeStream(p).Subscribe(n => UpdateState(n))`; one-shot = `Hub.GetMeshNode(p).Subscribe(n => …)` | | `return Task.FromResult(_suggestions.ToArray())` | bind directly to `_suggestions`; let `Subscribe` push updates | | `_ = LoadAsync(); await ...` | sync method that fires `Subscribe` | @@ -1400,7 +1419,7 @@ exist yet). |---|---|---| | `hub.Post(...)` | Yes | Fire-and-forget, safe from any thread | | `hub.Observe(request).Subscribe(onNext, onError)` | Yes | Reactive request/response; DeliveryFailure → onError | -| `hub.Observe(delivery).Subscribe(...)` | Yes | Same, when the delivery was already posted | +| `hub.Post(...)` then `hub.Observe(delivery).Subscribe(...)` | **NO** | Registers the response subject AFTER the post; a reply landing in that window is DROPPED and the callback hangs forever. Use `hub.Observe(request, options)` — see rule 3a | | `meshService.CreateNode(...).Subscribe()` | Yes | Fire-and-forget, no callback logic | | `workspace.GetMeshNodeStream(path).Update(...).Subscribe(...)` in handler body | Yes | Runs on grain scheduler | | `hub.RegisterCallback(...)` | **OBSOLETE** | Use `hub.Observe(...)` — RegisterCallback's Task short-circuits DeliveryFailure → callback silently never fires → caller hangs | diff --git a/src/MeshWeaver.Documentation/Data/WhatsNew/2026-08-10-upsert-always-answers.md b/src/MeshWeaver.Documentation/Data/WhatsNew/2026-08-10-upsert-always-answers.md new file mode 100644 index 0000000000..d510988f8c --- /dev/null +++ b/src/MeshWeaver.Documentation/Data/WhatsNew/2026-08-10-upsert-always-answers.md @@ -0,0 +1,18 @@ +--- +Name: Saving a node always gets an answer +Category: Fix +Description: A create-or-update could finish its work but never report back, leaving the caller waiting. +Icon: Sparkle +Order: -20260810 +--- + +# Saving a node always gets an answer + +When something saved a node that did not exist yet — a repository sync, an import, or the +compile status the platform records for you — the save occasionally completed but never +reported back. Whatever was waiting for the confirmation simply waited, and the work looked +stuck even though it had already succeeded. + +The reply was being discarded because the reply could arrive fractionally before the sender +was ready to listen for it, and a busy machine made that far more likely. The sender now +starts listening before it asks, so the confirmation can never be missed. diff --git a/src/MeshWeaver.Mesh.Contract/MeshExtensions.cs b/src/MeshWeaver.Mesh.Contract/MeshExtensions.cs index 03d9d804f1..f55a4733b2 100644 --- a/src/MeshWeaver.Mesh.Contract/MeshExtensions.cs +++ b/src/MeshWeaver.Mesh.Contract/MeshExtensions.cs @@ -3219,12 +3219,42 @@ private static IMessageDelivery HandleCreateOrUpdateNodeRequest( void DispatchInnerCreate() { var inner = new CreateNodeRequest(node) { CreatedBy = requestedBy }; - var forwarded = hub.Post(inner, o => - { - var withTarget = o.WithTarget(hub.Address); - return inboundCtx is not null ? withTarget.WithAccessContext(inboundCtx) : withTarget; - })!; - hub.Observe(forwarded) + // 🚨 PRE-REGISTERING Observe(request, options) — never Post-then-Observe(delivery) (#981). + // + // This runs OFF the hub action block: `existingObs` is `persistence.Read(...)`, which for + // any adapter that is not a synchronous in-memory hit (partition-routed / embedded-resource + // / pooled backends) emits on an I/O thread. So `hub.Post` here is NOT reentrant into the + // turn loop — `KickDrain` finds `draining == false` and schedules the turn onto + // `turnScheduler` STRAIGHT AWAY, on another thread, while this thread walks on to the next + // statement. `PostImplGeneric` also runs `ScheduleNotify` synchronously, so POSTED / + // RECEIVED / ENQUEUED are already behind us by the time `Post` returns. + // + // With the old `Post(...)` + `Observe(forwarded)` pair, everything between those two + // statements was unprotected: preempt this thread (a saturated CI runner does exactly + // that) and the turn can run the whole create and post its `CreateNodeResponse` first. + // `HandleCallbacks` finds no subject for the correlation, treats the response as consumed + // and DROPS it — then this line registers a subject that nothing will ever answer. The + // upsert never replies, and the caller's callback sits pending until the teardown + // quiescing budget reports it as a leaked `CreateNodeRequest@mesh/`. + // + // It also destroyed the evidence: `RequestFateLedger` starts a trail at REGISTRATION, so + // the post-hoc overload lost POSTED/RECEIVED/ENQUEUED/ROUTED and left a trail reading + // "AWAITING → REGISTERED_AFTER_POST" and nothing else — which is precisely the capture + // that could not name this mechanism. + // + // `Observe(request, options)` registers the AsyncSubject BEFORE posting, so the response + // is buffered no matter how early it lands, and the trail records every stage. + // + // Identity is unaffected by the swap. The overload re-seeds emissions from the AMBIENT + // context, which is exactly what is unreliable on this thread (MeshService.CreateNode's + // note) — but nothing here leans on it: `inner.CreatedBy` pins the identity as a request + // FIELD, the post stamps `inboundCtx` explicitly, `ApplyUpdateViaStream` opens its own + // `SwitchAccessContext(inboundCtx)`, and PostOk/PostFail are `ResponseFor` posts. + hub.Observe(inner, o => + { + var withTarget = o.WithTarget(hub.Address); + return inboundCtx is not null ? withTarget.WithAccessContext(inboundCtx) : withTarget; + }) .Subscribe( d => { diff --git a/test/MeshWeaver.Hosting.Monolith.Test/UpsertInnerCreateObservationTest.cs b/test/MeshWeaver.Hosting.Monolith.Test/UpsertInnerCreateObservationTest.cs new file mode 100644 index 0000000000..20765d9402 --- /dev/null +++ b/test/MeshWeaver.Hosting.Monolith.Test/UpsertInnerCreateObservationTest.cs @@ -0,0 +1,132 @@ +using System; +using System.Reactive.Linq; +using System.Threading.Tasks; +using MeshWeaver.Data; +using MeshWeaver.Hosting.Monolith.TestBase; +using MeshWeaver.Mesh; +using MeshWeaver.Messaging; +using Xunit; + +namespace MeshWeaver.Hosting.Monolith.Test; + +/// +/// Pins the correlation contract behind #981 — "callbacks still pending past the teardown quiescing +/// budget", always reported as a CreateNodeRequest@mesh/<self> that the mesh hub never +/// answered (FutuRe EuropeRe, Cession, PandasExplorer). +/// +/// The mechanism. MessageHub.HandleCallbacks DROPS a response whose correlation id +/// has no registered subject ("No subject found for response … treating as processed"). So the order of +/// the two steps decides whether a reply can be lost: +/// +/// Post, then Observe(delivery) — the subject is registered AFTER the post. +/// PostImplGeneric runs ScheduleNotify synchronously, so the delivery is already +/// enqueued and its turn scheduled onto turnScheduler (another thread) by the time +/// Post returns. Preempt the posting thread — routine on a saturated CI runner — and the +/// turn answers first, the reply is dropped, and the caller's callback is pending FOREVER. +/// Observe(request, options) — registers the AsyncSubject BEFORE posting, +/// so however early the reply lands it is buffered and replayed to a late subscriber. +/// +/// +/// The upsert handler's DispatchInnerCreate ran the unsafe order on exactly this message +/// shape — a self-targeted CreateNodeRequest on the mesh hub, dispatched OFF the action block +/// from the persistence.Read continuation — which is what made CreateOrUpdateNodeRequest +/// silently never answer. +/// +/// The race is made DETERMINISTIC here the same way +/// WorkspaceCacheEvictionTest.GetMeshNode_WarmOwner_DropsResponse_WhenSubjectRegisteredAfterPost +/// does it, but without a fixed sleep: the fence is the node becoming readable plus two full +/// request/response round-trips through the same mesh hub, which together prove the create's turn ran +/// to completion and its reply was pumped. +/// +public class UpsertInnerCreateObservationTest(ITestOutputHelper output) : MonolithMeshTestBase(output) +{ + /// Share Mesh/SP across [Fact]s — see MonolithMeshTestBase.ShareMeshAcrossTests. + protected override bool ShareMeshAcrossTests => true; + + /// + /// Fences until the create's turn has demonstrably finished: the node is readable, and two + /// complete round-trips have since travelled through the mesh hub. Anything the create posted is + /// therefore long since pumped — which is the interleaving that loses a post-hoc registration. + /// + private async Task FenceCreateCompleted(string path) + { + // GetMeshNode is one-shot request/response — it answers null and COMPLETES for a node that + // is not there yet, so it has to be re-queried rather than awaited (WritingTests.md, + // "Polling loops around QueryAsync"). Never a fixed sleep: the wait is on the condition. + await Observable.Interval(TimeSpan.FromMilliseconds(50)).StartWith(0L) + .SelectMany(_ => Mesh.GetMeshNode(path)) + .Where(n => n is not null) + .FirstAsync() + .Timeout(20.Seconds()); + + // Two further COMPLETE round-trips through the same mesh hub. Once these have answered, any + // reply the create posted is long since pumped — the interleaving that loses a post-hoc + // registration. + await Mesh.GetMeshNode(path).Should().Within(15.Seconds()).Match(n => n is not null); + await Mesh.GetMeshNode(path).Should().Within(15.Seconds()).Match(n => n is not null); + } + + [Fact(Timeout = 60000)] + public async Task SelfTargetedCreate_OnTheMeshHub_DropsItsReply_WhenTheSubjectIsRegisteredAfterPost() + { + // ARM 1 — the #981 shape: post first, register the callback afterwards. + var doomedId = $"upsert-race-doomed-{Guid.NewGuid():N}"; + var doomedPath = $"{TestPartition}/{doomedId}"; + var doomed = Mesh.Post( + new CreateNodeRequest(new MeshNode(doomedId, TestPartition) + { + Name = doomedId, + NodeType = "Markdown" + }), + o => o.WithTarget(Mesh.Address)); + doomed.Should().NotBeNull(); + + // The create really did run and answer — so its reply hit HandleCallbacks with no subject. + await FenceCreateCompleted(doomedPath); + + // Registering now cannot recover a reply that was already consumed: the callback is orphaned. + // THIS is the leaked `CreateNodeRequest@mesh/` the quiescing budget reports. + await Mesh.Observe(doomed!).Select(d => (object?)d.Message) + .Should().NotEmit(within: 3.Seconds(), + because: "a response with no registered subject is dropped by HandleCallbacks"); + + // ARM 2 — the fix shape: pre-register, then post. Same fence, reply still delivered. + var safeId = $"upsert-race-safe-{Guid.NewGuid():N}"; + var safePath = $"{TestPartition}/{safeId}"; + var safe = Mesh.Observe( + new CreateNodeRequest(new MeshNode(safeId, TestPartition) + { + Name = safeId, + NodeType = "Markdown" + }), + o => o.WithTarget(Mesh.Address)); + + await FenceCreateCompleted(safePath); + + await safe.Select(d => d.Message) + .Should().Within(10.Seconds()) + .Match(m => m is CreateNodeResponse { Success: true }, + "the pre-registered AsyncSubject buffers the reply for an arbitrarily late subscriber"); + } + + /// + /// The production seam: CreateOrUpdateNodeRequest for a node that does NOT exist forwards an + /// inner self-targeted CreateNodeRequest and must ANSWER. Before the fix that inner create + /// used the post-then-observe order, so whenever the turn won the race the upsert produced no + /// response at all and its caller's callback leaked. + /// + [Fact(Timeout = 60000)] + public async Task Upsert_OfAMissingNode_Answers_ThroughTheInnerCreate() + { + var id = $"upsert-answers-{Guid.NewGuid():N}"; + var node = new MeshNode(id, TestPartition) { Name = id, NodeType = "Markdown" }; + + var response = await Mesh + .Observe(new CreateOrUpdateNodeRequest(node)) + .Should().Within(30.Seconds()).Emit( + "the upsert's inner CreateNodeRequest must never lose its reply"); + + response.Message.Node.Should().NotBeNull(); + response.Message.Node!.Path.Should().Be($"{TestPartition}/{id}"); + } +}