diff --git a/src/GameCult.Caching/Contracts/cultcache-store-composition.md b/src/GameCult.Caching/Contracts/cultcache-store-composition.md index ca0abd61..3084a62f 100644 --- a/src/GameCult.Caching/Contracts/cultcache-store-composition.md +++ b/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` 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 diff --git a/src/GameCult.Caching/CultCache.cs b/src/GameCult.Caching/CultCache.cs index da005717..199e010f 100644 --- a/src/GameCult.Caching/CultCache.cs +++ b/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(CultRecordKey key, out T? document) where T : class { document = Get(key); diff --git a/src/GameCult.Mesh/CultMesh.cs b/src/GameCult.Mesh/CultMesh.cs index 6f89966d..e7436db2 100644 --- a/src/GameCult.Mesh/CultMesh.cs +++ b/src/GameCult.Mesh/CultMesh.cs @@ -1531,7 +1531,12 @@ namespace GameCult.Mesh return new CultMeshDocumentHandle( BindLiveFeed(context, feed), async value => await cache.UpsertAsync(value, new CultRecordHandle(key)).ConfigureAwait(false), - () => cache.WatchRecord(key)); + () => cache.WatchRecord(key), + () => + { + var (document, sequence) = cache.GetWithSequence(key); + return (RequireDocument(document, key), sequence); + }); } /// @@ -3370,13 +3375,15 @@ namespace GameCult.Mesh } private static TDocument ReadRequired(CultCache cache, CultRecordKey key) + where TDocument : class => + RequireDocument(cache.Get(key), key); + + private static TDocument RequireDocument(object? untyped, CultRecordKey key) where TDocument : class { - var document = cache.Get(key); - if (document != null) + if (untyped is TDocument document) return document; - var untyped = cache.Get(key); if (untyped != null && IsSameCultDocumentSchema(untyped.GetType())) return ConvertUntypedDocument(untyped); diff --git a/src/GameCult.Mesh/CultMeshPrimitives.cs b/src/GameCult.Mesh/CultMeshPrimitives.cs index 98ee230d..58723950 100644 --- a/src/GameCult.Mesh/CultMeshPrimitives.cs +++ b/src/GameCult.Mesh/CultMeshPrimitives.cs @@ -929,6 +929,7 @@ namespace GameCult.Mesh private readonly Func? _submitPrediction; // Set only over a CultCache record, whose changes carry the cache's Sequence. private readonly Func>>? _recordChanges; + private readonly Func<(TDocument Document, long Sequence)>? _sequencedLatest; /// Creates a document handle from a Verse-bound live feed. public CultMeshDocumentHandle( @@ -944,10 +945,12 @@ namespace GameCult.Mesh internal CultMeshDocumentHandle( CultMeshBoundLiveFeed feed, Func replace, - Func>> recordChanges) + Func>> recordChanges, + Func<(TDocument Document, long Sequence)> sequencedLatest) : this(feed, replace) { _recordChanges = recordChanges; + _sequencedLatest = sequencedLatest; } /// Gets the semantic document id. @@ -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 AdoptLatestAsync(object gate, Func 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); + } + /// /// 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 /// Creates a read-only in-memory mirror updated by the document watch. public async Task> ObserveAsync() { - var current = CloneDocument(await LatestAsync().ConfigureAwait(false)); - var observed = new CultMeshObservedDocument(this, current); - observed.Start(); + var observed = new CultMeshObservedDocument(this); + await observed.StartAsync().ConfigureAwait(false); return observed; } @@ -1224,12 +1244,10 @@ namespace GameCult.Mesh private long _appliedSequence; private bool _disposed; - internal CultMeshObservedDocument( - CultMeshDocumentHandle document, - TDocument current) + internal CultMeshObservedDocument(CultMeshDocumentHandle document) { Document = document ?? throw new ArgumentNullException(nameof(document)); - _current = current ?? throw new ArgumentNullException(nameof(current)); + _current = null!; } /// Gets the underlying read/watch handle. @@ -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; + } } /// Reads and adopts a fresh canonical snapshot. - public async Task RefreshAsync() + public Task RefreshAsync() { ThrowIfDisposed(); - var latest = CultMeshDocumentHandle.CloneDocument( - await Document.LatestAsync().ConfigureAwait(false)); - lock (_gate) + return Document.AdoptLatestAsync(_gate, (latest, sequence) => { ThrowIfDisposed(); - _current = latest; + _current = CultMeshDocumentHandle.CloneDocument(latest); + _appliedSequence = sequence ?? _appliedSequence; return _current; - } + }); } /// @@ -1368,10 +1394,8 @@ namespace GameCult.Mesh public async Task> ReactiveAsync( CultMeshReactiveDocumentOptions? options = null) { - var current = CultMeshDocumentHandle.CloneDocument( - await Document.LatestAsync().ConfigureAwait(false)); - var reactive = new CultMeshReactiveDocument(this, current, options); - reactive.Start(); + var reactive = new CultMeshReactiveDocument(this, options); + await reactive.StartAsync().ConfigureAwait(false); return reactive; } @@ -1465,12 +1489,11 @@ namespace GameCult.Mesh internal CultMeshReactiveDocument( CultMeshDocumentWriter 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 /// Gets the most recent reconciliation snapshot, when a canonical value arrived during local prediction. public CultMeshReactiveDocumentReconciliation? Reconciliation { get; private set; } - internal void Start() + internal async Task StartAsync() { _subscription = _document.WatchSequenced(ApplyCanonicalSnapshot); + try + { + await RefreshAsync().ConfigureAwait(false); + } + catch + { + Dispose(); + throw; + } } /// Mutates the current value and schedules a coalesced prediction or replacement. @@ -1543,18 +1575,17 @@ namespace GameCult.Mesh } /// Reads a fresh canonical snapshot and adopts it as the current value. - public async Task RefreshAsync() + public Task RefreshAsync() { ThrowIfDisposed(); - var latest = CultMeshDocumentHandle.CloneDocument( - await _document.LatestAsync().ConfigureAwait(false)); - lock (_gate) + return _document.AdoptLatestAsync(_gate, (latest, sequence) => { - _current = latest; + _current = CultMeshDocumentHandle.CloneDocument(latest); + _appliedSequence = sequence ?? _appliedSequence; _dirty = false; Reconciliation = null; return CultMeshDocumentHandle.CloneDocument(_current); - } + }); } /// Immediately sends the latest local dirty value, if any, through the document authority shape. diff --git a/tests/GameCult.Caching.Tests/ConditionalCommitTests.cs b/tests/GameCult.Caching.Tests/ConditionalCommitTests.cs index 5c36c713..875d7664 100644 --- a/tests/GameCult.Caching.Tests/ConditionalCommitTests.cs +++ b/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(key); + var changes = new ConcurrentBag>(); + using var subscription = cache.Watch().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) }); diff --git a/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs b/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs index 1dc938c9..da3851df 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs +++ b/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(key)); + using var release = new ManualResetEventSlim(); + using var parked = new ManualResetEventSlim(); + using var park = cache.WatchRecord(other).Subscribe(_ => + { + parked.Set(); + release.Wait(TimeSpan.FromSeconds(30)); + }); + + var batch = Task.Run(() => cache.Commit(stage => + { + stage.Upsert(Note("other", 1), new CultRecordHandle(other)); + stage.Upsert(Note("older", 2), new CultRecordHandle(key)); + })); + parked.Wait(TimeSpan.FromSeconds(5)).Should().BeTrue(); + await cache.UpsertAsync(Note("newer", 3), new CultRecordHandle(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(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(other).Subscribe(_ => + { + olderParked.Set(); + releaseOlder.Wait(TimeSpan.FromSeconds(30)); + }); + using var parkNewer = cache.WatchRecord(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(other)); + stage.Upsert(Note("older", 2), new CultRecordHandle(key)); + })); + olderParked.Wait(TimeSpan.FromSeconds(5)).Should().BeTrue(); + var newer = Task.Run(() => cache.UpsertAsync(Note("newer", 3), new CultRecordHandle(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 Text, Func Refresh)> Mirror( + CultCache cache, CultRecordKey key, bool reactive) + { + var handle = CultMesh.Document(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() {