diff --git a/src/MeshWeaver.Documentation/Data/WhatsNew/2026-08-21-activity-survives-a-write-collision.md b/src/MeshWeaver.Documentation/Data/WhatsNew/2026-08-21-activity-survives-a-write-collision.md new file mode 100644 index 0000000000..1fc1c3f5f0 --- /dev/null +++ b/src/MeshWeaver.Documentation/Data/WhatsNew/2026-08-21-activity-survives-a-write-collision.md @@ -0,0 +1,13 @@ +--- +Name: Your activity is recorded even when two pages update it at once +Category: Fix +Description: A visit that collided with another update to the same record used to be dropped; the record is now re-read and the visit applied. +Icon: Sparkle +Order: -20260821 +--- + +# Your activity is recorded even when two pages update it at once + +Every page you open updates one small record of what you have looked at and when. When two of those updates arrived at the same moment, one of them was refused β€” correctly, because it had been written against a version that had already moved on β€” and then thrown away. The visit was simply not recorded. Nothing failed visibly, which is the awkward part: the history looked complete while occasionally missing an entry. + +A refused update now waits for the newer version of the record to arrive, re-reads it, and applies the visit on top. The count and the "last opened" time come out right whether one page updated the record or several did at once. diff --git a/src/MeshWeaver.Mesh.Contract/MeshNodeStreamExtensions.cs b/src/MeshWeaver.Mesh.Contract/MeshNodeStreamExtensions.cs index 70b62b7463..4691b48151 100644 --- a/src/MeshWeaver.Mesh.Contract/MeshNodeStreamExtensions.cs +++ b/src/MeshWeaver.Mesh.Contract/MeshNodeStreamExtensions.cs @@ -1,4 +1,5 @@ ο»Ώusing System.Diagnostics; +using System.Reactive.Concurrency; using System.Reactive.Disposables; using System.Reactive.Linq; using System.Text.Json; @@ -112,6 +113,91 @@ public sealed class MeshNodeStreamHandle : IObservable // patch) and every other NACK code (validation/RLS/NotFound are terminal verdicts). private const int MaxOwnerDisposingReenqueues = 2; + // 🚨 How long a CONFLICT re-attempt waits for this hub's mirror to carry state the owner has + // not already refused. Not a retry interval and not a backoff β€” it is the bound on ONE wait + // for a fact that is already on its way: the owner committed the winning write BEFORE it + // NACK'd us, so the newer state is in flight to the mirror we are about to read. The wait is + // for the arrival, not for a repeat. Generous against a sub-second propagation, and short + // enough to sit well inside the caller's own budget. + private static readonly TimeSpan ConflictRebaseBound = TimeSpan.FromSeconds(5); + + /// + /// 🚨 The emission a (re)attempt rebuilds its patch from β€” and the whole of the #1910 fix. + /// + /// The defect. A cross-hub stream.Update reads this hub's MIRROR, runs the + /// lambda against it, and ships a merge patch carrying the base values it diffed against. The + /// owner three-way-merges: any key whose live value has moved since that base is REFUSED, and + /// when nothing lands it answers Conflict β€” "re-read and re-apply". UpdateRemote + /// does re-enqueue on that verdict, and it does re-run the lambda rather than re-post the + /// patch. But it re-ran the lambda against mirror.Take(1), i.e. against whatever the + /// mirror holds AT THAT INSTANT β€” which, immediately after a NACK, is very often still the + /// version the owner just refused. The lambda then recomputes the same values from the same + /// base, ExtractBaseValues produces the same base, and the owner refuses identically. + /// Three attempts, three identical patches, then a terminal MeshNodeStreamException and + /// the caller's write is gone. A retry whose input cannot change is not a retry. + /// + /// The rule. A Conflict means the owner is provably AHEAD of + /// β€” it minted a version for the write that beat us + /// before it answered us β€” so the state that makes the re-attempt converge already exists and + /// is on its way here. Wait for it, then rebase on it. + /// + /// Conflict only. OwnerDisposing / OwnerNotReady re-enqueues pass + /// 0 and are unaffected: those say the patch never reached a merge at all, so no newer + /// version exists and waiting for one would burn the bound for nothing. + /// + /// Never parks. If the mirror does not advance within + /// the re-attempt proceeds against what it has β€” exactly the + /// previous behaviour β€” and says so, so "the write was + /// refused AND the mirror never caught up" is a reported fact rather than a silent + /// re-refusal. This is why the change can only improve on the old behaviour: it converges + /// where the old shape could not, and degrades to the old shape where it cannot. + /// + /// Static, with the mirror and the scheduler as seams, so the re-read rule is asserted + /// deterministically β€” no hub, no cluster, no wall clock. + /// + /// This hub's view of the node β€” replays its current state to a new + /// subscriber and emits again as the owner's commits arrive. + /// The version the owner refused, or 0 for a first + /// attempt (and for the non-staleness re-enqueue codes), which reads the mirror as before. + /// Called with when the + /// bound elapsed and the re-attempt fell back to un-advanced state. + /// Timer seam for tests. + internal static IObservable RebaseSource( + IObservable mirror, + long refusedBaseVersion, + Action onStaleMirror, + IScheduler? scheduler = null) + => refusedBaseVersion <= 0 + // A first attempt reads the mirror exactly as it always did β€” the ordinary write path + // gains no filter, no timer and no second subscription. + ? mirror.Take(1) + : mirror + .Where(node => node.Version > refusedBaseVersion) + .Take(1) + .Timeout( + ConflictRebaseBound, + Observable.Defer(() => + { + onStaleMirror(refusedBaseVersion); + return mirror.Take(1); + }), + scheduler ?? Scheduler.Default) + // 🚨 An EMPTY completion must not reach the caller. Filtering introduces a + // completion the un-filtered shape could not produce: a mirror that ends (its hub + // torn down) while holding only the refused version completes this source with no + // value, the write's observer is never called, and the caller waits on a pipeline + // that has already finished. A source that cannot answer must SAY so β€” the same + // rule the compile pipeline's DefaultIfEmpty totality guard applies. + .Select(node => (MeshNode?)node) + .DefaultIfEmpty(null) + .SelectMany(node => node is null + ? Observable.Throw(new InvalidOperationException( + $"Update aborted: the owner refused this write as stale at version " + + $"{refusedBaseVersion} and the mirror ended before carrying anything " + + "newer, so there is no state to re-apply against. The write did NOT " + + "land; re-issue it.")) + : Observable.Return(node)); + internal MeshNodeStreamHandle(IWorkspace workspace, string? path = null, IMeshNodeStreamCache? cache = null, bool bypassCache = false) { @@ -1086,7 +1172,12 @@ void EmitOnce(MeshNode value) /// Re-enqueue depth: 0 for a caller write, incremented by the /// late OwnerDisposing-NACK re-enqueue, capped at . /// Carried as a parameter β€” never static state. - private IObservable UpdateRemote(Func update, int attempt = 0) + /// On a CONFLICT re-enqueue, the node version the owner + /// refused this write against, so the re-attempt rebases on something newer + /// (). 0 for a caller write and for the re-enqueue codes that never + /// reached a merge. + private IObservable UpdateRemote( + Func update, int attempt = 0, long refusedBaseVersion = 0) => Observable.Create(observer => { var diagLogger = _workspace.Hub.ServiceProvider @@ -1130,14 +1221,26 @@ private IObservable UpdateRemote(Func update, int // running the user lambda β€” the lambda needs a non-null current // to diff against. A 30 s outer timeout bounds the wait so a // missing per-node hub surfaces with a precise TimeoutException. - var initialSub = remoteStream - .Timeout(TimeSpan.FromSeconds(30)) - .Where(change => change.Value is not null) - .Take(1) + // + // 🚨 …and on a CONFLICT re-attempt, wait for state the owner has not already + // refused. See RebaseSource: re-running the lambda against the SAME mirror emission + // recomputes the same values from the same base and is refused identically, so the + // re-enqueue cannot converge (#1910). + var initialSub = RebaseSource( + remoteStream + .Timeout(TimeSpan.FromSeconds(30)) + .Where(change => change.Value is not null) + .Select(change => change.Value!), + refusedBaseVersion, + staleVersion => diagLogger?.LogWarning( + "[UpdateRemote] STALE_MIRROR hub={Hub} target={Path} attempt={Attempt} β€” the " + + "owner refused this write as stale at version {Version}, but this hub's " + + "mirror did not advance past it within {Bound}; rebuilding the patch " + + "against the state it has. The re-attempt may be refused again", + _workspace.Hub.Address, _path, attempt, staleVersion, ConflictRebaseBound)) .Subscribe( - change => + current => { - var current = change.Value!; try { var updated = update(current); @@ -1353,9 +1456,16 @@ or MeshNodeErrorCode.OwnerNotReady or MeshNodeErrorCode.Conflict && attempt < MaxOwnerDisposingReenqueues) { + // 🚨 Conflict alone carries a stale BASE β€” see RebaseSource. + // OwnerDisposing / OwnerNotReady mean the patch never reached + // a merge, so there is no newer version to wait for and + // waiting would only burn the bound. + var lateRebaseFrom = lateErr.Code == MeshNodeErrorCode.Conflict + ? current.Version + : 0; diagLogger?.LogWarning( - "[UpdateRemote] LATE_NACK_REENQUEUE hub={Hub} target={Path} attempt={Attempt} β€” owner disposed before the merge turn; re-enqueueing the original update against the fresh activation", - _workspace.Hub.Address, _path, attempt + 1); + "[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); // 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 @@ -1365,7 +1475,7 @@ or MeshNodeErrorCode.Conflict ? accessServiceAtEntry.SwitchAccessContext(capturedContextAtEntry) : null) { - UpdateRemote(update, attempt + 1).Subscribe( + UpdateRemote(update, attempt + 1, lateRebaseFrom).Subscribe( _ => { }, ex2 => diagLogger?.LogWarning(ex2, "[UpdateRemote] LATE_NACK_REENQUEUE failed hub={Hub} target={Path} attempt={Attempt}", @@ -1450,14 +1560,28 @@ or MeshNodeErrorCode.OwnerNotReady or MeshNodeErrorCode.Conflict && attempt < MaxOwnerDisposingReenqueues) { + // 🚨 A Conflict says the owner is provably PAST + // the base we diffed against β€” it committed the + // winning write before answering us. Re-running + // the lambda against the mirror as it stands + // recomputes the same patch from the same base + // and is refused identically, which is why the + // re-enqueue could not converge (#1910). Carry + // the refused version so the re-attempt rebases + // on state newer than it. OwnerDisposing / + // OwnerNotReady never reached a merge, so there + // is nothing newer to wait for β€” they pass 0. + var rebaseFrom = err.Code == 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 against the fresh activation", - _workspace.Hub.Address, _path, err.Code, attempt + 1); + "[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); using (accessServiceAtEntry is not null && capturedContextAtEntry is not null ? accessServiceAtEntry.SwitchAccessContext(capturedContextAtEntry) : null) { - composite.Add(UpdateRemote(update, attempt + 1) + composite.Add(UpdateRemote(update, attempt + 1, rebaseFrom) .Subscribe(observer.OnNext, observer.OnError, observer.OnCompleted)); } return; diff --git a/test/MeshWeaver.Query.Test/ActivityWriteSurvivesConflictTest.cs b/test/MeshWeaver.Query.Test/ActivityWriteSurvivesConflictTest.cs new file mode 100644 index 0000000000..f753be9fb3 --- /dev/null +++ b/test/MeshWeaver.Query.Test/ActivityWriteSurvivesConflictTest.cs @@ -0,0 +1,227 @@ +using System; +using System.Collections.Generic; +using System.Reactive.Linq; +using System.Reactive.Subjects; +using MeshWeaver.Mesh; +using Microsoft.Reactive.Testing; +using Xunit; + +namespace MeshWeaver.Query.Test; + +/// +/// 🚨 A cross-hub write refused as STALE must converge on the retry β€” issue #1910. +/// +/// The incident. ActivityTracking failed to record a navigation: +/// MeshNode Conflict at 'rbuergi/_UserActivity/rbuergi': cross-hub write refused: 3 field(s) +/// changed on the owner since the writer's base and nothing was applied β€” re-read and +/// re-apply. The activity entry for that moment simply does not exist. The issue proposed +/// "adding a retry-with-refresh loop, or switching to an append-only pattern". +/// +/// What was actually there, and why it could not work. A retry-with-refresh loop +/// already existed. UpdateRemote re-enqueues on Conflict (and re-runs the update +/// lambda rather than re-posting the patch β€” the #1814 defect, already fixed). What it did NOT do +/// is guarantee that the re-run sees anything DIFFERENT. It rebuilt from mirror.Take(1) β€” +/// whatever this hub's mirror holds at that instant β€” and immediately after a NACK that is very +/// often still the version the owner just refused. The lambda then recomputes the same values +/// from the same base, ExtractBaseValues emits the same base, and the owner refuses +/// identically. Three attempts, three identical patches, then a terminal +/// MeshNodeStreamException and the write is gone. A retry whose input cannot change is +/// not a retry β€” the same shape as #1814 one level down: there the patch was re-posted, here +/// the base is re-read from a source that had not moved. +/// +/// Why waiting is sound rather than hopeful. A Conflict is the owner stating +/// that it is AHEAD of the writer's base β€” it minted a version for the write that beat us BEFORE +/// it answered us. So the state that makes the re-attempt converge already exists and is on its +/// way to this mirror. The re-attempt waits for that specific fact (a version strictly greater +/// than the one refused), bounded, and falls back to the old behaviour if it never arrives β€” +/// which is why the change can only improve on what was there. +/// +/// Deterministic: the mirror and the clock are both seams. No hub, no cluster, no +/// wall-clock sleep β€” the interleaving that makes this defect rare in a test and common in +/// production is CONSTRUCTED here rather than raced for. +/// +public class ActivityWriteSurvivesConflictTest +{ + /// The activity node from the incident. + private const string Path = "rbuergi/_UserActivity/rbuergi"; + + /// + /// A stand-in for this hub's mirror: replays its current state to a new subscriber and emits + /// again as the owner's commits arrive β€” the shape UpdateRemote reads. + /// + private static ReplaySubject Mirror(long version) + { + var mirror = new ReplaySubject(1); + mirror.OnNext(NodeAt(version)); + return mirror; + } + + private static MeshNode NodeAt(long version) => + MeshNode.FromPath(Path) with { NodeType = "UserActivity", Version = version }; + + /// + /// 🚨 THE regression pin. The owner refused version 10 as stale; the re-attempt must not + /// rebuild from version 10. + /// + /// Against origin/main this fails immediately with 10 β€” see + /// , which + /// is that shape written out. + /// + [Fact] + public void A_conflict_retry_rebases_on_state_newer_than_the_version_that_was_refused() + { + var scheduler = new TestScheduler(); + var mirror = Mirror(version: 10); + var stale = new List(); + var seen = new List(); + + MeshNodeStreamHandle + .RebaseSource(mirror, refusedBaseVersion: 10, stale.Add, scheduler) + .Subscribe(node => seen.Add(node.Version)); + + // The mirror still holds exactly what the owner refused. Rebuilding from it would produce + // the same patch with the same base β€” refused again, and again, until the budget is spent + // and the caller's write is dropped. + seen.Should().BeEmpty( + "re-running the lambda against the very version the owner refused recomputes the same " + + "patch from the same base β€” the retry cannot converge, which is how a tracked " + + "activity ends up silently absent"); + + // The winning writer's commit arrives at the mirror β€” it was already in flight, because + // the owner committed it before it NACK'd us. + scheduler.AdvanceBy(TimeSpan.FromSeconds(1).Ticks); + mirror.OnNext(NodeAt(11)); + + seen.Should().Equal([11L], + "the re-attempt rebases on the state the owner actually has, which IS the 're-read and " + + "re-apply' the refusal asked for"); + stale.Should().BeEmpty("the mirror advanced, so nothing degraded"); + } + + /// + /// NEGATIVE CONTROL β€” the shape UpdateRemote had, written out. It is one operator, and + /// it is the whole defect: a re-attempt is handed the refused version and rebuilds from it. + /// Keeping it here means the regression above cannot quietly become vacuous. + /// + [Fact] + public void NegativeControl_TheOldShapeRebuildsFromTheVeryVersionTheOwnerRefused() + { + var mirror = Mirror(version: 10); + var seen = new List(); + + // origin/main: `remoteStream.Where(c => c.Value is not null).Take(1)`. + mirror.Take(1).Subscribe(node => seen.Add(node.Version)); + + seen.Should().Equal([10L], + "this is the defect: the retry reads the mirror as it stands, which right after a " + + "Conflict NACK is still the version the owner refused β€” so the recomputed patch is " + + "byte-identical to the one that was just rejected"); + } + + /// + /// The bound, and the reason it is a bound and not a wait: a mirror that never advances must + /// not park the caller. The re-attempt proceeds against what it has β€” exactly the previous + /// behaviour β€” and REPORTS that it did, so "refused as stale AND the mirror never caught up" + /// is a fact in the log rather than a second silent refusal. + /// + [Fact] + public void A_mirror_that_never_advances_is_not_parked_on_and_the_degradation_is_reported() + { + var scheduler = new TestScheduler(); + var mirror = Mirror(version: 10); + var stale = new List(); + var seen = new List(); + + MeshNodeStreamHandle + .RebaseSource(mirror, refusedBaseVersion: 10, stale.Add, scheduler) + .Subscribe(node => seen.Add(node.Version)); + + scheduler.AdvanceBy(TimeSpan.FromSeconds(4).Ticks); + seen.Should().BeEmpty("still inside the bound"); + + scheduler.AdvanceBy(TimeSpan.FromSeconds(2).Ticks); + + seen.Should().Equal([10L], + "a fresher base never arrived, so the re-attempt does the best it can rather than " + + "hanging β€” never worse than the behaviour it replaces"); + stale.Should().Equal([10L], + "…and says so, because a re-attempt that is about to be refused for the same reason " + + "is exactly the thing that was invisible"); + } + + /// + /// A FIRST attempt has no refused base and must not wait for anything β€” the fix costs the + /// normal write path nothing. This is the case every cross-hub write in the mesh takes. + /// + [Fact] + public void A_first_attempt_reads_the_mirror_immediately() + { + var scheduler = new TestScheduler(); + var seen = new List(); + + MeshNodeStreamHandle + .RebaseSource(Mirror(version: 10), refusedBaseVersion: 0, _ => { }, scheduler) + .Subscribe(node => seen.Add(node.Version)); + + seen.Should().Equal([10L], + "no timer, no gate β€” the ordinary write path is untouched, which is what makes this " + + "safe to put on every cross-hub Update"); + } + + /// + /// 🚨 The hazard the FILTER introduces, closed in the same change. Waiting for a newer version + /// creates a completion the un-filtered shape could not produce: a mirror that ENDS (its hub + /// torn down) while holding only the refused version would complete this source with no value + /// at all β€” the write's observer never fires and the caller waits on a pipeline that has + /// already finished. A source that cannot answer must say so; an empty completion is the one + /// outcome a caller cannot distinguish from "still working". + /// + [Fact] + public void A_mirror_that_ends_without_newer_state_errors_rather_than_completing_empty() + { + var scheduler = new TestScheduler(); + var mirror = Mirror(version: 10); + var seen = new List(); + Exception? error = null; + var completed = false; + + MeshNodeStreamHandle + .RebaseSource(mirror, refusedBaseVersion: 10, _ => { }, scheduler) + .Subscribe(node => seen.Add(node.Version), ex => error = ex, () => completed = true); + + // The owning hub goes away before anything newer arrives. + mirror.OnCompleted(); + + seen.Should().BeEmpty(); + completed.Should().BeFalse( + "an empty completion is indistinguishable from 'still working' to the caller β€” it is " + + "how a write turns into a hang instead of a failure"); + error.Should().NotBeNull( + "the re-attempt has nothing to rebase on and must terminate the caller, not strand it"); + error!.Message.Should().Contain("did NOT land", + "…and say plainly that the write is gone, so the caller can re-issue it"); + } + + /// + /// 🚨 The distinction that keeps the bound from being burned for nothing: only + /// Conflict means "the owner moved past you". OwnerDisposing / + /// OwnerNotReady mean the patch never reached a merge at all β€” no newer version was + /// minted, so there is nothing to wait for. Those re-enqueues pass 0 and behave exactly + /// as they did. + /// + [Fact] + public void A_non_staleness_reenqueue_does_not_wait_for_a_version_that_will_never_come() + { + var scheduler = new TestScheduler(); + var seen = new List(); + + // What an OwnerDisposing / OwnerNotReady re-enqueue passes. + MeshNodeStreamHandle + .RebaseSource(Mirror(version: 10), refusedBaseVersion: 0, _ => { }, scheduler) + .Subscribe(node => seen.Add(node.Version)); + + seen.Should().Equal([10L], + "a patch that never reached a merge left the owner's version untouched β€” waiting for " + + "it to advance would spend the whole bound before every one of these retries"); + } +} diff --git a/test/MeshWeaver.Query.Test/MeshWeaver.Query.Test.csproj b/test/MeshWeaver.Query.Test/MeshWeaver.Query.Test.csproj index 89774b8599..946e6e61c0 100644 --- a/test/MeshWeaver.Query.Test/MeshWeaver.Query.Test.csproj +++ b/test/MeshWeaver.Query.Test/MeshWeaver.Query.Test.csproj @@ -12,6 +12,9 @@ + +