Mirror snapshots carry the cache Sequence they reflect

ObserveAsync, ReactiveAsync and RefreshAsync read a snapshot without a
Sequence, so an older change admitted before the read but delivered after it
overwrote the newer value.

CultCache.GetWithSequence(key) returns the document and the cache's Sequence,
read together under the gate. A cache-backed CultMesh handle passes it to its
mirrors; each mirror subscribes first, then adopts the sequenced read under its
own gate and takes that Sequence as applied, so a straggler at or below it
drops. RefreshAsync on both mirrors uses the same adoption path. Public Mesh
surface unchanged. The read is non-generic because the Mesh route needs the
untyped document for same-schema conversion.

Contract: only Watch/WatchRecord changes carry Sequence; OnUpdate-derived
CultNet streams have no ordering or stale protection; schema-alias handles and
removals are not stale-protected; the sequence constructor argument changed
CultCacheDocumentChange<T>'s binary signature.

Tests: ObserveAfterNewerSnapshotIgnoresStraggler and
RefreshIgnoresStragglerOlderThanSnapshot (observed and reactive cases) fail on
4562340; SequencedReadReflectsEveryChangeAtOrBelowItsSequence.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

src/GameCult.Caching/Contracts/cultcache-store-composition.md

@@ -44,9 +44,19 @@ which route wrote it.
   single file keeps staged writes dirty and does not drop loaded records.
 - A load publishes changes to `Watch` subscribers and fires `OnUpdate`. A local
   write or commit publishes to `Watch` only; `OnUpdate` fires for loads alone.
-- Every change carries a `Sequence`: a per-cache, in-memory number assigned
-  when the cache admits the change, under its gate. It increases in admission
-  order and is not persisted.
+- Changes published through `Watch`/`WatchRecord` carry a `Sequence`: a
+  per-cache, in-memory number assigned when the cache admits the change, under
+  its gate. It increases in admission order and is not persisted.
+  `OnUpdate(previous, current)` carries none. Streams CultNet derives from
+  `OnUpdate` (`CultNetDatabase.WatchAllChanges`, the subscription server,
+  database-backed Mesh handles) have no ordering guarantee and no stale
+  protection.
+- `GetWithSequence(key)` returns the document and the cache's current
+  `Sequence`, read together under the gate: every change with a `Sequence` at
+  or below it is reflected in the document.
+- Adding the optional `sequence` constructor argument to
+  `CultCacheDocumentChange<T>` changed its binary signature; no external
+  constructor is known.
 - Observers and `OnUpdate` run after the cache releases its gate, never under
   it: an observer may read or write the cache from any thread. Each call
   publishes exactly its own changes, on its own thread, after releasing the gate
@@ -55,8 +65,15 @@ which route wrote it.
   publishes its own changes before that write returns. A direct `PullAll` on an
   attached store is such a call.
 - Cross-thread delivery order is not guaranteed. A consumer that keeps a latest
-  value per record ignores a change whose `Sequence` is lower than the one it
-  already applied for that record.
+  value per record ignores a change whose `Sequence` is not above the one it
+  already applied for that record. It subscribes first, then takes
+  `GetWithSequence` and adopts that `Sequence` as applied; a change delivered
+  after subscribing with a `Sequence` at or below the read is already in the
+  document and is dropped. CultMesh cache-record mirrors (`ObserveAsync`,
+  `ReactiveAsync`, and `RefreshAsync` on both) do this.
+- Not stale-protected: schema-alias handles (`AsSchemaAlias`) get no
+  `Sequence`; removals do not advance a mirror's applied `Sequence`, and the
+  mirrors do not surface removals.
 - An `OnUpdate` handler exception is rethrown to that caller after all of that
   call's changes are delivered (an `AggregateException` if several threw). If
   the call itself failed, its own exception is rethrown and handler exceptions

src/GameCult.Caching/CultCache.cs

@@ -1409,6 +1409,13 @@ namespace GameCult.Caching
             return Get(key) as T;
         }
 
