diff --git a/src/GameCult.Mesh/CultMesh.cs b/src/GameCult.Mesh/CultMesh.cs index fbb08f22..232f847e 100644 --- a/src/GameCult.Mesh/CultMesh.cs +++ b/src/GameCult.Mesh/CultMesh.cs @@ -493,56 +493,6 @@ namespace GameCult.Mesh return Document(recordKey, verse.Context, documentId, sources, routeHint); } - /// - /// Creates a managed reactive document mirror over one distributed record hosted by this node. - /// - public Task> ReactiveDocumentAsync( - CultRecordKey key, - CultMeshVerseContext context, - CultMeshReactiveDocumentOptions? options = null) - where TDocument : class - { - return Document(key, context).ReactiveAsync(options); - } - - /// - /// Creates a managed reactive document mirror over one distributed record hosted by this node. - /// - public Task> ReactiveDocumentAsync( - CultRecordKey key, - CultMeshVerse verse, - CultMeshReactiveDocumentOptions? options = null) - where TDocument : class - { - if (verse == null) throw new ArgumentNullException(nameof(verse)); - return ReactiveDocumentAsync(key, verse.Context, options); - } - - /// - /// Creates a managed reactive document mirror over one distributed record hosted by this node. - /// - public Task> ReactiveDocumentAsync( - string recordKey, - CultMeshVerseContext context, - CultMeshReactiveDocumentOptions? options = null) - where TDocument : class - { - return ReactiveDocumentAsync(new CultRecordKey(recordKey), context, options); - } - - /// - /// Creates a managed reactive document mirror over one distributed record hosted by this node. - /// - public Task> ReactiveDocumentAsync( - string recordKey, - CultMeshVerse verse, - CultMeshReactiveDocumentOptions? options = null) - where TDocument : class - { - if (verse == null) throw new ArgumentNullException(nameof(verse)); - return ReactiveDocumentAsync(recordKey, verse.Context, options); - } - /// public void Dispose() { @@ -2010,21 +1960,22 @@ namespace GameCult.Mesh var sourceList = sources?.ToArray() ?? new[] { ProjectionSource(key.Value, descriptor.SchemaId, "CultNet database record") }; var route = routeHint ?? new CultMeshRouteHint(CultMeshLocalityKind.Automatic, "CultNet database document"); - return Document( + var feed = LiveFeed( ResolveDocumentId(documentId, key), - context, - async _ => await ReadDatabaseDocumentRequiredAsync(database, key).ConfigureAwait(false), - _ => WatchDatabaseRecordAs(database, key), - async value => - { - await PutDatabaseDocumentAsync(database, key, value, predicted: false).ConfigureAwait(false); - }, - async value => - { - await PutDatabaseDocumentAsync(database, key, value, predicted: true).ConfigureAwait(false); - }, + (_parameters, _context) => ReadDatabaseDocumentRequiredAsync(database, key), + (_parameters, _context) => WatchDatabaseRecordAs(database, key), sourceList, route); + Func? replace = database.CanWriteAuthoritatively(key) + ? value => PutDatabaseDocumentAsync(database, key, value, predicted: false) + : null; + Func? submitPrediction = database.CanSubmitPrediction(key) + ? value => PutDatabaseDocumentAsync(database, key, value, predicted: true) + : null; + return new CultMeshDocumentHandle( + BindLiveFeed(context, feed), + replace, + submitPrediction); } /// diff --git a/src/GameCult.Mesh/CultMeshPrimitives.cs b/src/GameCult.Mesh/CultMeshPrimitives.cs index dcc58143..00ba60b7 100644 --- a/src/GameCult.Mesh/CultMeshPrimitives.cs +++ b/src/GameCult.Mesh/CultMeshPrimitives.cs @@ -911,9 +911,6 @@ namespace GameCult.Mesh /// Gets whether this handle can submit a client prediction for the underlying document value. bool CanSubmitPrediction { get; } - /// Gets whether this handle can accept a transparent document mutation. - bool CanSet { get; } - /// Creates a same-schema alias presentation for another CLR document type. CultMeshDocumentHandle AsSchemaAlias() where TAlias : class; } @@ -975,9 +972,6 @@ namespace GameCult.Mesh /// Gets whether this handle can submit a client prediction for the underlying document value. public bool CanSubmitPrediction => _submitPrediction != null; - /// Gets whether this handle can accept a transparent document mutation. - public bool CanSet => CanSubmitPrediction || CanReplace; - /// Reads one coherent document snapshot. public Task LatestAsync() { @@ -1048,57 +1042,41 @@ namespace GameCult.Mesh return _submitPrediction(value); } - /// - /// Sets the document value through the configured authority shape. - /// Prediction-backed documents publish a prediction; mutable documents replace authoritatively. - /// - public Task SetAsync(TDocument value) + /// Gets an explicit authoritative replacement writer for this document. + public CultMeshDocumentWriter AuthoritativeWriter() { - if (value == null) throw new ArgumentNullException(nameof(value)); - if (_submitPrediction != null) - return _submitPrediction(value); - if (_replace != null) - return _replace(value); - - throw new NotSupportedException($"Document handle '{DocumentId}' does not accept mutations."); + if (_replace == null) + throw new NotSupportedException($"Document handle '{DocumentId}' is read-only."); + return new CultMeshDocumentWriter( + this, + CultMeshDocumentWriteKind.AuthoritativeReplacement, + _replace); } - /// Reads, updates, and sets the document value through the configured authority shape. - public async Task UpdateAsync(Func update) + /// Gets an explicit client prediction writer for this document. + public CultMeshDocumentWriter PredictionWriter() { - if (update == null) throw new ArgumentNullException(nameof(update)); - var next = update(await LatestAsync().ConfigureAwait(false)); - await SetAsync(next).ConfigureAwait(false); - return next; + if (_submitPrediction == null) + throw new NotSupportedException($"Document handle '{DocumentId}' does not accept client predictions."); + return new CultMeshDocumentWriter( + this, + CultMeshDocumentWriteKind.Prediction, + _submitPrediction); } - /// Reads, updates, and sets the document value through the configured authority shape. - public async Task UpdateAsync(Func> update) - { - if (update == null) throw new ArgumentNullException(nameof(update)); - var next = await update(await LatestAsync().ConfigureAwait(false)).ConfigureAwait(false); - await SetAsync(next).ConfigureAwait(false); - return next; - } - - /// - /// Creates a managed reactive document mirror whose local edits can be coalesced into one prediction or replacement. - /// - public async Task> ReactiveAsync( - CultMeshReactiveDocumentOptions? options = null) + /// Creates a read-only in-memory mirror updated by the document watch. + public async Task> ObserveAsync() { var current = CloneDocument(await LatestAsync().ConfigureAwait(false)); - var reactive = new CultMeshReactiveDocument(this, current, options); - reactive.Start(); - return reactive; + var observed = new CultMeshObservedDocument(this, current); + observed.Start(); + return observed; } - /// - /// Creates a managed reactive document mirror synchronously for host APIs that cannot be async. - /// - public CultMeshReactiveDocument Reactive(CultMeshReactiveDocumentOptions? options = null) + /// Creates a read-only in-memory mirror synchronously. + public CultMeshObservedDocument Observe() { - return ReactiveAsync(options).ConfigureAwait(false).GetAwaiter().GetResult(); + return ObserveAsync().ConfigureAwait(false).GetAwaiter().GetResult(); } /// Creates a same-schema alias presentation for another CLR document type. @@ -1215,6 +1193,174 @@ namespace GameCult.Mesh } } + /// + /// A read-only local mirror of a watched typed document. Observation never acquires a write + /// capability and never polls or serializes the document while idle. + /// + public sealed class CultMeshObservedDocument : IDisposable + where TDocument : class + { + private readonly object _gate = new(); + private IDisposable? _subscription; + private TDocument _current; + private bool _disposed; + + internal CultMeshObservedDocument( + CultMeshDocumentHandle document, + TDocument current) + { + Document = document ?? throw new ArgumentNullException(nameof(document)); + _current = current ?? throw new ArgumentNullException(nameof(current)); + } + + /// Gets the underlying read/watch handle. + public CultMeshDocumentHandle Document { get; } + + /// + /// Gets the current borrowed snapshot. Treat the object graph as read-only; local mutation + /// cannot publish and is replaced by the next canonical snapshot. + /// + public TDocument Current + { + get + { + lock (_gate) + { + ThrowIfDisposed(); + return _current; + } + } + } + + internal void Start() + { + _subscription = Document.Watch(ApplyCanonicalSnapshot); + } + + /// Reads and adopts a fresh canonical snapshot. + public async Task RefreshAsync() + { + ThrowIfDisposed(); + var latest = CultMeshDocumentHandle.CloneDocument( + await Document.LatestAsync().ConfigureAwait(false)); + lock (_gate) + { + ThrowIfDisposed(); + _current = latest; + return _current; + } + } + + /// + public void Dispose() + { + lock (_gate) + { + if (_disposed) + return; + _disposed = true; + } + + _subscription?.Dispose(); + } + + private void ApplyCanonicalSnapshot(TDocument canonical) + { + if (canonical == null) + return; + var next = CultMeshDocumentHandle.CloneDocument(canonical); + lock (_gate) + { + if (!_disposed) + _current = next; + } + } + + private void ThrowIfDisposed() + { + if (_disposed) + throw new ObjectDisposedException(nameof(CultMeshObservedDocument)); + } + } + + /// Names the one immutable authority path selected by a document writer. + public enum CultMeshDocumentWriteKind + { + /// The writer replaces canonical state through its owning provider. + AuthoritativeReplacement, + + /// The writer submits a client prediction or candidate for validation. + Prediction + } + + /// + /// An explicit, immutable write capability for one typed document. A writer never switches + /// between authoritative replacement and prediction based on runtime availability. + /// + public sealed class CultMeshDocumentWriter + where TDocument : class + { + private readonly Func _write; + + internal CultMeshDocumentWriter( + CultMeshDocumentHandle document, + CultMeshDocumentWriteKind kind, + Func write) + { + Document = document ?? throw new ArgumentNullException(nameof(document)); + Kind = kind; + _write = write ?? throw new ArgumentNullException(nameof(write)); + } + + /// Gets the document read/watch handle paired with this writer. + public CultMeshDocumentHandle Document { get; } + + /// Gets the immutable authority path selected for this writer. + public CultMeshDocumentWriteKind Kind { get; } + + /// Writes one value through this writer's named authority path. + public Task WriteAsync(TDocument value) + { + if (value == null) throw new ArgumentNullException(nameof(value)); + return _write(value); + } + + /// Reads, transforms, and writes through this writer's named authority path. + public async Task UpdateAsync(Func update) + { + if (update == null) throw new ArgumentNullException(nameof(update)); + var next = update(await Document.LatestAsync().ConfigureAwait(false)); + await WriteAsync(next).ConfigureAwait(false); + return next; + } + + /// Reads, transforms, and writes through this writer's named authority path. + public async Task UpdateAsync(Func> update) + { + if (update == null) throw new ArgumentNullException(nameof(update)); + var next = await update(await Document.LatestAsync().ConfigureAwait(false)).ConfigureAwait(false); + await WriteAsync(next).ConfigureAwait(false); + return next; + } + + /// Creates an explicitly writable reactive mirror. + 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(); + return reactive; + } + + /// Creates an explicitly writable reactive mirror synchronously. + public CultMeshReactiveDocument Reactive(CultMeshReactiveDocumentOptions? options = null) + { + return ReactiveAsync(options).ConfigureAwait(false).GetAwaiter().GetResult(); + } + } + /// /// Configures a managed CultMesh reactive document mirror. /// @@ -1225,9 +1371,6 @@ namespace GameCult.Mesh /// public TimeSpan FlushDelay { get; set; } = TimeSpan.FromMilliseconds(16); - /// Gets or sets whether direct edits to Current should be detected and flushed automatically. - public bool DetectLocalChanges { get; set; } = true; - /// Gets or sets whether canonical snapshots replace dirty local predictions immediately. public bool ReplaceDirtyCurrentOnCanonicalSnapshot { get; set; } } @@ -1272,18 +1415,18 @@ namespace GameCult.Mesh } /// - /// Managed typed document mirror that keeps an editable current value and coalesces local edits into CultMesh mutations. + /// Managed typed document mirror that coalesces explicit local edits through one selected writer. /// public sealed class CultMeshReactiveDocument : IDisposable where TDocument : class { + private readonly CultMeshDocumentWriter _writer; private readonly CultMeshDocumentHandle _document; private readonly CultMeshReactiveDocumentOptions _options; private readonly object _gate = new(); private IDisposable? _subscription; private Timer? _flushTimer; - private Timer? _changeDetectionTimer; - private byte[] _lastCleanSnapshot; + private TDocument _current; private bool _dirty; private bool _flushQueued; private bool _flushing; @@ -1291,21 +1434,37 @@ namespace GameCult.Mesh private int _reconciliationVersion; internal CultMeshReactiveDocument( - CultMeshDocumentHandle document, + CultMeshDocumentWriter writer, TDocument current, CultMeshReactiveDocumentOptions? options) { - _document = document ?? throw new ArgumentNullException(nameof(document)); - Current = current ?? throw new ArgumentNullException(nameof(current)); + _writer = writer ?? throw new ArgumentNullException(nameof(writer)); + _document = writer.Document; + _current = current ?? throw new ArgumentNullException(nameof(current)); _options = options ?? new CultMeshReactiveDocumentOptions(); - _lastCleanSnapshot = SerializeDocument(Current); } + /// Gets the immutable write capability selected for this mirror. + public CultMeshDocumentWriter Writer => _writer; + /// Gets the underlying document handle. public CultMeshDocumentHandle Document => _document; - /// Gets the editable local document value. - public TDocument Current { get; private set; } + /// + /// Gets a cloned local snapshot. Mutating the returned object cannot mutate or publish the mirror; + /// use or for edits. + /// + public TDocument Snapshot + { + get + { + lock (_gate) + { + ThrowIfDisposed(); + return CultMeshDocumentHandle.CloneDocument(_current); + } + } + } /// Gets whether local edits are waiting to be sent. public bool IsDirty @@ -1323,25 +1482,6 @@ namespace GameCult.Mesh internal void Start() { _subscription = _document.Watch(ApplyCanonicalSnapshot); - if (_options.DetectLocalChanges && _document.CanSet) - { - var interval = _options.FlushDelay <= TimeSpan.Zero - ? TimeSpan.FromMilliseconds(1) - : _options.FlushDelay; - _changeDetectionTimer = new Timer(_ => _ = Task.Run(DetectLocalChangesAsync)); - _changeDetectionTimer.Change(interval, interval); - } - } - - /// Marks the current value dirty and schedules a coalesced prediction or replacement. - public void MarkDirty() - { - ThrowIfDisposed(); - lock (_gate) - { - _dirty = true; - ScheduleFlushLocked(); - } } /// Mutates the current value and schedules a coalesced prediction or replacement. @@ -1351,24 +1491,24 @@ namespace GameCult.Mesh ThrowIfDisposed(); lock (_gate) { - update(Current); + update(_current); _dirty = true; ScheduleFlushLocked(); - return Current; + return CultMeshDocumentHandle.CloneDocument(_current); } } /// Replaces the current local value and schedules a coalesced prediction or replacement. - public TDocument SetCurrent(TDocument value) + public TDocument ReplaceLocal(TDocument value) { if (value == null) throw new ArgumentNullException(nameof(value)); ThrowIfDisposed(); lock (_gate) { - Current = CultMeshDocumentHandle.CloneDocument(value); + _current = CultMeshDocumentHandle.CloneDocument(value); _dirty = true; ScheduleFlushLocked(); - return Current; + return CultMeshDocumentHandle.CloneDocument(_current); } } @@ -1380,11 +1520,10 @@ namespace GameCult.Mesh await _document.LatestAsync().ConfigureAwait(false)); lock (_gate) { - Current = latest; - _lastCleanSnapshot = SerializeDocument(Current); + _current = latest; _dirty = false; Reconciliation = null; - return Current; + return CultMeshDocumentHandle.CloneDocument(_current); } } @@ -1395,8 +1534,6 @@ namespace GameCult.Mesh lock (_gate) { ThrowIfDisposed(); - if (!_dirty && _document.CanSet) - DetectLocalChangesLocked(); if (!_dirty) return; if (_flushing) @@ -1408,13 +1545,12 @@ namespace GameCult.Mesh _flushTimer?.Change(Timeout.InfiniteTimeSpan, Timeout.InfiniteTimeSpan); _flushing = true; _dirty = false; - predicted = CultMeshDocumentHandle.CloneDocument(Current); - _lastCleanSnapshot = SerializeDocument(predicted); + predicted = CultMeshDocumentHandle.CloneDocument(_current); } try { - await _document.SetAsync(predicted).ConfigureAwait(false); + await _writer.WriteAsync(predicted).ConfigureAwait(false); } finally { @@ -1422,8 +1558,6 @@ namespace GameCult.Mesh lock (_gate) { _flushing = false; - if (_document.CanSet) - DetectLocalChangesLocked(); shouldFlushAgain = _flushQueued || _dirty; _flushQueued = false; } @@ -1452,7 +1586,6 @@ namespace GameCult.Mesh _subscription?.Dispose(); _flushTimer?.Dispose(); - _changeDetectionTimer?.Dispose(); } private void ApplyCanonicalSnapshot(TDocument canonical) @@ -1468,7 +1601,7 @@ namespace GameCult.Mesh var nextCanonical = CultMeshDocumentHandle.CloneDocument(canonical); if (_dirty || _flushing) { - var predicted = CultMeshDocumentHandle.CloneDocument(Current); + var predicted = CultMeshDocumentHandle.CloneDocument(_current); var delta = CreateReconciliationDelta(predicted, nextCanonical); if (delta.Count == 0) { @@ -1488,34 +1621,11 @@ namespace GameCult.Mesh return; } - Current = nextCanonical; - _lastCleanSnapshot = SerializeDocument(Current); + _current = nextCanonical; Reconciliation = null; } } - private Task DetectLocalChangesAsync() - { - lock (_gate) - { - if (_disposed || !_document.CanSet || _dirty || _flushing) - return Task.CompletedTask; - DetectLocalChangesLocked(); - if (_dirty) - ScheduleFlushLocked(); - } - - return Task.CompletedTask; - } - - private void DetectLocalChangesLocked() - { - var currentSnapshot = SerializeDocument(Current); - if (_lastCleanSnapshot.SequenceEqual(currentSnapshot)) - return; - _dirty = true; - } - private void ScheduleFlushLocked() { if (_options.FlushDelay <= TimeSpan.Zero) @@ -1528,11 +1638,6 @@ namespace GameCult.Mesh _flushTimer.Change(_options.FlushDelay, Timeout.InfiniteTimeSpan); } - private static byte[] SerializeDocument(TDocument document) - { - return CultDocumentMessagePackSerialization.SerializeUntyped(document, typeof(TDocument)); - } - private static IReadOnlyDictionary CreateReconciliationDelta( TDocument predicted, TDocument canonical) @@ -1778,13 +1883,6 @@ namespace GameCult.Mesh return TryGetDocument(out var document) && document.CanSubmitPrediction; } - /// Gets whether one typed document can accept transparent mutations by CLR type or same-schema alias. - public bool CanSet() - where TDocument : class - { - return TryGetDocument(out var document) && document.CanSet; - } - /// Submits a client prediction for one typed document by CLR type or same-schema alias. public Task SubmitPredictionAsync(TDocument value) where TDocument : class @@ -1792,25 +1890,18 @@ namespace GameCult.Mesh return Document().SubmitPredictionAsync(value); } - /// Sets one typed document through its configured authority shape by CLR type or same-schema alias. - public Task SetAsync(TDocument value) + /// Gets an explicit authoritative writer by CLR type or same-schema alias. + public CultMeshDocumentWriter AuthoritativeWriter() where TDocument : class { - return Document().SetAsync(value); + return Document().AuthoritativeWriter(); } - /// Reads, updates, and sets one typed document through its configured authority shape. - public Task UpdateAsync(Func update) + /// Gets an explicit prediction writer by CLR type or same-schema alias. + public CultMeshDocumentWriter PredictionWriter() where TDocument : class { - return Document().UpdateAsync(update); - } - - /// Reads, updates, and sets one typed document through its configured authority shape. - public Task UpdateAsync(Func> update) - where TDocument : class - { - return Document().UpdateAsync(update); + return Document().PredictionWriter(); } /// Watches one typed document by CLR type or same-schema alias. @@ -1827,21 +1918,6 @@ namespace GameCult.Mesh return Document().Watch(onNext); } - /// Creates a managed reactive document mirror by CLR type or same-schema alias. - public Task> ReactiveAsync( - CultMeshReactiveDocumentOptions? options = null) - where TDocument : class - { - return Document().ReactiveAsync(options); - } - - /// Creates a managed reactive document mirror synchronously by CLR type or same-schema alias. - public CultMeshReactiveDocument Reactive( - CultMeshReactiveDocumentOptions? options = null) - where TDocument : class - { - return Document().Reactive(options); - } } /// diff --git a/src/GameCult.Mesh/docs/document-write-authority.md b/src/GameCult.Mesh/docs/document-write-authority.md new file mode 100644 index 00000000..19cc4bb3 --- /dev/null +++ b/src/GameCult.Mesh/docs/document-write-authority.md @@ -0,0 +1,72 @@ +# CultMesh Document Write Authority + +CultMesh document reads, authoritative replacements, and client predictions are +different capabilities. A generic setter cannot choose between them without +hiding an authority decision from the caller. + +## Authority map + +Owner: + +- `CultMeshDocumentHandle` owns typed snapshot and watch access; +- an authoritative writer owns canonical replacement capability; +- a prediction writer owns candidate/prediction submission capability; +- `CultMeshReactiveDocument` owns only a local editable mirror and its + coalescing schedule. + +Inputs: + +- a bound typed live feed; +- an optional authoritative replacement delegate; +- an optional prediction delegate; +- explicit local mutations made through the reactive document API. + +Outputs: + +- coherent typed snapshots and watches; +- authoritative replacements only through an explicitly requested + authoritative writer; +- predictions only through an explicitly requested prediction writer; +- one coalesced write for each explicit burst of local mutations. + +Derived state: + +- the reactive snapshot is a local mirror, not canonical truth; +- reconciliation metadata compares a pending local write with an incoming + canonical snapshot; +- debounce timers schedule already-declared work and never inspect documents + for undeclared mutations. + +Forbidden writers: + +- `SetAsync` may not select prediction or replacement based on whichever + delegate happens to exist; +- reading or watching a document may not start serialization polling; +- mutating an exposed object graph may not silently become a network write; +- alias conversion may not erase which writer was selected. + +Shared paths: + +- direct edits, nested edits, automation, and UI bindings use `Update` or + `ReplaceLocal` on the same reactive mirror; +- explicit flush and debounced flush use the same selected writer; +- canonical watch updates and refresh use the same snapshot adoption path. + +Cut line: + +- remove `CanSet`, `SetAsync`, and ambiguous `UpdateAsync`/`ReactiveAsync` + entry points from document handles and catalogs; +- obtain either an authoritative writer or a prediction writer before editing; +- remove change-detection timers and full-document idle serialization; +- expose cloned snapshots so callers cannot mutate the live mirror behind the + mutation boundary. + +## Required proof + +- a handle carrying both capabilities cannot write until the caller selects a + writer; +- authoritative and prediction writers reach only their named delegates; +- nested edits made through `Update` are coalesced and published; +- direct mutation of a returned snapshot cannot publish; +- an idle reactive document creates no periodic detection work; +- canonical reconciliation and in-flight edits remain deterministic. diff --git a/src/GameCult.Networking/CultNetDatabase.cs b/src/GameCult.Networking/CultNetDatabase.cs index 6e1ed41f..8000e45b 100644 --- a/src/GameCult.Networking/CultNetDatabase.cs +++ b/src/GameCult.Networking/CultNetDatabase.cs @@ -503,6 +503,22 @@ namespace GameCult.Networking /// public IReadOnlyList ClientAuthorityScopes => _clientAuthorityScopes; + /// Gets whether this database instance can authoritatively replace the document at a key. + public bool CanWriteAuthoritatively(CultRecordKey key) where T : class + { + ThrowIfDisposed(); + var descriptor = _cache.Registry.GetRequired(); + return ResolveShardInternal(descriptor, key).IsPrimary; + } + + /// Gets whether this database instance can submit a prediction for the document at a key. + public bool CanSubmitPrediction(CultRecordKey key) where T : class + { + ThrowIfDisposed(); + var descriptor = _cache.Registry.GetRequired(); + return _clientAuthorityScopes.Any(scope => scope.Matches(_runtimeId, descriptor, key)); + } + /// /// Gets mutation log entries for a shard after the supplied sequence. /// diff --git a/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs b/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs index 05956d0d..6fafbf16 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs +++ b/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs @@ -940,16 +940,17 @@ public sealed class CultMeshStreamingTests CultMesh.Verse("starbridge", "pilot-a")); var alias = handle.AsSchemaAlias(); - handle.CanReplace.Should().BeTrue(); + handle.CanReplace.Should().BeFalse(); + alias.CanReplace.Should().BeFalse(); alias.CanSubmitPrediction.Should().BeTrue(); - alias.CanSet.Should().BeTrue(); var catalog = CultMesh.Documents(handle); catalog.CanSubmitPrediction().Should().BeTrue(); - catalog.CanSet().Should().BeTrue(); + var predictionWriter = catalog.PredictionWriter(); + predictionWriter.Kind.Should().Be(CultMeshDocumentWriteKind.Prediction); MeshNoteDocument observed = null!; using var subscription = handle.Watch(value => observed = value); - await catalog.SetAsync(new MeshNoteAliasDocument + await predictionWriter.WriteAsync(new MeshNoteAliasDocument { Schema = "tests.mesh_note.v1", Text = "predicted-thermal", @@ -961,7 +962,7 @@ public sealed class CultMeshStreamingTests observed.Revision.Should().Be(5); (await handle.LatestAsync()).Text.Should().Be("predicted-thermal"); - var updated = await alias.UpdateAsync(value => new MeshNoteAliasDocument + var updated = await alias.PredictionWriter().UpdateAsync(value => new MeshNoteAliasDocument { Schema = value.Schema, Text = "updated-as-prediction", @@ -1002,13 +1003,13 @@ public sealed class CultMeshStreamingTests }); var catalog = CultMesh.Documents(handle); - using var reactive = await catalog.ReactiveAsync( + using var reactive = await catalog.PredictionWriter().ReactiveAsync( new CultMeshReactiveDocumentOptions { FlushDelay = TimeSpan.FromMinutes(1) }); - reactive.Current.Text.Should().Be("initial"); + reactive.Snapshot.Text.Should().Be("initial"); reactive.Update(document => { document.Text = "first-local-edit"; @@ -1052,7 +1053,7 @@ public sealed class CultMeshStreamingTests return Task.CompletedTask; }); - using var reactive = await handle.ReactiveAsync( + using var reactive = await handle.AuthoritativeWriter().ReactiveAsync( new CultMeshReactiveDocumentOptions { FlushDelay = TimeSpan.FromMinutes(1) @@ -1071,7 +1072,7 @@ public sealed class CultMeshStreamingTests Revision = 7 }); - reactive.Current.Text.Should().Be("local-prediction"); + reactive.Snapshot.Text.Should().Be("local-prediction"); reactive.Reconciliation.Should().NotBeNull(); reactive.Reconciliation!.Version.Should().Be(1); reactive.Reconciliation!.Canonical.Text.Should().Be("canonical-correction"); @@ -1112,7 +1113,7 @@ public sealed class CultMeshStreamingTests return Task.CompletedTask; }); - using var reactive = await handle.ReactiveAsync( + using var reactive = await handle.PredictionWriter().ReactiveAsync( new CultMeshReactiveDocumentOptions { FlushDelay = TimeSpan.FromMilliseconds(25) @@ -1137,7 +1138,7 @@ public sealed class CultMeshStreamingTests } [Test] - public async Task ReactiveDocument_DetectsAndCoalescesDirectCurrentMemberWrites() + public async Task ReactiveDocument_CoalescesExplicitMemberUpdates() { var subject = new Subject(); var current = new MeshNoteDocument @@ -1166,26 +1167,29 @@ public sealed class CultMeshStreamingTests return Task.CompletedTask; }); - using var reactive = await handle.ReactiveAsync( + using var reactive = await handle.PredictionWriter().ReactiveAsync( new CultMeshReactiveDocumentOptions { FlushDelay = TimeSpan.FromMilliseconds(25) }); - reactive.Current.Text = "direct-member-one"; - reactive.Current.Text = "direct-member-two"; - reactive.Current.Revision = 2; + reactive.Update(document => document.Text = "explicit-member-one"); + reactive.Update(document => + { + document.Text = "explicit-member-two"; + document.Revision = 2; + }); await WaitForAsync(() => predictions.Count == 1); - predictions[0].Text.Should().Be("direct-member-two"); + predictions[0].Text.Should().Be("explicit-member-two"); predictions[0].Revision.Should().Be(2); - current.Text.Should().Be("direct-member-two"); + current.Text.Should().Be("explicit-member-two"); reactive.IsDirty.Should().BeFalse(); } [Test] - public async Task ReactiveDocument_QueuesDirectCurrentEditsWhileFlushIsInFlight() + public async Task ReactiveDocument_QueuesExplicitEditsWhileFlushIsInFlight() { var subject = new Subject(); var current = new MeshNoteDocument @@ -1221,7 +1225,7 @@ public sealed class CultMeshStreamingTests subject.OnNext(value); }); - using var reactive = await handle.ReactiveAsync( + using var reactive = await handle.PredictionWriter().ReactiveAsync( new CultMeshReactiveDocumentOptions { FlushDelay = TimeSpan.FromMinutes(1) @@ -1235,8 +1239,11 @@ public sealed class CultMeshStreamingTests var flushTask = reactive.FlushAsync(); await firstPredictionStarted.Task; - reactive.Current.Text = "second-after-flight-started"; - reactive.Current.Revision = 3; + reactive.Update(document => + { + document.Text = "second-after-flight-started"; + document.Revision = 3; + }); releaseFirstPrediction.SetResult(); await flushTask; @@ -1274,7 +1281,7 @@ public sealed class CultMeshStreamingTests return Task.CompletedTask; }); - using var reactive = await handle.ReactiveAsync( + using var reactive = await handle.AuthoritativeWriter().ReactiveAsync( new CultMeshReactiveDocumentOptions { FlushDelay = TimeSpan.FromMinutes(1) @@ -1293,7 +1300,7 @@ public sealed class CultMeshStreamingTests } [Test] - public async Task ReactiveDocument_ReadOnlyHandleRejectsFlush() + public void ReactiveDocument_ReadOnlyHandleRejectsWriterSelection() { var current = new MeshNoteDocument { @@ -1307,16 +1314,45 @@ public sealed class CultMeshStreamingTests _ => Task.FromResult(current), _ => new Subject()); - using var reactive = await handle.ReactiveAsync( - new CultMeshReactiveDocumentOptions - { - FlushDelay = TimeSpan.FromMinutes(1) - }); - reactive.Update(document => document.Text = "not-allowed"); + Action authoritative = () => handle.AuthoritativeWriter(); + Action prediction = () => handle.PredictionWriter(); - Func act = () => reactive.FlushAsync(); + authoritative.Should().Throw(); + prediction.Should().Throw(); + } - await act.Should().ThrowAsync(); + [Test] + public async Task ObservedDocument_WatchesWithoutWriteCapabilityOrIdleTimer() + { + var subject = new Subject(); + var current = new MeshNoteDocument + { + Schema = "tests.mesh_note.v1", + Text = "initial", + Revision = 1 + }; + var handle = CultMesh.Document( + "mesh.note.observed.readonly", + CultMesh.Verse("starbridge", "observer"), + _ => Task.FromResult(current), + _ => subject); + + using var observed = await handle.ObserveAsync(); + observed.Current.Text.Should().Be("initial"); + typeof(CultMeshObservedDocument) + .GetFields(System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.NonPublic) + .Should() + .NotContain(field => field.FieldType == typeof(System.Threading.Timer)); + + subject.OnNext(new MeshNoteDocument + { + Schema = "tests.mesh_note.v1", + Text = "canonical-update", + Revision = 2 + }); + + observed.Current.Text.Should().Be("canonical-update"); + observed.Current.Revision.Should().Be(2); } [Test] @@ -1349,7 +1385,7 @@ public sealed class CultMeshStreamingTests return Task.CompletedTask; }); - var reactive = await handle.ReactiveAsync( + var reactive = await handle.PredictionWriter().ReactiveAsync( new CultMeshReactiveDocumentOptions { FlushDelay = TimeSpan.FromMilliseconds(25) @@ -1384,7 +1420,7 @@ public sealed class CultMeshStreamingTests return Task.CompletedTask; }); - using var reactive = await handle.ReactiveAsync( + using var reactive = await handle.AuthoritativeWriter().ReactiveAsync( new CultMeshReactiveDocumentOptions { FlushDelay = TimeSpan.FromMinutes(1), @@ -1399,8 +1435,8 @@ public sealed class CultMeshStreamingTests Revision = 9 }); - reactive.Current.Text.Should().Be("canonical-wins"); - reactive.Current.Revision.Should().Be(9); + reactive.Snapshot.Text.Should().Be("canonical-wins"); + reactive.Snapshot.Revision.Should().Be(9); reactive.Reconciliation.Should().BeNull(); } @@ -1426,7 +1462,7 @@ public sealed class CultMeshStreamingTests return Task.CompletedTask; }); - using var reactive = await handle.ReactiveAsync( + using var reactive = await handle.AuthoritativeWriter().ReactiveAsync( new CultMeshReactiveDocumentOptions { FlushDelay = TimeSpan.FromMinutes(1) @@ -1442,7 +1478,7 @@ public sealed class CultMeshStreamingTests await reactive.RefreshAsync(); await reactive.FlushAsync(); - reactive.Current.Text.Should().Be("fresh-canonical"); + reactive.Snapshot.Text.Should().Be("fresh-canonical"); reactive.IsDirty.Should().BeFalse(); replacements.Should().BeEmpty(); } @@ -1495,7 +1531,8 @@ public sealed class CultMeshStreamingTests .Watch(observedByCommander.Add); using var pilotReactive = await CultMesh .Documents(pilotHandle) - .ReactiveAsync( + .PredictionWriter() + .ReactiveAsync( new CultMeshReactiveDocumentOptions { FlushDelay = TimeSpan.FromMinutes(1) @@ -1557,14 +1594,13 @@ public sealed class CultMeshStreamingTests database, key, CultMesh.Verse("starbridge", "commander-rts")); - using var commanderReactive = await commanderHandle + MeshNoteAliasDocument commanderSnapshot = null!; + using var commanderSubscription = commanderHandle .AsSchemaAlias() - .ReactiveAsync(new CultMeshReactiveDocumentOptions - { - FlushDelay = TimeSpan.FromMinutes(1) - }); + .Watch(value => commanderSnapshot = value); using var pilotReactive = await pilotHandle .AsSchemaAlias() + .PredictionWriter() .ReactiveAsync(new CultMeshReactiveDocumentOptions { FlushDelay = TimeSpan.FromMinutes(1) @@ -1582,16 +1618,16 @@ public sealed class CultMeshStreamingTests }); await pilotReactive.FlushAsync(); - await WaitForAsync(() => commanderReactive.Current.Text == "pilot-coalesced-order"); + await WaitForAsync(() => commanderSnapshot?.Text == "pilot-coalesced-order"); pilotReactive.IsDirty.Should().BeFalse(); - commanderReactive.Document.Context.RuntimeId.Should().Be("commander-rts"); - commanderReactive.Current.Revision.Should().Be(3); + commanderHandle.Context.RuntimeId.Should().Be("commander-rts"); + commanderSnapshot.Revision.Should().Be(3); cache.Get(key)!.Text.Should().Be("pilot-coalesced-order"); } [Test] - public async Task NodeReactiveDocument_OpensSameSchemaAliasWithOneCall() + public async Task NodeDocument_OpensExplicitPredictionWriterForSameSchemaAlias() { var schemaId = CultDocumentRegistry.Shared.GetRequired().SchemaId; using var temp = new TemporaryDirectory(); @@ -1630,10 +1666,10 @@ public sealed class CultMeshStreamingTests Revision = 1 }, new CultRecordHandle(key)); - using var pilotReactive = await node.ReactiveDocumentAsync( - key, - CultMesh.Verse("starbridge", "pilot-a"), - new CultMeshReactiveDocumentOptions + using var pilotReactive = await node + .Document(key, CultMesh.Verse("starbridge", "pilot-a")) + .PredictionWriter() + .ReactiveAsync(new CultMeshReactiveDocumentOptions { FlushDelay = TimeSpan.FromMinutes(1) }); @@ -1658,7 +1694,7 @@ public sealed class CultMeshStreamingTests } [Test] - public async Task ReactiveDocument_SyncedSnapshotEndpointCapturesCanonicalCorrectionAcrossRuntimeBoundary() + public async Task SyncedSnapshotEndpointObservesCanonicalCorrectionAcrossRuntimeBoundary() { var sourceCache = new CultCache(); var registry = new CultNetDocumentRegistry(CultDocumentRegistry.Shared); @@ -1700,18 +1736,10 @@ public sealed class CultMeshStreamingTests { DocumentRegistry = registry } - }); + }); var handle = surface.SyncTo(node).Document(key.Value); - using var reactive = await handle.ReactiveAsync(new CultMeshReactiveDocumentOptions - { - FlushDelay = TimeSpan.FromMinutes(1) - }); - - reactive.Update(document => - { - document.Text = "local-prediction"; - document.Revision = 2; - }); + MeshNoteAliasDocument observed = null!; + using var subscription = handle.Watch(value => observed = value); await sourceCache.UpsertAsync(new MeshNoteDocument { Schema = "tests.mesh_note.v1", @@ -1719,11 +1747,9 @@ public sealed class CultMeshStreamingTests Revision = 7 }, new CultRecordHandle(key)); - await WaitForAsync(() => reactive.Reconciliation?.Canonical.Text == "remote-authority-correction"); + await WaitForAsync(() => observed?.Text == "remote-authority-correction"); - reactive.Current.Text.Should().Be("local-prediction"); - reactive.Reconciliation!.Predicted.Text.Should().Be("local-prediction"); - reactive.Reconciliation.Canonical.Revision.Should().Be(7); + observed.Revision.Should().Be(7); node.Cache.Get(key)!.Text.Should().Be("remote-authority-correction"); } @@ -1796,13 +1822,13 @@ public sealed class CultMeshStreamingTests DocumentRegistry = registry } }); - using var reactive = await surface + var reconstructed = await surface .SyncTo(freshRuntime) .Documents(CultMesh.SnapshotDocument(key.Value)) - .ReactiveAsync(); + .LatestAsync(); - reactive.Current.Text.Should().Be("after-reboot-authority"); - reactive.Current.Revision.Should().Be(4); + reconstructed.Text.Should().Be("after-reboot-authority"); + reconstructed.Revision.Should().Be(4); freshRuntime.Cache.Get(key)!.Text.Should().Be("after-reboot-authority"); } @@ -1849,17 +1875,11 @@ public sealed class CultMeshStreamingTests DocumentRegistry = registry } }); - using var reactive = await surface + var readOnlyHandle = surface .SyncTo(node) - .Document(key.Value) - .ReactiveAsync(new CultMeshReactiveDocumentOptions - { - FlushDelay = TimeSpan.FromMinutes(1) - }); - reactive.Update(document => document.Text = "observer-should-not-write"); - - Func flush = () => reactive.FlushAsync(); - await flush.Should().ThrowAsync(); + .Document(key.Value); + readOnlyHandle.CanReplace.Should().BeFalse(); + readOnlyHandle.CanSubmitPrediction.Should().BeFalse(); await sourceCache.UpsertAsync(new MeshNoteDocument { @@ -1867,16 +1887,14 @@ public sealed class CultMeshStreamingTests Text = "fresh-authority", Revision = 5 }, new CultRecordHandle(key)); - await reactive.RefreshAsync(); - await reactive.FlushAsync(); + var refreshed = await readOnlyHandle.LatestAsync(); - reactive.Current.Text.Should().Be("fresh-authority"); - reactive.IsDirty.Should().BeFalse(); + refreshed.Text.Should().Be("fresh-authority"); node.Cache.Get(key)!.Revision.Should().Be(5); } [Test] - public async Task ReactiveDocument_MarkDirtyFlushesDirectMemberEdits() + public async Task ReactiveDocument_SnapshotMutationCannotPublish() { var subject = new Subject(); var current = new MeshNoteDocument @@ -1905,24 +1923,24 @@ public sealed class CultMeshStreamingTests return Task.CompletedTask; }); - using var reactive = await handle.ReactiveAsync( + using var reactive = await handle.PredictionWriter().ReactiveAsync( new CultMeshReactiveDocumentOptions { FlushDelay = TimeSpan.FromMinutes(1) }); - reactive.Current.Text = "direct-member-edit"; - reactive.Current.Revision = 12; - reactive.MarkDirty(); + var escapedSnapshot = reactive.Snapshot; + escapedSnapshot.Text = "must-not-publish"; + escapedSnapshot.Revision = 12; await reactive.FlushAsync(); - predictions.Should().ContainSingle(); - predictions[0].Text.Should().Be("direct-member-edit"); - predictions[0].Revision.Should().Be(12); + predictions.Should().BeEmpty(); + reactive.Snapshot.Text.Should().Be("initial"); + reactive.IsDirty.Should().BeFalse(); } [Test] - public async Task ReactiveDocument_DetectsDirectMemberEditsWithoutMarkDirty() + public async Task ReactiveDocument_IdleMirrorDoesNotPublishUntilExplicitUpdate() { var subject = new Subject(); var current = new MeshNoteDocument @@ -1951,26 +1969,30 @@ public sealed class CultMeshStreamingTests return Task.CompletedTask; }); - using var reactive = await handle.ReactiveAsync( + using var reactive = await handle.PredictionWriter().ReactiveAsync( new CultMeshReactiveDocumentOptions { FlushDelay = TimeSpan.FromMilliseconds(10) }); - reactive.Current.Text = "direct-member-edit"; - reactive.Current.Revision = 2; - reactive.Current.Text = "direct-member-edit-final"; - reactive.Current.Revision = 3; + await Task.Delay(75); + predictions.Should().BeEmpty(); + + reactive.Update(document => + { + document.Text = "explicit-member-edit"; + document.Revision = 3; + }); await WaitForAsync(() => predictions.Count == 1); - predictions[0].Text.Should().Be("direct-member-edit-final"); + predictions[0].Text.Should().Be("explicit-member-edit"); predictions[0].Revision.Should().Be(3); reactive.IsDirty.Should().BeFalse(); } [Test] - public async Task ReactiveDocument_FlushAsyncCapturesDirectMemberEdit() + public async Task ReactiveDocument_ReplaceLocalFlushesWholeDocument() { var subject = new Subject(); var current = new MeshNoteDocument @@ -1999,18 +2021,22 @@ public sealed class CultMeshStreamingTests return Task.CompletedTask; }); - using var reactive = await handle.ReactiveAsync( + using var reactive = await handle.PredictionWriter().ReactiveAsync( new CultMeshReactiveDocumentOptions { FlushDelay = TimeSpan.FromMinutes(1) }); - reactive.Current.Text = "direct-member-flush"; - reactive.Current.Revision = 4; + reactive.ReplaceLocal(new MeshNoteDocument + { + Schema = "tests.mesh_note.v1", + Text = "whole-document-replacement", + Revision = 4 + }); await reactive.FlushAsync(); predictions.Should().ContainSingle(); - predictions[0].Text.Should().Be("direct-member-flush"); + predictions[0].Text.Should().Be("whole-document-replacement"); predictions[0].Revision.Should().Be(4); } @@ -2036,7 +2062,7 @@ public sealed class CultMeshStreamingTests return Task.CompletedTask; }); - using var reactive = await handle.ReactiveAsync(); + using var reactive = await handle.AuthoritativeWriter().ReactiveAsync(); await reactive.FlushAsync(); @@ -2081,7 +2107,7 @@ public sealed class CultMeshStreamingTests subject.OnNext(value); }); - using var reactive = await handle.ReactiveAsync( + using var reactive = await handle.PredictionWriter().ReactiveAsync( new CultMeshReactiveDocumentOptions { FlushDelay = TimeSpan.FromMinutes(1) @@ -2138,7 +2164,7 @@ public sealed class CultMeshStreamingTests return Task.CompletedTask; }); - using var reactive = await handle.ReactiveAsync( + using var reactive = await handle.PredictionWriter().ReactiveAsync( new CultMeshReactiveDocumentOptions { FlushDelay = TimeSpan.FromMilliseconds(35) @@ -2161,7 +2187,7 @@ public sealed class CultMeshStreamingTests } [Test] - public async Task ReactiveDocument_DirectEditsDuringInFlightFlushQueueLatestPrediction() + public async Task ReactiveDocument_ExplicitEditsDuringInFlightFlushQueueLatestPrediction() { var subject = new Subject(); var current = new MeshNoteDocument @@ -2197,24 +2223,30 @@ public sealed class CultMeshStreamingTests subject.OnNext(value); }); - using var reactive = await handle.ReactiveAsync( + using var reactive = await handle.PredictionWriter().ReactiveAsync( new CultMeshReactiveDocumentOptions { FlushDelay = TimeSpan.FromMinutes(1) }); - reactive.Current.Text = "first-direct"; - reactive.Current.Revision = 2; + reactive.Update(document => + { + document.Text = "first-explicit"; + document.Revision = 2; + }); var flush = reactive.FlushAsync(); await firstWriteStarted.Task; - reactive.Current.Text = "second-direct"; - reactive.Current.Revision = 3; + reactive.Update(document => + { + document.Text = "second-explicit"; + document.Revision = 3; + }); allowFirstWrite.SetResult(true); await flush; predictions.Should().HaveCount(2); - predictions[0].Text.Should().Be("first-direct"); - predictions[1].Text.Should().Be("second-direct"); + predictions[0].Text.Should().Be("first-explicit"); + predictions[1].Text.Should().Be("second-explicit"); reactive.IsDirty.Should().BeFalse(); } @@ -2246,7 +2278,7 @@ public sealed class CultMeshStreamingTests return Task.CompletedTask; }); - using var reactive = await handle.ReactiveAsync( + using var reactive = await handle.PredictionWriter().ReactiveAsync( new CultMeshReactiveDocumentOptions { FlushDelay = TimeSpan.FromMinutes(1) @@ -2259,7 +2291,7 @@ public sealed class CultMeshStreamingTests await reactive.FlushAsync(); - reactive.Current.Text.Should().Be("accepted"); + reactive.Snapshot.Text.Should().Be("accepted"); reactive.Reconciliation.Should().BeNull(); reactive.IsDirty.Should().BeFalse(); }