A view catching up got handed a snapshot older than what it had already applied

src/MeshWeaver.Data/Serialization/JsonSynchronizationStream.cs

@@ -661,6 +661,28 @@ void Resubscribe(string reason)
         return reduced;
     }
 
+    /// <summary>
+    /// The frame a resubscribing subscriber is served: the stream's CURRENT state re-asserted as a
+    /// <see cref="ChangeType.Full"/>, stamped with the version that state already carries.
+    /// <para>🚨 MUST be invoked from INSIDE the stream's update turn (i.e. from the
+    /// <c>Update</c> lambda) — that is what makes <c>stream.Current</c> the freshest committed
+    /// state rather than a snapshot bound before the turn was scheduled. Re-asserting a
+    /// pre-turn capture rolls the stream BACKWARD over frames that committed in between, and
+    /// those entries are never re-sent (issue #945). And the version MUST come off the state
+    /// itself — the owner's content clock — never <c>stream.Hub.Version</c>, which is this
+    /// per-subscriber sync hub's unrelated message counter; mixing clocks on one stream either
+    /// sneaks a stale Full past the receive-side monotonicity guard or freezes the mirror by
+    /// making every later content patch look stale. Same rule as
+    /// <c>StandardReducers.PatchJsonElement</c>.</para>
+    /// <returns><c>null</c> when there is nothing to assert yet (the initial subscribe is still
+    /// hydrating) — a no-op update.</returns>
+    /// </summary>
+    internal static ChangeItem<TReduced>? BuildReassertFrame<TReduced>(
+        ISynchronizationStream<TReduced> stream)
+        => stream.Current is { Value: not null } live
+            ? new ChangeItem<TReduced>(live.Value, stream.StreamId, live.Version)
+            : null;
+
     internal static ISynchronizationStream CreateSynchronizationStream<TReduced, TReference>(
         this IWorkspace workspace,
         IMessageDelivery<SubscribeRequest> delivery
@@ -694,9 +716,34 @@ IMessageDelivery<SubscribeRequest> delivery
             logger.LogDebug(
                 "Owner {Owner} already serves stream {StreamId} for {Subscriber} — re-asserting the current snapshot instead of creating a second stream",
                 hub.Address, request.StreamId, request.Subscriber);
-            if (alreadyServing.Current is { Value: not null } snapshot)
+            // 🚨 Read the snapshot IN-TURN and carry ITS version — never a value captured out
+            // here, never this stream's own hub clock. Both were data loss (issue #945):
+            //
+            //   • STALE CAPTURE. `Update` posts an UpdateStreamRequest; the lambda runs LATER on
+            //     the stream's action block. A snapshot bound out here and written back blindly
+            //     (`_ => snapshot`) OVERWRITES every frame that committed in between — the stream
+            //     moves BACKWARD, and because the owner-side outbound JSON cursor is re-based on
+            //     that rolled-back state, the erased entries are never re-sent. Subsequent patches
+            //     keep applying cleanly on top, so the mirror tracks the owner forever at a
+            //     CONSTANT deficit with no error anywhere. Measured under a 288-write burst:
+            //     mirror at 245 entries/v246 → 234 entries/v235, ending 11 entries short of the
+            //     owner while every write acked success.
+            //   • WRONG CLOCK. Content frames on this stream carry the OWNER's data-source clock
+            //     (WorkspaceExtensions.ApplyChanges → primary stream Hub.Version, carried through
+            //     the reduce as ChangeItem.Version). Stamping the per-subscriber SYNC hub's clock
+            //     here mixes two unrelated counters on one stream — the same defect StandardReducers
+            //     documents ("CARRY the base version … NEVER stream.Hub.Version"). It is what lets
+            //     a stale Full slip past the receive-side monotonicity guard that exists to drop
+            //     exactly that; when the sync clock instead runs AHEAD, every later content patch
+            //     is dropped as "stale" and the mirror freezes.
+            //
+            // Re-asserting the CURRENT state at ITS OWN version is still a full authoritative
+            // snapshot — the orphaned mirror re-converges (a Full always lands, and the mirror
+            // that resubscribed is by definition not ahead) — but it can neither erase newer
+            // state nor poison the mirror's version.
+            if (alreadyServing.Current is { Value: not null })
                 alreadyServing.Update(
-                    _ => new ChangeItem<TReduced>(snapshot.Value, alreadyServing.StreamId, alreadyServing.Hub.Version),
+                    _ => BuildReassertFrame(alreadyServing),
                     ex => logger.LogWarning(ex,
                         "Stream {StreamId}: could not re-assert snapshot for resubscribing subscriber {Subscriber}",
                         alreadyServing.StreamId, request.Subscriber));

src/MeshWeaver.Documentation/Data/WhatsNew/2026-08-08-live-views-no-longer-fall-behind.md

@@ -0,0 +1,25 @@
+---
+Name: Live views no longer quietly fall behind under heavy editing
+Category: What's New
+Description: A view watching a node that many people (or agents) were writing to at once could silently stop showing some of the newest entries; it now stays in step with the node.
+Icon: Sparkle
+---
+
+# Live views no longer quietly fall behind under heavy editing
+
+When a lot of writes landed on the same node at once — a busy chat inbox, an
+import, several agents editing in parallel — a view watching that node could
+quietly lose a few of the entries that were being added. Nothing failed: every
+write was accepted and safely stored, the page kept updating, and later entries
+kept appearing. The view simply stayed a few entries short of the truth, and
+stayed that way until it was reopened.
+
+The cause was the recovery step a view takes when it notices it has fallen
+behind: it asks the node to re-send its current state, and that re-send could
+carry a slightly older snapshot than the view had already applied — quietly
+erasing whatever arrived in between. The re-send now always carries the node's
+state as of the moment it is sent, so catching up can only ever move a view
+forward.
+
+Your stored data was never affected: the node itself always held every write.
+This only affected what an already-open view was showing.

test/MeshWeaver.AI.Test/CrossHubPatchAtomicityTest.cs

@@ -44,6 +44,20 @@ namespace MeshWeaver.AI.Test;
 /// persisted. The version echo-suppression only skipped the single latest write, not the lagging
 /// echoes. The fix makes that refresh FORWARD-ONLY (apply a persisted snapshot only when strictly
 /// newer), so the in-RAM commit is authoritative and never moves backward.</para>
+///
+/// <para>SECOND ROOT CAUSE, issue #945 — the same symptom on the READ side. Under CI load this
+/// test kept failing with a handful of entries missing and ZERO write errors, and the owner-truth
+/// probe below proved the owner held all 288: the loss was in the shared-cache MIRROR. A write
+/// burst makes the mesh change feed announce versions faster than a loaded mirror applies them, so
+/// the mirror's version-gated <c>Resubscribe</c> fires; the owner answers on the
+/// <c>alreadyServing</c> path in <c>JsonSynchronizationStream.CreateSynchronizationStream</c> by
+/// re-asserting its snapshot. That re-assert bound <c>Current</c> OUTSIDE the stream's update turn
+/// and wrote it back blindly, stamped with the per-subscriber SYNC hub's clock instead of the
+/// state's own content version — so it rolled the stream BACKWARD over every frame that had
+/// committed in between (measured: 245 entries@v246 → 234 entries@v235), re-based the outbound
+/// JSON cursor on the rolled-back state, and left the mirror trailing the owner by a CONSTANT
+/// deficit forever. The fix reads the snapshot in-turn and carries its own version
+/// (<c>BuildReassertFrame</c>, pinned by <c>ResubscribeReassertFrameTest</c>).</para>
 /// </summary>
 public class CrossHubPatchAtomicityTest(ITestOutputHelper output) : AITestBase(output)
 {
@@ -90,8 +104,31 @@ await client.GetWorkspace().GetMeshNodeStream(path)
                     .Where(n => n.Content is MeshThread)
                     .Take(1).Timeout(TimeSpan.FromSeconds(30)).ToTask(ct);
 
+            // INSTRUMENT: record every emission the shared-cache mirror delivers, timestamped, so a
+            // failure distinguishes the three shapes on its own instead of leaving it to inference:
+            // went BACKWARD (a stale snapshot re-applied over newer state — the #945 defect),
+            // stopped advancing (never converged), or merely arrived LATE (converged past the settle
+            // bound — a slow box, not a loss). No logging pipeline involved.
+            // 🚨 `using` + an explicit OnError, both load-bearing on the FAILURE path — the one
+            // path this instrument exists for. Without `using`, Assert.Fail below unwinds past a
+            // plain Dispose() and leaves the subscription pending into teardown (the "left Observe
+            // subscriptions pending past the Quiescing budget" class). Without an OnError handler,
+            // a stream fault would be rethrown on the producer's thread as an unhandled exception,
+            // replacing the very diagnostics the failure message is built from; instead the fault
+            // is captured and REPORTED alongside the sequence recorded so far.
+            var clock = System.Diagnostics.Stopwatch.StartNew();
+            var observed = new System.Collections.Concurrent.ConcurrentQueue<(long Version, int Keys, long Ms)>();
+            Exception? watchError = null;
+            using var watch = Mesh.GetWorkspace().GetMeshNodeStream(path)
+                .Subscribe(
+                    n => observed.Enqueue(
+                        (n.Version, (n.Content as MeshThread)?.PendingUserMessages.Count ?? 0,
+                         clock.ElapsedMilliseconds)),
+                    ex => watchError = ex);
+
             var allKeys = new List<string>(total);
             var writes = new List<IObservable<MeshNode>>(total);
+            var writeErrors = new System.Collections.Concurrent.ConcurrentBag<string>();
             for (var m = 0; m < mirrors; m++)
             {
                 var client = clients[m];
@@ -114,7 +151,11 @@ await client.GetWorkspace().GetMeshNodeStream(path)
                         })
                         .Take(1)
                         // A failed write must surface as a MISSING key (the loss), not abort the burst.
-                        .Catch((Exception _) => Observable.Empty<MeshNode>()));
+                        .Catch((Exception ex) =>
+                        {
+                            writeErrors.Add($"{key}: {ex.GetType().Name}: {ex.Message}");
+                            return Observable.Empty<MeshNode>();
+                        }));
                 }
             }
 
