diff --git a/src/MeshWeaver.Data/Serialization/StreamNotConvergingException.cs b/src/MeshWeaver.Data/Serialization/StreamNotConvergingException.cs new file mode 100644 index 0000000000..a933cbf4dc --- /dev/null +++ b/src/MeshWeaver.Data/Serialization/StreamNotConvergingException.cs @@ -0,0 +1,40 @@ +ο»Ώnamespace MeshWeaver.Data.Serialization; + +/// +/// A mirror asked its owner for a fresh authoritative snapshot, was ACKNOWLEDGED, and never +/// received one β€” repeatedly. Terminal: the stream that carries this can never emit again. +/// +/// 🚨 This exception exists so that a lost frame cannot become a SILENT permanent wedge +/// (Systemorph/MeshWeaver#1384). The recovery path it ends is event-driven by design and has no +/// timer anywhere in it: a frame whose BasedOnVersion does not chain onto the last one this +/// mirror applied PROVES a frame was lost in transport, and the only sound reaction is a fresh +/// snapshot from the owner. #2654 made that recovery survive a single lost answer β€” the resync gate +/// is released by the re-ask's round trip, so the next proven gap earns one new re-ask. What it +/// could not do is TERMINATE: when the ownerβ†’subscriber leg keeps eating this stream's snapshots, +/// every re-ask is acknowledged, every acknowledgement releases the gate, every answer dies on the +/// way back, and the mirror re-asks for the rest of the process's life. Its subscribers see none of +/// it β€” a layout area holds "awaiting first data", a data-bound view holds its last value, and no +/// error and no completion ever reaches them. +/// +/// Measured on memex-cloud, 2026-09-01, on Event/SavGeneralversammlung2026/Talk: +/// [SYNC_STREAM] Frame loss detected … incoming Patch v13 chains onto v12 but the last applied +/// frame is v11, followed by Layout area 'Present' … was torn down having never rendered +/// β€” while plain node reads on the same path answered instantly. Recycling the pod that held the +/// activation did not clear it: the owner was healthy, and the wedge lived entirely in a subscriber +/// that had not been told anything was wrong. +/// +/// Fault, therefore, rather than wait. A faulted stream is the one state every consumer +/// already knows how to act on: the store's terminal error reaches every reader, +/// StreamLiveness.IsUsable stops serving the stream from the workspace's caches, and the +/// next natural caller opens a fresh stream that subscribes from scratch. Nothing here retries β€” +/// re-establishing is the subscriber's decision, taken because it was finally told. +/// +public sealed class StreamNotConvergingException : SynchronizationException +{ + /// + /// Initializes a new instance of the class. + /// + /// A message naming the stream, its owner and how many acknowledged + /// fresh-snapshot requests went unanswered. + public StreamNotConvergingException(string message) : base(message) { } +} diff --git a/src/MeshWeaver.Data/Serialization/SynchronizationStream.cs b/src/MeshWeaver.Data/Serialization/SynchronizationStream.cs index 0a7977aa37..36e0c0f46d 100644 --- a/src/MeshWeaver.Data/Serialization/SynchronizationStream.cs +++ b/src/MeshWeaver.Data/Serialization/SynchronizationStream.cs @@ -1629,6 +1629,10 @@ private void UpdateStream(IMessageDelivery delivery, IMessageH // did-not-converge counter: THIS is the event that says the resync worked. ReleaseResyncGate(); resyncAttempts = 0; + // …and the evidence of non-convergence with it: a base snapshot ARRIVED, so every + // ack that preceded it turned out to be answered after all (#1384). Interlocked + // because the counter's other writer is the ack arm, on the response thread. + Interlocked.Exchange(ref unansweredResyncs, 0); } catch (Exception ex) { @@ -1816,6 +1820,67 @@ private void SyncFailed(IMessageDelivery delivery, Exception exception) /// private int resyncAttempts; + /// + /// Re-asks the owner ACKNOWLEDGED and then never followed with a base snapshot, reset by the + /// that finally arrives. This β€” not β€” + /// is the evidence bounds, and the difference is the whole + /// of the #2745 policy: a re-ask REFUSED transiently (a silo whose pod-hub claim has not landed + /// during a rolling deploy answers ) is deliberately ridden + /// out, so it must not accumulate towards a fault. Only an owner that said "I have processed + /// your re-subscribe" and then produced nothing is evidence that the leg is eating this + /// stream's snapshots. + /// + /// Written from the response thread (the ack arm) and from the hub turn (the reset), so + /// every access is interlocked. + /// + private int unansweredResyncs; + + /// + /// How many acknowledged-but-unanswered fresh-snapshot requests this mirror accumulates before + /// it stops asking and FAULTS itself instead β€” the bound that turns #2654's residual silent + /// forever-wedge into a surfaced error (Systemorph/MeshWeaver#1384). + /// + /// What #2654 left behind. The resync gate is released by the re-ask's ROUND TRIP + /// (the Full, the owner's ack, or a verdict), so a lost fresh snapshot no longer latches the + /// mirror shut: the next frame that proves the gap earns one new re-ask. That converges when + /// the leg loses ONE frame. It does not terminate when the leg keeps losing this stream's + /// Fulls β€” every re-ask is acked, every acked re-ask releases the gate, every answering Full + /// dies on the same leg, and the mirror re-asks forever at Warning while its subscriber sits on + /// "awaiting first data" with no error, no completion and nothing to re-establish from. That is + /// the state measured on memex-cloud 2026-09-01 on Event/SavGeneralversammlung2026/Talk: + /// Frame loss detected … incoming Patch v13 chains onto v12 but the last applied frame is + /// v11, then a layout area torn down having never rendered. Recycling the pod holding + /// the activation did not clear it, because nothing on the subscriber side had learned that + /// anything had gone wrong. + /// + /// Why a COUNT is not a retry budget. Nothing here retries and nothing polls: this + /// counts EVIDENCE, not attempts at success. An increment costs a full round trip to the owner + /// (the gate suppresses every re-ask while one is outstanding, and only the ANSWER releases it) + /// PLUS a subsequent owner frame that proves the mirror still has no base. So + /// unansweredResyncs == n means n separate authoritative snapshots were asked for, + /// acknowledged, and never arrived. Raising this bound buys a longer silence, not a better + /// chance β€” which is the opposite of a widened timeout, and why the number is small. + /// + /// Why 3. The one healthy way to spend an attempt without converging is a Patch + /// overtaking the answering Full β€” and the owner queues its re-assert on the SAME stream hub + /// that produces the patches, so a patch can only overtake a re-assert that was already in + /// flight when the re-ask landed. That is one redundant round trip, twice over at the very + /// worst. Three consecutive acknowledged, unanswered asks is not that shape. The two failure + /// directions are also not symmetric: faulting a stream that would eventually have converged + /// costs one re-establish (StreamLiveness.IsUsable refuses to serve a faulted stream, so + /// the workspace cache evicts it and the next natural caller opens a fresh one that subscribes + /// from scratch), while NOT faulting costs a view that never loads and never says so. + /// + private const int MaxUnansweredResyncs = 3; + + /// + /// Set once this mirror has given up (see ) so a later frame + /// arriving on the already-faulted stream re-reports nothing. The store is terminal by then and + /// would swallow a second per the Rx grammar, but the Warning beside it + /// would repeat per frame β€” log volume that says nothing new. + /// + private int resyncGaveUp; + /// /// The re-ask currently outstanding. At most ONE at a time (), so a /// both bounds the registration β€” a CompositeDisposable entry per @@ -1857,8 +1922,62 @@ private void RequestFreshSnapshot() StreamId, Reference); return; } + // Terminal already declared below β€” this mirror has stopped asking for good. A later frame + // must not re-enter: the stream is faulted, and re-reporting it every frame is log volume + // that says nothing new. + if (Volatile.Read(ref resyncGaveUp) == 1) + return; if (Interlocked.Exchange(ref resyncInFlight, 1) == 1) return; + + // 🚨 …AND THE ONE THAT ENDS IT (#1384). VISIBLE is not the same as OVER: #2654 made the + // non-convergence audible in a portal log and left the SUBSCRIBER exactly where it was β€” + // holding a placeholder for a snapshot that is not coming, with no error and no completion, + // so nothing downstream could re-establish. A log line is not an API. Past + // MaxUnansweredResyncs acknowledged-and-unanswered asks this mirror stops asking and FAULTS + // instead, which is the one signal a subscriber can act on: the store's terminal error + // reaches every reader, StreamLiveness.IsUsable stops serving this stream from the + // workspace caches, and the next natural caller opens a fresh one. + // + // Read AFTER the gate is taken, so "no ask is outstanding" is part of the verdict β€” an + // answer still in flight is not evidence of anything. The gate is then released BEFORE the + // fault, in the order this method's opening comment teaches: never leave a latch behind a + // decision that posted no request. Releasing it asks for nothing by itself, and resyncGaveUp + // keeps a later frame from re-reporting a stream that is already terminal. + // + // 🚨 A MIRROR ONLY, and the qualifier is load-bearing. PatchDataChangeRequest reaches + // UpdateStream on the OWNER's server-side stream too, where a subscriber's write arriving + // before that stream has emitted its first frame (its outbound JSON cursor still null) + // takes the same "Patch before base Full" branch. On such a stream the count measures + // nothing: the re-ask is addressed to this very hub, and the Full that would reset it is + // one this stream SENDS rather than receives β€” so it could only ever climb, and faulting on + // it would kill an owner's stream for a subscriber-side race. Convergence is a claim about + // a REMOTE owner; only a stream that has one can fail to converge. Same predicate as + // OwnerVersion() uses to tell the two apart. + // + // Read ONCE into a local, and report from that local everywhere. The verdict, the Warning and + // the exception must be the same number: a diagnostic that says "gave up after 3" beside an + // exception that says "after 4" costs the next reader more than it tells them, and a second + // read is one more thing they would have to prove is stable. (It is β€” the gate is HELD from + // here, so the ack arm cannot increment, and the reset runs on this same hub turn β€” but the + // local means nobody has to reconstruct that argument to trust the number.) + var unanswered = Volatile.Read(ref unansweredResyncs); + if (unanswered >= MaxUnansweredResyncs && !Owner.Equals(Host.Address)) + { + Volatile.Write(ref resyncGaveUp, 1); + ReleaseResyncGate(); + logger.LogWarning( + "[SYNC_STREAM] Resync gave up for {StreamId}: {Attempts} consecutive fresh-snapshot requests to {Owner} were acknowledged and none produced a base snapshot β€” faulting the mirror so its subscribers can re-establish", + StreamId, unanswered, Owner); + OnError(new StreamNotConvergingException( + $"Synchronization stream '{StreamId}' could not recover from a lost frame: " + + $"{unanswered} consecutive fresh-snapshot requests to owner '{Owner}' were " + + "acknowledged and none produced a base snapshot. The mirror holds no state it can " + + "patch onto, so it is faulted rather than left waiting for a snapshot that is not " + + "arriving.")); + return; + } + // 🚨 THE ONE LINE THAT MAKES NON-CONVERGENCE VISIBLE. A second consecutive ask means the // first one produced no base snapshot β€” the answer was lost, refused or thrown away β€” which // is exactly the state #2654 spent its whole life in at Debug level, with a stuck layout @@ -1920,9 +2039,18 @@ private void RequestFreshSnapshot() // the mirror still has no base, so the worst case is one redundant round // trip when a Patch overtakes the answering Full β€” and that redundant // re-ask is itself answered with a Full, which ends the cycle. + // + // 🚨 …AND THE ACK IS ALSO THE EVIDENCE (#1384). An acknowledgement says the + // owner processed this re-subscribe; if a base snapshot never follows, the + // answer died on the leg rather than never being sent, which is exactly the + // non-convergence MaxUnansweredResyncs bounds. Counted HERE and nowhere + // else, deliberately: a re-ask REFUSED transiently is ridden out (#2745) + // and must not accumulate towards a fault, and an increment that races an + // answering Full is undone by that Full's reset a moment later. logger.LogDebug( "[SYNC_STREAM] Fresh-snapshot request for {StreamId} acknowledged by owner {Owner}", StreamId, StreamIdentity.Owner); + Interlocked.Increment(ref unansweredResyncs); ReleaseResyncGate(); }, ResyncRefused); diff --git a/src/MeshWeaver.Documentation/Data/Architecture/DataSyncAndCrdt.md b/src/MeshWeaver.Documentation/Data/Architecture/DataSyncAndCrdt.md index b64ec11728..69e13bd770 100644 --- a/src/MeshWeaver.Documentation/Data/Architecture/DataSyncAndCrdt.md +++ b/src/MeshWeaver.Documentation/Data/Architecture/DataSyncAndCrdt.md @@ -355,6 +355,98 @@ on a rebased clock, the re-ask refused **terminally** (must fault), and the re-a **transiently** with the identical `TargetUnserved` stamp (must be ridden out and still converge). Each fails on the pre-#2654 tree. +### The termination contract β€” when the mirror stops asking and faults (#1384) + +The contract above recovers a mirror when the leg loses **one** thing. It does not +terminate when the leg keeps losing **this stream's** snapshots, and that residual is a +silent forever-wedge: every re-ask is acknowledged, every acknowledgement re-opens the +gate, every answering Full dies on the way back, and the mirror asks again for the rest of +the process's life. The `Resync has not converged` Warning above is the only trace, and a +**log line is not an API** β€” nothing downstream of the stream is told anything, so nothing +can re-establish. + +Measured on memex-cloud, 2026-09-01, on `Event/SavGeneralversammlung2026/Talk`: + +``` +[SYNC_STREAM] Frame loss detected for G4LJWZjBXkWdULLsT-5N5g: incoming Patch v13 chains onto v12 + but the last applied frame is v11 β€” a frame was lost in transport; requesting fresh snapshot +Layout area 'Present' on Event/SavGeneralversammlung2026/Talk was torn down having never rendered + β€” the subscriber only ever saw the "awaiting first data" placeholder. +``` + +Plain node reads on that path answered instantly, so the owner was healthy; **recycling the +pod that held the activation did not clear it**, because the wedge lived entirely in a +subscriber that had not been told anything was wrong. + +So the count of consecutive **acknowledged-and-unanswered** re-asks is bounded, and at +`SynchronizationStream.MaxUnansweredResyncs` (3) the mirror stops asking and calls +`OnError` with a `StreamNotConvergingException`. + +🚨 **The count is of ACKS, not of asks, and that is the #2745 policy in one line.** The +increment lives in the ack arm of `RequestFreshSnapshot` and nowhere else. An +acknowledgement means the owner *processed* the re-subscribe, so a base snapshot that never +follows died on the leg β€” evidence. A re-ask **refused transiently** (`ShuttingDown`, the +rolling-deploy overlap window) is deliberately ridden out by the table above, and counting +it would fault every mirror in that window, which is worse than the bug this fixes. An +increment that races an answering Full is undone by that Full's own reset a moment later. +A re-ask that gets **no verdict at all** (undeliverable, so the hub's own request/response +terminal fires) is likewise not counted β€” "we could not find out" is not an answer about +the owner, and the table above already keeps that case recoverable and loud, one Warning +carrying the exception per attempt. Widening the count to cover it would be a separate +decision with a separate justification, not a free extension of this one. + +🚨 **Mirrors only.** The give-up applies where `StreamIdentity.Owner` is another hub β€” the +same predicate `OwnerVersion()` uses. `PatchDataChangeRequest` reaches `UpdateStream` on the +*owner's* server-side stream as well, where a subscriber write that races that stream's +first outbound frame takes the same "Patch before base Full" branch; there the counter never +resets (the Full that would reset it is one that stream **sends**), so it measures nothing +and faulting on it would kill an owner's stream for a subscriber-side race. + +🚨 **This is a bound on EVIDENCE, not a retry budget.** Nothing here retries and nothing +polls. An increment costs a full round trip to the owner β€” the gate suppresses every re-ask +while one is outstanding, and only the answer releases it β€” **plus** a subsequent owner +frame that proves the mirror still has no base. Raising the bound buys a longer silence, not +a better chance; that is the opposite of a widened timeout, and it is why the number is +small. + +**Why 3.** The one healthy way to spend an attempt without converging is a Patch overtaking +the answering Full. The owner queues its re-assert on the **same** stream hub that produces +the patches, so a patch can only overtake a re-assert that was already in flight when the +re-ask landed β€” one redundant round trip, twice over at the very worst. The two failure +directions are also not symmetric: + +| bound too low | bound too high | +|---|---| +| a stream that would have converged is faulted β‡’ **one re-establish**: `StreamLiveness.IsUsable` refuses to serve a faulted stream (#2387), the cache evicts it, and the next natural caller opens a fresh one that subscribes from scratch | a view that never loads and never says so β‡’ the incident above | + +**The fault is the recovery signal**, not the end of the road. A faulted stream is the one +state every consumer already knows how to act on: the store's terminal error reaches every +reader, `Workspace` drops it from `_remoteStreamCache` and closes it, and `OnError` also +calls `Hub.FailStartup` + `Hub.OpenGate(SynchronizationGate)` β€” which releases whatever was +deferred behind that gate. Re-establishing stays the *subscriber's* decision, taken because +it was finally told. + +**The operator signal** is `[SYNC_STREAM] Resync gave up for {StreamId}: N consecutive +fresh-snapshot requests to {Owner} were acknowledged and none produced a base snapshot` at +**Warning**, followed by the stream's own `OnError` line. Seeing it means the +ownerβ†’subscriber leg is losing frames systematically β€” read it together with the query +pressure work in +[Cross-Schema Fan-Out Elimination](/Doc/Architecture/CrossSchemaFanOutElimination): a leg +drops frames when the process behind it is saturated, and that saturation is the thing to +fix. This bound exists so the saturation cannot cash out as a view that is blank forever. + +**A note on the ordering inside `RequestFreshSnapshot`.** The method establishes that it +*can* ask β€” `Reference is WorkspaceReference` β€” **before** it nulls the cached JSON or takes +the gate. That order matters (the reverse closes the gate on a re-ask that was never made, +permanently by construction) and it is already how the code reads since #2654. It is **not** +what produced the incident above: `LayoutAreaReference` *is* a `WorkspaceReference`, so the +Present-area stream took the ordinary path, asked, was acknowledged, and was never answered. + +Pinned by `StreamResyncGivesUpTest`: one mid-burst Patch eaten to start the resync, then +**every** fresh snapshot eaten, one write per proven gap β€” the mirror must fault with a +`StreamNotConvergingException`, and a fresh subscriber must then get a new stream that +converges on the owner's complete state. It hangs on the pre-#1384 tree. + --- ## 7. Minimal bytes on the wire diff --git a/src/MeshWeaver.Documentation/Data/WhatsNew/2026-09-02-a-view-that-never-loads-now-says-so.md b/src/MeshWeaver.Documentation/Data/WhatsNew/2026-09-02-a-view-that-never-loads-now-says-so.md new file mode 100644 index 0000000000..f8dee31143 --- /dev/null +++ b/src/MeshWeaver.Documentation/Data/WhatsNew/2026-09-02-a-view-that-never-loads-now-says-so.md @@ -0,0 +1,37 @@ +--- +Name: A view that never loads now says so instead of waiting forever +Category: Fix +Description: When the same connection kept losing the fresh copy a stuck page asked for, the page waited in silence for good; it now reports the failure so it can reconnect. +Icon: ArrowSyncCheckmark +Order: -20260902 +--- + +# A view that never loads now says so instead of waiting forever + +Live areas on a page are kept current by a stream of small updates. If one goes missing on the way, +the page notices and asks the server for a complete, fresh copy. A previous fix made that recovery +survive the fresh copy *also* going missing: the page asks again the next time something changes, +which costs one more round trip instead of the rest of the session. + +What was still missing is what happens when the same connection keeps losing **every** fresh copy. +The page asked, the server answered, the answer never arrived β€” over and over, forever, without a +word to the part of the app that draws the area. On screen that was an area stuck on "loading" on a +page whose menus, banner and breadcrumb were fine, with nothing to click and no error to report. +It happened during a live presentation on 1 September: a slide deck whose presentation view never +appeared, while the same page's plain content loaded instantly. Reloading the page could clear it; +restarting the server could not, because nothing was wrong on the server. + +Now the page stops asking after a few unanswered attempts and **reports the failure** instead. That +one change is what makes recovery possible at all: the failed view is dropped rather than kept, so +the next thing that needs it opens a fresh connection and loads normally β€” the same thing a reload +used to do, without the reload. + +Two things this deliberately is not. It is not a timer: the page only ever asks again when a real +update arrives proving it still has nothing to show, so a quiet page stays quiet. And it is not a +retry budget β€” only attempts the server *confirmed it had handled* count, each one costing a full +round trip plus a fresh piece of evidence, so a handful of them is proof the connection is dropping +this view's data rather than impatience. An attempt the server turns down because it is briefly +busy β€” during a rolling update, say β€” is waited out as before and never counts against the view. + +For anyone reading server logs, the give-up is recorded at warning level naming the view and the +server it was asking, right after the `Resync has not converged` warnings that precede it. diff --git a/test/MeshWeaver.Data.Test/StreamResyncGivesUpTest.cs b/test/MeshWeaver.Data.Test/StreamResyncGivesUpTest.cs new file mode 100644 index 0000000000..46a4a033a2 --- /dev/null +++ b/test/MeshWeaver.Data.Test/StreamResyncGivesUpTest.cs @@ -0,0 +1,244 @@ +using System.Linq; +using System.Reactive; +using System.Reactive.Linq; +using System.Reactive.Subjects; +using System.Threading; +using MeshWeaver.Data.Serialization; +using MeshWeaver.Fixture; +using MeshWeaver.Messaging; +using Microsoft.Extensions.DependencyInjection; +using Xunit; + +namespace MeshWeaver.Data.Test; + +/// +/// A RESYNC THAT NEVER CONVERGES MUST STOP BEING SILENT β€” Systemorph/MeshWeaver#1384. +/// +/// This is the third member of the frame-loss family, and it starts where the other two stop. +/// proves the mirror DETECTS a lost frame and asks for a +/// fresh snapshot. proves it still converges when that ONE +/// answer fails β€” the resync gate is released by the re-ask's round trip (#2654), so the next frame +/// that proves the gap earns one new re-ask. Neither of them TERMINATES when the ownerβ†’subscriber +/// leg keeps eating this stream's snapshots, and that is the state a live portal was measured +/// in. +/// +/// The live incident. memex-cloud, 2026-09-01, node +/// Event/SavGeneralversammlung2026/Talk: [SYNC_STREAM] Frame loss detected … incoming +/// Patch v13 chains onto v12 but the last applied frame is v11, then Layout area 'Present' … +/// was torn down having never rendered β€” the subscriber only ever saw the "awaiting first data" +/// placeholder. Plain node reads on the same path answered instantly and recycling the pod that +/// held the activation did not clear it: the owner was healthy, and the wedge lived entirely in a +/// subscriber nobody had told anything was wrong. Every re-ask was acknowledged, every +/// acknowledgement released the gate, every answering Full died on the same leg, and the mirror +/// asked again for the rest of the process's life β€” at Warning in a log, and at NOTHING AT ALL in +/// the API a consumer can act on. +/// +/// What this test injects, and why it is deterministic. Two losses on the owner's post +/// pipeline, both scoped to the one mirror stream under test: one mid-burst Patch (the wire loss +/// that starts the resync β€” the same injection the other two tests use), and then EVERY +/// the owner sends on that stream, which is what a leg that keeps +/// losing frames looks like from the mirror. The test posts the next write only after the loss it +/// depends on has actually happened β€” an event, never a delay β€” so the sequence of proven gaps is +/// fixed rather than raced. +/// +/// The contract. Past MaxUnansweredResyncs ACKNOWLEDGED, unanswered re-asks the +/// mirror stops asking and FAULTS. That is not a retry budget: an increment costs a full round trip +/// to the owner plus a subsequent frame proving the mirror still has no base, so the count is +/// evidence, not effort. And the fault is not the end of the story β€” it is the only signal a +/// subscriber can act on. StreamLiveness.IsUsable refuses to serve a faulted stream (#2387), +/// so the workspace cache evicts it and the next natural caller opens a fresh one that subscribes +/// from scratch; the second half of this test proves exactly that. +/// +public class StreamResyncGivesUpTest(ITestOutputHelper output) : HubTestBase(output) +{ + /// + /// How many acknowledged-but-unanswered fresh-snapshot requests the mirror tolerates before it + /// faults. Mirrors SynchronizationStream.MaxUnansweredResyncs, which is private β€” the + /// coupling is deliberate and asserted by construction: the drive loop below spends exactly this + /// many attempts and then expects the next proven gap to fault. If the production bound moves + /// and this does not, the loop under-drives and the fault wait goes red, which is the right way + /// round for a guard. + /// + private const int MaxUnansweredResyncs = 3; + + /// 1 after the pipeline ate its one mid-burst Patch. Instance field β€” per-test lifetime. + private int droppedPatches; + + /// + /// Patch frames the owner has emitted on the mirror's stream. The loss is injected on the SECOND + /// one so the mirror has applied a real patch first β€” a gap can only be PROVEN against a frame + /// that landed. + /// + private int patchFramesSeen; + + /// + /// How many fresh snapshots the pipeline has eaten. Every one of them is a re-ask the owner + /// answered and the leg destroyed. + /// + private int droppedFreshSnapshots; + + /// + /// Fires once per eaten fresh snapshot, carrying the running count. A + /// because the injection runs on the owner's post pipeline and + /// can complete before the test subscribes. + /// + private readonly ReplaySubject freshSnapshotDropped = new(MaxUnansweredResyncs + 2); + + /// + /// The stream carrying the mirror this test is about, learned from its INITIAL Full. Both hubs + /// run a data source, so more than one sync stream is alive; every injection is scoped to this + /// id so the test can never eat a frame belonging to somebody else's stream β€” including the + /// FRESH stream the recovery half opens, whose id is different by construction. + /// + private volatile string? mirrorStreamId; + + protected override MessageHubConfiguration ConfigureHost(MessageHubConfiguration configuration) + => base.ConfigureHost(configuration) + .AddData(data => data.AddSource(ds => ds + .WithType(t => t.WithKey(d => d.Id)))) + .AddPostPipeline(p => p.AddPipeline((d, next) => + { + if (d.Message is not DataChangedEvent evt) + return next.Invoke(d); + + // Learn the mirror's stream from its initial Full β€” the first frame it ever gets. + if (evt.ChangeType == ChangeType.Full && mirrorStreamId is null) + { + mirrorStreamId = evt.StreamId; + return next.Invoke(d); + } + if (evt.StreamId != mirrorStreamId) + return next.Invoke(d); + + // THE WIRE LOSS that starts the resync: eat exactly one mid-burst Patch. An Ignored + // post is dropped by ScheduleNotify, exactly the shape of the at-most-once transport + // losing a published frame between owner and mirror. + if (evt.ChangeType == ChangeType.Patch + && Interlocked.Increment(ref patchFramesSeen) == 2 + && Interlocked.Exchange(ref droppedPatches, 1) == 0) + return d.Ignored(); + + // …AND THE LEG KEEPS LOSING. Every fresh snapshot the owner sends in answer to a + // re-ask dies exactly like the frame that started the resync. This is the ONE thing + // this test adds to StreamResyncConvergenceTest, which eats a single answer and then + // lets the mirror converge. + if (evt.ChangeType == ChangeType.Full) + { + // Signalling HERE is ORDERED, not hopeful: the owner posts its SubscribeAck from + // HandleSubscribeRequest the moment SubscribeToClient returns, while the + // re-assert runs later on the stream's own turn β€” so by the time this frame + // exists the ack is already on its way to the mirror's host hub, ahead of + // anything the test posts next. The write below therefore lands on an OPEN gate + // with still no base: the frame that earns the next re-ask. + freshSnapshotDropped.OnNext(Interlocked.Increment(ref droppedFreshSnapshots)); + return d.Ignored(); + } + return next.Invoke(d); + })); + + protected override MessageHubConfiguration ConfigureClient(MessageHubConfiguration configuration) + => base.ConfigureClient(configuration) + .AddData(data => data.AddSource(ds => ds + .WithType(t => t.WithKey(d => d.Id)))); + + /// + /// THE WEDGE BECOMES AN ERROR, AND THE ERROR BECOMES A RECOVERY. + /// + /// Pre-fix this test hangs on its fault wait: the mirror re-asks forever, the log fills + /// with "Resync has not converged", and the subscriber is handed neither a value nor a terminal + /// β€” which is precisely why the production incident survived a pod recycle. + /// + [HubFact] + public async Task AResyncThatNeverConverges_FaultsTheMirror_AndAFreshSubscriberRecovers() + { + var host = GetHost(); + var client = GetClient(); + var accessService = host.ServiceProvider.GetRequiredService(); + accessService.SetContext(new AccessContext { ObjectId = "alice", Name = "Alice" }); + + var collectionName = host.GetWorkspace().DataContext.GetTypeSource(typeof(MyData))!.CollectionName; + var clientWorkspace = client.GetWorkspace(); + var clientStream = clientWorkspace + .GetRemoteStream(CreateHostAddress(), new CollectionsReference(collectionName)); + + // The mirror is live and mirrorStreamId is known β€” every injection below is scoped to it. + await clientStream.Should().Within(10.Seconds()).Emit(); + + // Arm the terminal wait BEFORE driving. A store is a ReplaySubject, so subscribing after the + // fault would still replay it; arming first means the assertion cannot be satisfied by + // anything that happened before the drive, and it makes the pre-fix failure a TIMEOUT on the + // thing that never happens rather than a race with the drive loop. + var faulted = clientStream + .Materialize() + .Where(n => n.Kind == NotificationKind.OnError) + .Select(n => n.Exception!) + .Take(1) + .Replay(1); + using var faultedConnection = faulted.Connect(); + + // A burst whose SECOND patch the wire eats β€” the third patch chains onto a version the + // mirror never applied, which is proven gap #1 and earns re-ask #1. + for (var i = 1; i <= 3; i++) + host.Post( + new DataChangeRequest().WithUpdates(new MyData($"doc-{i}", $"value-{i}")), + o => o.WithAccessContext(accessService.Context!)); + + // Then one write per further proven gap. Each iteration waits for the answer to the PREVIOUS + // re-ask to have been destroyed β€” the event, never a delay β€” so the next write is guaranteed + // to be a frame the mirror sees while it still holds no base. + for (var attempt = 1; attempt <= MaxUnansweredResyncs; attempt++) + { + var expected = attempt; + await freshSnapshotDropped + .Where(count => count >= expected) + .Take(1) + .Should().Within(15.Seconds()) + .Emit($"the owner must have ANSWERED re-ask {expected} with a fresh snapshot for the " + + "leg to destroy β€” an unanswered re-ask would mean this test is measuring a " + + "refusal, which StreamResyncConvergenceTest already covers"); + + host.Post( + new DataChangeRequest().WithUpdates(new MyData($"doc-{3 + attempt}", $"value-{3 + attempt}")), + o => o.WithAccessContext(accessService.Context!)); + } + + // THE ASSERTION. The gap proven by that last write must fault the mirror instead of asking a + // fourth time into a leg that has now destroyed three acknowledged snapshots. + var error = await faulted + .Should().Within(20.Seconds()) + .Emit("a mirror whose fresh snapshots keep being lost must SURFACE that, not re-ask " + + "forever: a subscriber that is never told cannot re-establish, which is how one " + + "lost frame became a layout area that never rendered"); + + error.Should().BeOfType( + "the fault must name the non-convergence rather than arrive as some incidental " + + "exception β€” a consumer classifies on the type"); + + Volatile.Read(ref droppedPatches).Should().Be(1, + "the wire loss is what starts the resync β€” without it there is no re-ask to leave " + + "unanswered and this test asserts nothing"); + Volatile.Read(ref droppedFreshSnapshots).Should().BeGreaterThanOrEqualTo(MaxUnansweredResyncs, + "every re-ask must have been ANSWERED and the answer destroyed β€” that is the shape this " + + "test exists for, as opposed to an owner that simply went quiet or refused"); + + // AND THE RECOVERY. The faulted stream is unusable, so the workspace cache must not hand it + // back: a fresh subscriber gets a NEW stream (a new StreamId, therefore untouched by the + // injections above) which subscribes from scratch and converges on the owner's state. This + // is the whole point of faulting rather than waiting. + var recovered = clientWorkspace + .GetRemoteStream(CreateHostAddress(), new CollectionsReference(collectionName)); + + ReferenceEquals(recovered, clientStream).Should().BeFalse( + "a faulted mirror must be evicted from the stream cache, not replayed to the next caller"); + + await recovered + .Where(ci => ci.Value?.Collections.GetValueOrDefault(collectionName) is { } coll + && coll.Instances.Values.OfType() + .Count(dd => dd.Text?.StartsWith("value-") == true) == 3 + MaxUnansweredResyncs) + .Take(1) + .Should().Within(20.Seconds()) + .Emit("the subscriber that learned of the failure must be able to re-establish and see " + + "the owner's complete state β€” surfacing the wedge is only useful if recovery " + + "follows from it"); + } +}