Skip to content

fix(#3477): correlate a re-enqueued write with the NACK that caused it #3532

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
meshweaver-cloud merged 3 commits into from
Sep 7, 2026
Merged

fix(#3477): correlate a re-enqueued write with the NACK that caused it #3532

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
22b4297
fix(#3477): correlate a re-enqueued write with the NACK that caused it
rbuergi Sep 7, 2026
e77b22e
fix(test): stop the log capture awaiting a subject signalled by the hub
rbuergi Sep 7, 2026
cd0c0fe
docs: never await an observable the code under test signals
rbuergi Sep 7, 2026
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
67 changes: 67 additions & 0 deletions src/MeshWeaver.Documentation/Data/Architecture/WritingTests.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -384,6 +384,73 @@ Alternatively, start the assertion first without awaiting it (`var assertion = o

---

## 🚨 Never `await` an Observable the Code Under Test Signals

**`await someObservable` resumes its continuation INLINE on whatever thread called `OnNext`.** If
that thread belongs to the code under test — a hub's action-block thread, a grain turn — then
*everything after the await* runs there: the remaining assertions, the `finally`, the method return,
and the framework's teardown.

If the test is also waiting on that same component (disposing it, draining it, parking it), it
**wedges in total silence**. No assertion fails and no `.Timeout()` fires, because the thread that
would carry the test forward is the one being waited on.

### The fingerprint: absent teardown lines

```
=== TEST START: … ===
[+0.2s] [Warning] … LATE_NACK_REENQUEUE … corr=… ← the awaited line WAS written
[+5.2s] [Warning] … Dropping UnsubscribeRequest …
← 84 seconds of nothing
[+89.8s] [FAIL] Test execution timed out after 90000 milliseconds
```

**No `=== TEST END ===` and no `[DISPOSE]` lines.** Those are present on every ordinary failure, so
their absence means the method never returned — which separates "an inner wait lost" from "the
thread is gone". Read that before theorising about which bound expired.

### Why it presents as a flake

A capture that replays a buffer and then concats a live subject has **two paths that resume on
different threads**:

| path taken | continuation resumes on | observed |
|---|---|---|
| the **replay** of already-buffered values | the test's own thread | 663 ms, green |
| the **live subject** | the component's own thread | 90 s, killed |

So it passes locally and on the PR build, and fails in the merge queue. **Re-running proves
nothing** — local runs may take the replay path every time. Measured on #3477/PR #3532: five green
local runs did not discriminate, because the *un-fixed* code also passed locally.

### The control that names it in one step

Its sibling test — identical park/dispose choreography, differing only in that it asserted on
durable storage rather than on a log line — **passed in the same shard of the same run**. When two
tests of one mechanism disagree on a single host, diff what they *observe*, not what they *do*.

### What to write instead

**Poll a buffer from `Observable.Interval`**, so the continuation resumes on the scheduler's thread:

```csharp
// ✅ the continuation lands on the scheduler's thread, never on the hub's
return Observable.Interval(TimeSpan.FromMilliseconds(50)).StartWith(0L)
.Select(_ => lines.ToImmutableArray().Select(l => re.Match(l)).FirstOrDefault(m => m.Success))
.Where(m => m is not null).Select(m => m!)
.FirstAsync().Timeout(within);
```

Delete the subject rather than leaving it unused — keeping one invites the await back. And **prefer
asserting ground truth (what landed in storage) over a log line**: a log line buys determinism only
if *observing* it cannot perturb what produced it.

🚨 **Avoiding `.ToTask()` does not avoid this.** The ban on `.ToTask()` exists because *"a Task
completed inside an Rx pipeline resumes its awaiter inline on the signalling thread"* — and that
reasoning applies verbatim to awaiting the observable directly. In the incident above, the offending
method's own comment cited that rule to justify returning `IObservable`, and then the caller awaited
it. The hazard is the **inline resume**, not the `Task`.

## "One Emission Carrying Everything" — Batched or Late? The First Snapshot Decides

A test that asserts progress **streams** (several distinct snapshots, not one lump) fails with a
Expand Down
56 changes: 41 additions & 15 deletions src/MeshWeaver.Mesh.Contract/MeshNodeStreamExtensions.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1590,17 +1590,29 @@ void EmitOnce(MeshNode value)
/// refused this write against, so the re-attempt rebases on something newer
/// (<see cref="RebaseSource"/>). 0 for a caller write and for the re-enqueue codes that never
/// reached a merge.</param>
/// <param name="correlationId">
/// 🚨 #3477: the id that makes a re-enqueued write followable. Each attempt registers under a
/// FRESH <c>requestId</c> (the late registry mints one per attempt), so after
/// <c>LATE_NACK_REENQUEUE</c> the trail went dark — nothing related the child's registration to
/// the parent's, and "never left the cache hub", "never activated the owner" and "activated and
/// never answered" were indistinguishable in the log. That is the ambiguity #3477 exists to
/// remove. Minted once at the caller's entry and passed UNCHANGED into every re-attempt, so one
/// <c>grep corr=&lt;id&gt;</c> yields the whole chain across attempts.
/// </param>
private IObservable<MeshNode> UpdateRemote(
Func<MeshNode, MeshNode> update, int attempt = 0, long refusedBaseVersion = 0,
MeshNode? pendingSelfWrite = null, Action<MeshNode?>? onLocalState = null)
MeshNode? pendingSelfWrite = null, Action<MeshNode?>? onLocalState = null,
string? correlationId = null)
=> Observable.Create<MeshNode>(observer =>
{
// Minted here on the FIRST attempt only; a re-enqueue passes the parent's through.
var corr = correlationId ?? Guid.NewGuid().AsString();
var diagLogger = _workspace.Hub.ServiceProvider
.GetService<Microsoft.Extensions.Logging.ILoggerFactory>()
?.CreateLogger("MeshWeaver.Mesh.MeshNodeStreamHandle");
diagLogger?.LogDebug(
"[UpdateRemote] BEGIN hub={Hub} target={Path} attempt={Attempt}",
_workspace.Hub.Address, _path, attempt);
"[UpdateRemote] BEGIN hub={Hub} target={Path} attempt={Attempt} corr={Corr}",
_workspace.Hub.Address, _path, attempt, corr);

// 🚨 Capture AccessContext SYNCHRONOUSLY here, NOT inside the
// deferred initialSub.Subscribe callback below. The outer
Expand Down Expand Up @@ -1939,6 +1951,18 @@ void ChainTerminal(IObservable<MeshNode> reattempt, bool attachToCaller)
// 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();
// 🚨 #3477. A fresh requestId is minted PER ATTEMPT, so a re-enqueue
// registers under a different one and nothing related the child to the
// parent — after LATE_NACK_REENQUEUE the trail went dark and "never left
// the cache hub", "never activated the owner" and "activated and never
// answered" read identically. This line is what makes the chain
// followable: Debug because it is one per late-registered write and this
// is the hot path; the WARNING that pays for it is VERDICT_TIMEOUT,
// which names the same corr and only fires once a write has already
// failed to settle.
diagLogger?.LogDebug(
"[UpdateRemote] LATE_VERDICT_REGISTERED hub={Hub} target={Path} attempt={Attempt} corr={Corr} requestId={RequestId}",
_workspace.Hub.Address, _path, attempt, corr, requestId);
lateRegistry?.Register(requestId, _path!, resp =>
{
// 🚨 EVERY branch below now settles the CALLER too (#2661). Before,
Expand All @@ -1960,8 +1984,8 @@ void ChainTerminal(IObservable<MeshNode> reattempt, bool attachToCaller)
// starts on the ack instead of sitting out QueueAdvanceBound.
onLocalState?.Invoke(updated);
diagLogger?.LogDebug(
"[UpdateRemote] LATE_ACK hub={Hub} target={Path} — the owner committed; completing the caller on its verdict",
_workspace.Hub.Address, _path);
"[UpdateRemote] LATE_ACK hub={Hub} target={Path} corr={Corr} — the owner committed; completing the caller on its verdict",
_workspace.Hub.Address, _path, corr);
// The owner COMMITTED. That — not the elapsed bound — is what
// "saved" means, so this is the caller's success terminal.
EmitTerminal();
Expand Down Expand Up @@ -2005,8 +2029,8 @@ or MeshNodeErrorCode.Conflict
? current.Version
: 0;
diagLogger?.LogWarning(
"[UpdateRemote] LATE_NACK_REENQUEUE hub={Hub} target={Path} attempt={Attempt} code={Code} — the patch was never applied; re-enqueueing the original update, rebased on state newer than {RebaseFrom}",
_workspace.Hub.Address, _path, attempt + 1, lateErr.Code, lateRebaseFrom);
"[UpdateRemote] LATE_NACK_REENQUEUE hub={Hub} target={Path} attempt={Attempt} code={Code} corr={Corr} — the patch was never applied; re-enqueueing the original update, rebased on state newer than {RebaseFrom}",
_workspace.Hub.Address, _path, attempt + 1, lateErr.Code, corr, lateRebaseFrom);
// Restore the writer's identity: this callback runs on the
// cache hub's action block where the AsyncLocal context is
// the hub's own, and the re-posted patch must carry the
Expand All @@ -2029,10 +2053,11 @@ or MeshNodeErrorCode.Conflict
UpdateRemote(update, attempt + 1, lateRebaseFrom,
onLocalState: onLocalState is null
? null
: _ => onLocalState(null))
: _ => onLocalState(null),
correlationId: corr)
.Do(_ => { }, ex2 => diagLogger?.LogWarning(ex2,
"[UpdateRemote] LATE_NACK_REENQUEUE failed hub={Hub} target={Path} attempt={Attempt}",
_workspace.Hub.Address, _path, attempt + 1)),
"[UpdateRemote] LATE_NACK_REENQUEUE failed hub={Hub} target={Path} attempt={Attempt} corr={Corr}",
_workspace.Hub.Address, _path, attempt + 1, corr)),
attachToCaller: false);
}
}
Expand Down Expand Up @@ -2191,8 +2216,8 @@ or MeshNodeErrorCode.Conflict
? current.Version
: 0;
diagLogger?.LogWarning(
"[UpdateRemote] OWNER_NACK_REENQUEUE hub={Hub} target={Path} code={Code} attempt={Attempt} — the patch was never applied; re-enqueueing rebased on state newer than {RebaseFrom}",
_workspace.Hub.Address, _path, err.Code, attempt + 1, rebaseFrom);
"[UpdateRemote] OWNER_NACK_REENQUEUE hub={Hub} target={Path} code={Code} attempt={Attempt} corr={Corr} — the patch was never applied; re-enqueueing rebased on state newer than {RebaseFrom}",
_workspace.Hub.Address, _path, err.Code, attempt + 1, corr, rebaseFrom);
// 🚨 The queue slot stays HELD across the re-attempt
// and is released by ITS verdict — the re-attempt is
// this write, still in flight, and letting the
Expand All @@ -2212,7 +2237,8 @@ or MeshNodeErrorCode.Conflict
ChainTerminal(UpdateRemote(update, attempt + 1, rebaseFrom,
onLocalState: onLocalState is null
? null
: _ => onLocalState(null)),
: _ => onLocalState(null),
correlationId: corr),
attachToCaller: true);
}
return;
Expand Down Expand Up @@ -2362,8 +2388,8 @@ or MeshNodeErrorCode.Conflict
// timeout precisely so it can be read here.
var trail = _workspace.Hub.DescribeRequestFate(requestId);
diagLogger?.LogWarning(
"[UpdateRemote] VERDICT_TIMEOUT hub={Hub} target={Path} bound={BoundSeconds}s — the owner produced no terminal for this patch inside the late-response window, which dominates every owner-side terminal path; the write is NOT confirmed. Request trail: {Trail}",
_workspace.Hub.Address, _path, verdictBoundSeconds, trail);
"[UpdateRemote] VERDICT_TIMEOUT hub={Hub} target={Path} bound={BoundSeconds}s corr={Corr} — the owner produced no terminal for this patch inside the late-response window, which dominates every owner-side terminal path; the write is NOT confirmed. 🚨 #3477: `grep corr={Corr}` yields every attempt of this write, including re-enqueues that registered under a different requestId. Request trail: {Trail}",
_workspace.Hub.Address, _path, verdictBoundSeconds, corr, corr, trail);
RaiseError(new MeshNodeStreamException(new MeshNodeError(
MeshNodeErrorCode.OwnerUnreachable, _path!,
$"The owner of '{_path}' returned no verdict for this update within "
Expand Down
Loading
Loading