+        // Every change whose Sequence is at or below the returned one is reflected in the returned document.
+        public (object? Document, long Sequence) GetWithSequence(CultRecordKey key)
+        {
+            lock (_gate)
+                return (_entries.TryGetValue(key.Value, out var stored) ? stored.Document : null, _sequence);
+        }
+
         public bool TryGet<T>(CultRecordKey key, out T? document) where T : class
         {
             document = Get<T>(key);

src/GameCult.Mesh/CultMesh.cs

@@ -1531,7 +1531,12 @@ namespace GameCult.Mesh
             return new CultMeshDocumentHandle<TDocument>(
                 BindLiveFeed(context, feed),
                 async value => await cache.UpsertAsync(value, new CultRecordHandle<TDocument>(key)).ConfigureAwait(false),
-                () => cache.WatchRecord<TDocument>(key));
+                () => cache.WatchRecord<TDocument>(key),
+                () =>
+                {
+                    var (document, sequence) = cache.GetWithSequence(key);
+                    return (RequireDocument<TDocument>(document, key), sequence);
+                });
         }
 
         /// <summary>
@@ -3370,13 +3375,15 @@ namespace GameCult.Mesh
         }
 
         private static TDocument ReadRequired<TDocument>(CultCache cache, CultRecordKey key)
+            where TDocument : class =>
+            RequireDocument<TDocument>(cache.Get(key), key);
+
+        private static TDocument RequireDocument<TDocument>(object? untyped, CultRecordKey key)
             where TDocument : class
         {
-            var document = cache.Get<TDocument>(key);
-            if (document != null)
+            if (untyped is TDocument document)
                 return document;
 
-            var untyped = cache.Get(key);
             if (untyped != null && IsSameCultDocumentSchema<TDocument>(untyped.GetType()))
                 return ConvertUntypedDocument<TDocument>(untyped);
 

src/GameCult.Mesh/CultMeshPrimitives.cs

@@ -929,6 +929,7 @@ namespace GameCult.Mesh
         private readonly Func<TDocument, Task>? _submitPrediction;
         // Set only over a CultCache record, whose changes carry the cache's Sequence.
         private readonly Func<Observable<CultCacheDocumentChange<TDocument>>>? _recordChanges;
+        private readonly Func<(TDocument Document, long Sequence)>? _sequencedLatest;
 
         /// <summary>Creates a document handle from a Verse-bound live feed.</summary>
         public CultMeshDocumentHandle(
@@ -944,10 +945,12 @@ namespace GameCult.Mesh
         internal CultMeshDocumentHandle(
             CultMeshBoundLiveFeed<CultMeshDocumentQueryParameters, TDocument> feed,
             Func<TDocument, Task> replace,
-            Func<Observable<CultCacheDocumentChange<TDocument>>> recordChanges)
+            Func<Observable<CultCacheDocumentChange<TDocument>>> recordChanges,
+            Func<(TDocument Document, long Sequence)> sequencedLatest)
             : this(feed, replace)
         {
             _recordChanges = recordChanges;
+            _sequencedLatest = sequencedLatest;
         }
 
         /// <summary>Gets the semantic document id.</summary>
@@ -1015,6 +1018,24 @@ namespace GameCult.Mesh
                 .Where(change => change.Document != null)
                 .Subscribe(change => onNext(change.Document!, change.Sequence));
 
+        // A mirror subscribes first, then adopts under its own gate. A cache-backed handle reads synchronously there, so
+        // its Sequence is at or above every change the mirror has applied and a later straggler at or below it drops.
+        internal async Task<T> AdoptLatestAsync<T>(object gate, Func<TDocument, long?, T> adopt)
+        {
+            if (_sequencedLatest != null)
+            {
+                lock (gate)
+                {
+                    var (document, sequence) = _sequencedLatest();
+                    return adopt(document, sequence);
+                }
+            }
+
+            var latest = await LatestAsync().ConfigureAwait(false);
+            lock (gate)
+                return adopt(latest, null);
+        }
+
         /// <summary>
         /// Projects one typed document field into a stable reactive state pointer. The expression is
         /// structural metadata: only member access and constant collection indices are accepted.
@@ -1085,9 +1106,8 @@ namespace GameCult.Mesh
         /// <summary>Creates a read-only in-memory mirror updated by the document watch.</summary>
         public async Task<CultMeshObservedDocument<TDocument>> ObserveAsync()
         {
-            var current = CloneDocument(await LatestAsync().ConfigureAwait(false));
-            var observed = new CultMeshObservedDocument<TDocument>(this, current);
-            observed.Start();
+            var observed = new CultMeshObservedDocument<TDocument>(this);
+            await observed.StartAsync().ConfigureAwait(false);
             return observed;
         }
 
@@ -1224,12 +1244,10 @@ namespace GameCult.Mesh
         private long _appliedSequence;
         private bool _disposed;
 
-        internal CultMeshObservedDocument(
-            CultMeshDocumentHandle<TDocument> document,
-            TDocument current)
+        internal CultMeshObservedDocument(CultMeshDocumentHandle<TDocument> document)
         {
             Document = document ?? throw new ArgumentNullException(nameof(document));
-            _current = current ?? throw new ArgumentNullException(nameof(current));
+            _current = null!;
         }
 
         /// <summary>Gets the underlying read/watch handle.</summary>
@@ -1251,23 +1269,31 @@ namespace GameCult.Mesh
             }
         }
 