@@ -141,11 +182,47 @@ await Mesh.GetWorkspace().GetMeshNodeStream(path)
             catch (TimeoutException)
             {
                 var missing = allKeys.Where(k => !finalKeys.Contains(k)).ToArray();
+                var errs = writeErrors.ToArray();
+                // OWNER TRUTH: a brand-new client opens a FRESH remote subscription (bypassing the
+                // shared cache), so the initial frame it receives is the owner's authoritative
+                // Current. If the key is present here but absent above, the loss is on the READ
+                // (mirror) side; if it is absent here too, the owner genuinely dropped it.
+                var probe = GetClient();
+                var ownerNode = await probe.GetWorkspace().GetMeshNodeStreamBypassCache(path)
+                    .Where(n => n.Content is MeshThread)
+                    .Take(1).Timeout(TimeSpan.FromSeconds(30)).ToTask(ct);
+                var ownerKeys = (ownerNode.Content as MeshThread)!.PendingUserMessages.Keys.ToImmutableHashSet();
+                var ownerMissing = allKeys.Where(k => !ownerKeys.Contains(k)).ToArray();
+                var seq = observed.ToArray();
+                var regressions = seq.Zip(seq.Skip(1))
+                    .Where(p => p.Second.Keys < p.First.Keys)
+                    .Select(p => $"{p.First.Keys}@v{p.First.Version}(t{p.First.Ms}ms)"
+                                 + $"->{p.Second.Keys}@v{p.Second.Version}(t{p.Second.Ms}ms)")
+                    .ToArray();
+                // Did the mirror converge at all — before or after the settle bound expired?
+                var converged = seq.LastOrDefault(e => e.Keys >= total);
                 Assert.Fail(
-                    $"Round {round}: owner dropped {missing.Length} of {total} concurrent cross-hub adds "
-                    + $"(present={finalKeys.Count}). Missing (first 20): [{string.Join(",", missing.Take(20))}]. "
-                    + "The owner-side patch apply is non-atomic — a concurrent apply read a stale base and "
-                    + "overwrote a sibling writer's just-added key.");
+                    $"Round {round}: {missing.Length} of {total} concurrent cross-hub adds are not visible "
+                    + $"(present={finalKeys.Count}). Missing: [{string.Join(",", missing)}]. "
+                    + $"WRITE ERRORS ({errs.Length}): [{string.Join(" | ", errs)}]. "
+                    + $"OWNER-TRUTH probe: version={ownerNode.Version} present={ownerKeys.Count} "
+                    + $"missing=[{string.Join(",", ownerMissing)}]. "
+                    + (watchError is null
+                        ? ""
+                        : $"MIRROR STREAM FAULTED: {watchError.GetType().Name}: {watchError.Message} "
+                          + "(the sequence below stops there). ")
+                    + $"MIRROR SEQUENCE ({seq.Length} emissions), REGRESSIONS ({regressions.Length}): "
+                    + $"[{string.Join(",", regressions)}]; converged="
+                    + (converged.Keys >= total ? $"YES at t{converged.Ms}ms" : "NEVER")
+                    + "; tail="
+                    + $"[{string.Join(",", seq.TakeLast(6).Select(e => $"{e.Keys}@v{e.Version}(t{e.Ms}ms)"))}]. "
+                    + "READ THE DIAGNOSTICS BEFORE THEORISING: write errors non-empty ⇒ the write failed, "
+                    + "not the merge. Owner-truth missing ⇒ the OWNER apply lost it (a non-atomic "
+                    + "read-modify-write). Owner-truth complete but entries missing here ⇒ the shared-cache "
+                    + "MIRROR is behind — and then: a REGRESSION names the exact frame where the mirror "
+                    + "moved BACKWARD (the #945 stale resubscribe re-assert, a real loss), while "
+                    + "converged=YES with no regression means the mirror got everything but only after "
+                    + "the settle bound — a slow box, not a lost write.");
             }
 
             finalKeys.Count.Should().BeGreaterThanOrEqualTo(total,