-        internal void Start()
+        internal async Task StartAsync()
         {
             _subscription = Document.WatchSequenced(ApplyCanonicalSnapshot);
+            try
+            {
+                await RefreshAsync().ConfigureAwait(false);
+            }
+            catch
+            {
+                Dispose();
+                throw;
+            }
         }
 
         /// <summary>Reads and adopts a fresh canonical snapshot.</summary>
-        public async Task<TDocument> RefreshAsync()
+        public Task<TDocument> RefreshAsync()
         {
             ThrowIfDisposed();
-            var latest = CultMeshDocumentHandle<TDocument>.CloneDocument(
-                await Document.LatestAsync().ConfigureAwait(false));
-            lock (_gate)
+            return Document.AdoptLatestAsync(_gate, (latest, sequence) =>
             {
                 ThrowIfDisposed();
-                _current = latest;
+                _current = CultMeshDocumentHandle<TDocument>.CloneDocument(latest);
+                _appliedSequence = sequence ?? _appliedSequence;
                 return _current;
-            }
+            });
         }
 
         /// <inheritdoc />
@@ -1368,10 +1394,8 @@ namespace GameCult.Mesh
         public async Task<CultMeshReactiveDocument<TDocument>> ReactiveAsync(
             CultMeshReactiveDocumentOptions? options = null)
         {
-            var current = CultMeshDocumentHandle<TDocument>.CloneDocument(
-                await Document.LatestAsync().ConfigureAwait(false));
-            var reactive = new CultMeshReactiveDocument<TDocument>(this, current, options);
-            reactive.Start();
+            var reactive = new CultMeshReactiveDocument<TDocument>(this, options);
+            await reactive.StartAsync().ConfigureAwait(false);
             return reactive;
         }
 
@@ -1465,12 +1489,11 @@ namespace GameCult.Mesh
 
         internal CultMeshReactiveDocument(
             CultMeshDocumentWriter<TDocument> writer,
-            TDocument current,
             CultMeshReactiveDocumentOptions? options)
         {
             _writer = writer ?? throw new ArgumentNullException(nameof(writer));
             _document = writer.Document;
-            _current = current ?? throw new ArgumentNullException(nameof(current));
+            _current = null!;
             _options = options ?? new CultMeshReactiveDocumentOptions();
         }
 
@@ -1509,9 +1532,18 @@ namespace GameCult.Mesh
         /// <summary>Gets the most recent reconciliation snapshot, when a canonical value arrived during local prediction.</summary>
         public CultMeshReactiveDocumentReconciliation<TDocument>? Reconciliation { get; private set; }
 
-        internal void Start()
+        internal async Task StartAsync()
         {
             _subscription = _document.WatchSequenced(ApplyCanonicalSnapshot);
+            try
+            {
+                await RefreshAsync().ConfigureAwait(false);
+            }
+            catch
+            {
+                Dispose();
+                throw;
+            }
         }
 
         /// <summary>Mutates the current value and schedules a coalesced prediction or replacement.</summary>
@@ -1543,18 +1575,17 @@ namespace GameCult.Mesh
         }
 
         /// <summary>Reads a fresh canonical snapshot and adopts it as the current value.</summary>
-        public async Task<TDocument> RefreshAsync()
+        public Task<TDocument> RefreshAsync()
         {
             ThrowIfDisposed();
-            var latest = CultMeshDocumentHandle<TDocument>.CloneDocument(
-                await _document.LatestAsync().ConfigureAwait(false));
-            lock (_gate)
+            return _document.AdoptLatestAsync(_gate, (latest, sequence) =>
             {
-                _current = latest;
+                _current = CultMeshDocumentHandle<TDocument>.CloneDocument(latest);
+                _appliedSequence = sequence ?? _appliedSequence;
                 _dirty = false;
                 Reconciliation = null;
                 return CultMeshDocumentHandle<TDocument>.CloneDocument(_current);
-            }
+            });
         }
 
         /// <summary>Immediately sends the latest local dirty value, if any, through the document authority shape.</summary>

tests/GameCult.Caching.Tests/ConditionalCommitTests.cs

@@ -813,6 +813,41 @@ namespace GameCult.Caching.Tests
             Assert.That(broken, Is.EqualTo(0), "a change's previous document is not the one installed by the next-lower Sequence");
         }
 
+        // Each read must hold the document of the highest-Sequence change at or below the read's Sequence.
+        [Test]
+        public void SequencedReadReflectsEveryChangeAtOrBelowItsSequence()
+        {
+            using var cache = new CultCache(Registry);
+            var key = new CultRecordKey("sequenced");
+            var handle = new CultRecordHandle<Counter>(key);
+            var changes = new ConcurrentBag<CultCacheDocumentChange<Counter>>();
+            using var subscription = cache.Watch<Counter>().Subscribe(changes.Add);
+            cache.UpsertAsync(new Counter { Name = "sequenced", Value = -1 }, handle);
+
+            var reads = new ConcurrentBag<(object? Document, long Sequence)>();
+            var writers = Enumerable.Range(0, 4).Select(_ => Task.Run(() =>
+            {
+                for (var i = 0; i < 2000; i++)
+                    cache.UpsertAsync(new Counter { Name = "sequenced", Value = i }, handle);
+            })).ToArray();
+            var readers = Enumerable.Range(0, 4).Select(_ => Task.Run(() =>
+            {
+                while (!writers.All(writer => writer.IsCompleted))
+                    reads.Add(cache.GetWithSequence(key));
+            })).ToArray();
+
+            Assert.That(Task.WaitAll(writers.Concat(readers).ToArray(), TimeSpan.FromSeconds(60)), Is.True);
+            var ordered = changes.OrderBy(change => change.Sequence).ToArray();
+            var sequences = ordered.Select(change => change.Sequence).ToArray();
+            Assert.That(reads, Is.Not.Empty);
+            foreach (var (document, sequence) in reads)
+            {
+                var index = Array.BinarySearch(sequences, sequence);
+                Assert.That(index, Is.GreaterThanOrEqualTo(0), "a read's Sequence names no admitted change");
+                Assert.That(document, Is.SameAs(ordered[index].Document), $"read at Sequence {sequence} missed a change at or below it");
+            }
+        }
+
         private string PathOf(string fileName) => System.IO.Path.Combine(_directory, fileName);
 
         private static readonly CultDocumentRegistry Registry = CultDocumentRegistry.ForTypes(new[] { typeof(Counter), typeof(Tally), typeof(Slow) });

tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs

@@ -1419,6 +1419,103 @@ public sealed class CultMeshStreamingTests
         cache.Dispose();
     }
 
+    // A batch parks in an observer of another key before publishing its (older) target-key change; the mirror is
+    // created after a newer write landed, then the straggler is delivered.
+    [TestCase(false)]
+    [TestCase(true)]
+    public async Task ObserveAfterNewerSnapshotIgnoresStraggler(bool reactive)
+    {
+        var cache = new CultCache();
+        var key = new CultRecordKey("mesh-note:straggler");
+        var other = new CultRecordKey("mesh-note:straggler-other");
+        await cache.UpsertAsync(Note("initial", 1), new CultRecordHandle<MeshNoteDocument>(key));
+        using var release = new ManualResetEventSlim();
+        using var parked = new ManualResetEventSlim();
+        using var park = cache.WatchRecord<MeshNoteDocument>(other).Subscribe(_ =>
+        {
+            parked.Set();
+            release.Wait(TimeSpan.FromSeconds(30));
+        });
+
+        var batch = Task.Run(() => cache.Commit(stage =>
+        {
+            stage.Upsert(Note("other", 1), new CultRecordHandle<MeshNoteDocument>(other));
+            stage.Upsert(Note("older", 2), new CultRecordHandle<MeshNoteDocument>(key));
+        }));
+        parked.Wait(TimeSpan.FromSeconds(5)).Should().BeTrue();
+        await cache.UpsertAsync(Note("newer", 3), new CultRecordHandle<MeshNoteDocument>(key));
+        var (mirror, text, _) = await Mirror(cache, key, reactive);
+        release.Set();
+
+        batch.Wait(TimeSpan.FromSeconds(10)).Should().BeTrue();
+        text().Should().Be("newer");
+        mirror.Dispose();
+        cache.Dispose();
+    }
+
+    // The mirror's own delivery of the newer write is parked too, so only the refresh saw it before the straggler.
+    [TestCase(false)]
+    [TestCase(true)]
+    public async Task RefreshIgnoresStragglerOlderThanSnapshot(bool reactive)
+    {
+        var cache = new CultCache();
+        var key = new CultRecordKey("mesh-note:refresh-straggler");
+        var other = new CultRecordKey("mesh-note:refresh-straggler-other");
+        await cache.UpsertAsync(Note("initial", 1), new CultRecordHandle<MeshNoteDocument>(key));
+        using var releaseOlder = new ManualResetEventSlim();
+        using var olderParked = new ManualResetEventSlim();
+        using var releaseNewer = new ManualResetEventSlim();
+        using var newerParked = new ManualResetEventSlim();
+        using var parkOlder = cache.WatchRecord<MeshNoteDocument>(other).Subscribe(_ =>
+        {
+            olderParked.Set();
+            releaseOlder.Wait(TimeSpan.FromSeconds(30));
+        });
+        using var parkNewer = cache.WatchRecord<MeshNoteDocument>(key).Subscribe(change =>
+        {
+            if (change.Document?.Text != "newer") return;
+            newerParked.Set();
+            releaseNewer.Wait(TimeSpan.FromSeconds(30));
+        });
+        var (mirror, text, refresh) = await Mirror(cache, key, reactive);
+
+        var batch = Task.Run(() => cache.Commit(stage =>
+        {
+            stage.Upsert(Note("other", 1), new CultRecordHandle<MeshNoteDocument>(other));
+            stage.Upsert(Note("older", 2), new CultRecordHandle<MeshNoteDocument>(key));
+        }));
+        olderParked.Wait(TimeSpan.FromSeconds(5)).Should().BeTrue();
+        var newer = Task.Run(() => cache.UpsertAsync(Note("newer", 3), new CultRecordHandle<MeshNoteDocument>(key)));
+        newerParked.Wait(TimeSpan.FromSeconds(5)).Should().BeTrue();
+        await refresh();
+        releaseOlder.Set();
+        batch.Wait(TimeSpan.FromSeconds(10)).Should().BeTrue();
+        text().Should().Be("newer");
+        releaseNewer.Set();
+
+        newer.Wait(TimeSpan.FromSeconds(10)).Should().BeTrue();
+        text().Should().Be("newer");
+        mirror.Dispose();
+        cache.Dispose();
+    }
+
+    private static MeshNoteDocument Note(string text, int revision) =>
+        new() { Schema = "tests.mesh_note.v1", Text = text, Revision = revision };
+
+    private static async Task<(IDisposable Mirror, Func<string> Text, Func<Task> Refresh)> Mirror(
+        CultCache cache, CultRecordKey key, bool reactive)
+    {
+        var handle = CultMesh.Document<MeshNoteDocument>(cache, key);
+        if (reactive)
+        {
+            var mirror = await handle.AuthoritativeWriter().ReactiveAsync();
+            return (mirror, () => mirror.Snapshot.Text, () => mirror.RefreshAsync());
+        }
+
+        var observed = await handle.ObserveAsync();
+        return (observed, () => observed.Current.Text, () => observed.RefreshAsync());
+    }
+
     [Test]
     public async Task ReactiveDocument_DisposeSuppressesScheduledFlush()
     {