Make CultMesh document writes authority-explicit

This commit is contained in:
Meta Vens 2026-08-17 19:47:10 +02:00
commit 98bec47d75
5 changed files with 485 additions and 338 deletions

View file

@ -493,56 +493,6 @@ namespace GameCult.Mesh
return Document<TDocument>(recordKey, verse.Context, documentId, sources, routeHint);
}
/// <summary>
/// Creates a managed reactive document mirror over one distributed record hosted by this node.
/// </summary>
public Task<CultMeshReactiveDocument<TDocument>> ReactiveDocumentAsync<TDocument>(
CultRecordKey key,
CultMeshVerseContext context,
CultMeshReactiveDocumentOptions? options = null)
where TDocument : class
{
return Document<TDocument>(key, context).ReactiveAsync(options);
}
/// <summary>
/// Creates a managed reactive document mirror over one distributed record hosted by this node.
/// </summary>
public Task<CultMeshReactiveDocument<TDocument>> ReactiveDocumentAsync<TDocument>(
CultRecordKey key,
CultMeshVerse verse,
CultMeshReactiveDocumentOptions? options = null)
where TDocument : class
{
if (verse == null) throw new ArgumentNullException(nameof(verse));
return ReactiveDocumentAsync<TDocument>(key, verse.Context, options);
}
/// <summary>
/// Creates a managed reactive document mirror over one distributed record hosted by this node.
/// </summary>
public Task<CultMeshReactiveDocument<TDocument>> ReactiveDocumentAsync<TDocument>(
string recordKey,
CultMeshVerseContext context,
CultMeshReactiveDocumentOptions? options = null)
where TDocument : class
{
return ReactiveDocumentAsync<TDocument>(new CultRecordKey(recordKey), context, options);
}
/// <summary>
/// Creates a managed reactive document mirror over one distributed record hosted by this node.
/// </summary>
public Task<CultMeshReactiveDocument<TDocument>> ReactiveDocumentAsync<TDocument>(
string recordKey,
CultMeshVerse verse,
CultMeshReactiveDocumentOptions? options = null)
where TDocument : class
{
if (verse == null) throw new ArgumentNullException(nameof(verse));
return ReactiveDocumentAsync<TDocument>(recordKey, verse.Context, options);
}
/// <inheritdoc />
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<TDocument>(
var feed = LiveFeed<CultMeshDocumentQueryParameters, TDocument>(
ResolveDocumentId(documentId, key),
context,
async _ => await ReadDatabaseDocumentRequiredAsync<TDocument>(database, key).ConfigureAwait(false),
_ => WatchDatabaseRecordAs<TDocument>(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<TDocument>(database, key),
(_parameters, _context) => WatchDatabaseRecordAs<TDocument>(database, key),
sourceList,
route);
Func<TDocument, Task>? replace = database.CanWriteAuthoritatively<TDocument>(key)
? value => PutDatabaseDocumentAsync(database, key, value, predicted: false)
: null;
Func<TDocument, Task>? submitPrediction = database.CanSubmitPrediction<TDocument>(key)
? value => PutDatabaseDocumentAsync(database, key, value, predicted: true)
: null;
return new CultMeshDocumentHandle<TDocument>(
BindLiveFeed(context, feed),
replace,
submitPrediction);
}
/// <summary>

View file

@ -911,9 +911,6 @@ namespace GameCult.Mesh
/// <summary>Gets whether this handle can submit a client prediction for the underlying document value.</summary>
bool CanSubmitPrediction { get; }
/// <summary>Gets whether this handle can accept a transparent document mutation.</summary>
bool CanSet { get; }
/// <summary>Creates a same-schema alias presentation for another CLR document type.</summary>
CultMeshDocumentHandle<TAlias> AsSchemaAlias<TAlias>() where TAlias : class;
}
@ -975,9 +972,6 @@ namespace GameCult.Mesh
/// <summary>Gets whether this handle can submit a client prediction for the underlying document value.</summary>
public bool CanSubmitPrediction => _submitPrediction != null;
/// <summary>Gets whether this handle can accept a transparent document mutation.</summary>
public bool CanSet => CanSubmitPrediction || CanReplace;
/// <summary>Reads one coherent document snapshot.</summary>
public Task<TDocument> LatestAsync()
{
@ -1048,57 +1042,41 @@ namespace GameCult.Mesh
return _submitPrediction(value);
}
/// <summary>
/// Sets the document value through the configured authority shape.
/// Prediction-backed documents publish a prediction; mutable documents replace authoritatively.
/// </summary>
public Task SetAsync(TDocument value)
/// <summary>Gets an explicit authoritative replacement writer for this document.</summary>
public CultMeshDocumentWriter<TDocument> 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<TDocument>(
this,
CultMeshDocumentWriteKind.AuthoritativeReplacement,
_replace);
}
/// <summary>Reads, updates, and sets the document value through the configured authority shape.</summary>
public async Task<TDocument> UpdateAsync(Func<TDocument, TDocument> update)
/// <summary>Gets an explicit client prediction writer for this document.</summary>
public CultMeshDocumentWriter<TDocument> 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<TDocument>(
this,
CultMeshDocumentWriteKind.Prediction,
_submitPrediction);
}
/// <summary>Reads, updates, and sets the document value through the configured authority shape.</summary>
public async Task<TDocument> UpdateAsync(Func<TDocument, Task<TDocument>> 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;
}
/// <summary>
/// Creates a managed reactive document mirror whose local edits can be coalesced into one prediction or replacement.
/// </summary>
public async Task<CultMeshReactiveDocument<TDocument>> ReactiveAsync(
CultMeshReactiveDocumentOptions? options = null)
/// <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 reactive = new CultMeshReactiveDocument<TDocument>(this, current, options);
reactive.Start();
return reactive;
var observed = new CultMeshObservedDocument<TDocument>(this, current);
observed.Start();
return observed;
}
/// <summary>
/// Creates a managed reactive document mirror synchronously for host APIs that cannot be async.
/// </summary>
public CultMeshReactiveDocument<TDocument> Reactive(CultMeshReactiveDocumentOptions? options = null)
/// <summary>Creates a read-only in-memory mirror synchronously.</summary>
public CultMeshObservedDocument<TDocument> Observe()
{
return ReactiveAsync(options).ConfigureAwait(false).GetAwaiter().GetResult();
return ObserveAsync().ConfigureAwait(false).GetAwaiter().GetResult();
}
/// <summary>Creates a same-schema alias presentation for another CLR document type.</summary>
@ -1215,6 +1193,174 @@ namespace GameCult.Mesh
}
}
/// <summary>
/// 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.
/// </summary>
public sealed class CultMeshObservedDocument<TDocument> : IDisposable
where TDocument : class
{
private readonly object _gate = new();
private IDisposable? _subscription;
private TDocument _current;
private bool _disposed;
internal CultMeshObservedDocument(
CultMeshDocumentHandle<TDocument> document,
TDocument current)
{
Document = document ?? throw new ArgumentNullException(nameof(document));
_current = current ?? throw new ArgumentNullException(nameof(current));
}
/// <summary>Gets the underlying read/watch handle.</summary>
public CultMeshDocumentHandle<TDocument> Document { get; }
/// <summary>
/// Gets the current borrowed snapshot. Treat the object graph as read-only; local mutation
/// cannot publish and is replaced by the next canonical snapshot.
/// </summary>
public TDocument Current
{
get
{
lock (_gate)
{
ThrowIfDisposed();
return _current;
}
}
}
internal void Start()
{
_subscription = Document.Watch(ApplyCanonicalSnapshot);
}
/// <summary>Reads and adopts a fresh canonical snapshot.</summary>
public async Task<TDocument> RefreshAsync()
{
ThrowIfDisposed();
var latest = CultMeshDocumentHandle<TDocument>.CloneDocument(
await Document.LatestAsync().ConfigureAwait(false));
lock (_gate)
{
ThrowIfDisposed();
_current = latest;
return _current;
}
}
/// <inheritdoc />
public void Dispose()
{
lock (_gate)
{
if (_disposed)
return;
_disposed = true;
}
_subscription?.Dispose();
}
private void ApplyCanonicalSnapshot(TDocument canonical)
{
if (canonical == null)
return;
var next = CultMeshDocumentHandle<TDocument>.CloneDocument(canonical);
lock (_gate)
{
if (!_disposed)
_current = next;
}
}
private void ThrowIfDisposed()
{
if (_disposed)
throw new ObjectDisposedException(nameof(CultMeshObservedDocument<TDocument>));
}
}
/// <summary>Names the one immutable authority path selected by a document writer.</summary>
public enum CultMeshDocumentWriteKind
{
/// <summary>The writer replaces canonical state through its owning provider.</summary>
AuthoritativeReplacement,
/// <summary>The writer submits a client prediction or candidate for validation.</summary>
Prediction
}
/// <summary>
/// An explicit, immutable write capability for one typed document. A writer never switches
/// between authoritative replacement and prediction based on runtime availability.
/// </summary>
public sealed class CultMeshDocumentWriter<TDocument>
where TDocument : class
{
private readonly Func<TDocument, Task> _write;
internal CultMeshDocumentWriter(
CultMeshDocumentHandle<TDocument> document,
CultMeshDocumentWriteKind kind,
Func<TDocument, Task> write)
{
Document = document ?? throw new ArgumentNullException(nameof(document));
Kind = kind;
_write = write ?? throw new ArgumentNullException(nameof(write));
}
/// <summary>Gets the document read/watch handle paired with this writer.</summary>
public CultMeshDocumentHandle<TDocument> Document { get; }
/// <summary>Gets the immutable authority path selected for this writer.</summary>
public CultMeshDocumentWriteKind Kind { get; }
/// <summary>Writes one value through this writer's named authority path.</summary>
public Task WriteAsync(TDocument value)
{
if (value == null) throw new ArgumentNullException(nameof(value));
return _write(value);
}
/// <summary>Reads, transforms, and writes through this writer's named authority path.</summary>
public async Task<TDocument> UpdateAsync(Func<TDocument, TDocument> update)
{
if (update == null) throw new ArgumentNullException(nameof(update));
var next = update(await Document.LatestAsync().ConfigureAwait(false));
await WriteAsync(next).ConfigureAwait(false);
return next;
}
/// <summary>Reads, transforms, and writes through this writer's named authority path.</summary>
public async Task<TDocument> UpdateAsync(Func<TDocument, Task<TDocument>> 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;
}
/// <summary>Creates an explicitly writable reactive mirror.</summary>
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();
return reactive;
}
/// <summary>Creates an explicitly writable reactive mirror synchronously.</summary>
public CultMeshReactiveDocument<TDocument> Reactive(CultMeshReactiveDocumentOptions? options = null)
{
return ReactiveAsync(options).ConfigureAwait(false).GetAwaiter().GetResult();
}
}
/// <summary>
/// Configures a managed CultMesh reactive document mirror.
/// </summary>
@ -1225,9 +1371,6 @@ namespace GameCult.Mesh
/// </summary>
public TimeSpan FlushDelay { get; set; } = TimeSpan.FromMilliseconds(16);
/// <summary>Gets or sets whether direct edits to Current should be detected and flushed automatically.</summary>
public bool DetectLocalChanges { get; set; } = true;
/// <summary>Gets or sets whether canonical snapshots replace dirty local predictions immediately.</summary>
public bool ReplaceDirtyCurrentOnCanonicalSnapshot { get; set; }
}
@ -1272,18 +1415,18 @@ namespace GameCult.Mesh
}
/// <summary>
/// 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.
/// </summary>
public sealed class CultMeshReactiveDocument<TDocument> : IDisposable
where TDocument : class
{
private readonly CultMeshDocumentWriter<TDocument> _writer;
private readonly CultMeshDocumentHandle<TDocument> _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<TDocument> document,
CultMeshDocumentWriter<TDocument> 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);
}
/// <summary>Gets the immutable write capability selected for this mirror.</summary>
public CultMeshDocumentWriter<TDocument> Writer => _writer;
/// <summary>Gets the underlying document handle.</summary>
public CultMeshDocumentHandle<TDocument> Document => _document;
/// <summary>Gets the editable local document value.</summary>
public TDocument Current { get; private set; }
/// <summary>
/// Gets a cloned local snapshot. Mutating the returned object cannot mutate or publish the mirror;
/// use <see cref="Update"/> or <see cref="ReplaceLocal"/> for edits.
/// </summary>
public TDocument Snapshot
{
get
{
lock (_gate)
{
ThrowIfDisposed();
return CultMeshDocumentHandle<TDocument>.CloneDocument(_current);
}
}
}
/// <summary>Gets whether local edits are waiting to be sent.</summary>
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);
}
}
/// <summary>Marks the current value dirty and schedules a coalesced prediction or replacement.</summary>
public void MarkDirty()
{
ThrowIfDisposed();
lock (_gate)
{
_dirty = true;
ScheduleFlushLocked();
}
}
/// <summary>Mutates the current value and schedules a coalesced prediction or replacement.</summary>
@ -1351,24 +1491,24 @@ namespace GameCult.Mesh
ThrowIfDisposed();
lock (_gate)
{
update(Current);
update(_current);
_dirty = true;
ScheduleFlushLocked();
return Current;
return CultMeshDocumentHandle<TDocument>.CloneDocument(_current);
}
}
/// <summary>Replaces the current local value and schedules a coalesced prediction or replacement.</summary>
public TDocument SetCurrent(TDocument value)
public TDocument ReplaceLocal(TDocument value)
{
if (value == null) throw new ArgumentNullException(nameof(value));
ThrowIfDisposed();
lock (_gate)
{
Current = CultMeshDocumentHandle<TDocument>.CloneDocument(value);
_current = CultMeshDocumentHandle<TDocument>.CloneDocument(value);
_dirty = true;
ScheduleFlushLocked();
return Current;
return CultMeshDocumentHandle<TDocument>.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<TDocument>.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<TDocument>.CloneDocument(Current);
_lastCleanSnapshot = SerializeDocument(predicted);
predicted = CultMeshDocumentHandle<TDocument>.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<TDocument>.CloneDocument(canonical);
if (_dirty || _flushing)
{
var predicted = CultMeshDocumentHandle<TDocument>.CloneDocument(Current);
var predicted = CultMeshDocumentHandle<TDocument>.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<string, object?> CreateReconciliationDelta(
TDocument predicted,
TDocument canonical)
@ -1778,13 +1883,6 @@ namespace GameCult.Mesh
return TryGetDocument<TDocument>(out var document) && document.CanSubmitPrediction;
}
/// <summary>Gets whether one typed document can accept transparent mutations by CLR type or same-schema alias.</summary>
public bool CanSet<TDocument>()
where TDocument : class
{
return TryGetDocument<TDocument>(out var document) && document.CanSet;
}
/// <summary>Submits a client prediction for one typed document by CLR type or same-schema alias.</summary>
public Task SubmitPredictionAsync<TDocument>(TDocument value)
where TDocument : class
@ -1792,25 +1890,18 @@ namespace GameCult.Mesh
return Document<TDocument>().SubmitPredictionAsync(value);
}
/// <summary>Sets one typed document through its configured authority shape by CLR type or same-schema alias.</summary>
public Task SetAsync<TDocument>(TDocument value)
/// <summary>Gets an explicit authoritative writer by CLR type or same-schema alias.</summary>
public CultMeshDocumentWriter<TDocument> AuthoritativeWriter<TDocument>()
where TDocument : class
{
return Document<TDocument>().SetAsync(value);
return Document<TDocument>().AuthoritativeWriter();
}
/// <summary>Reads, updates, and sets one typed document through its configured authority shape.</summary>
public Task<TDocument> UpdateAsync<TDocument>(Func<TDocument, TDocument> update)
/// <summary>Gets an explicit prediction writer by CLR type or same-schema alias.</summary>
public CultMeshDocumentWriter<TDocument> PredictionWriter<TDocument>()
where TDocument : class
{
return Document<TDocument>().UpdateAsync(update);
}
/// <summary>Reads, updates, and sets one typed document through its configured authority shape.</summary>
public Task<TDocument> UpdateAsync<TDocument>(Func<TDocument, Task<TDocument>> update)
where TDocument : class
{
return Document<TDocument>().UpdateAsync(update);
return Document<TDocument>().PredictionWriter();
}
/// <summary>Watches one typed document by CLR type or same-schema alias.</summary>
@ -1827,21 +1918,6 @@ namespace GameCult.Mesh
return Document<TDocument>().Watch(onNext);
}
/// <summary>Creates a managed reactive document mirror by CLR type or same-schema alias.</summary>
public Task<CultMeshReactiveDocument<TDocument>> ReactiveAsync<TDocument>(
CultMeshReactiveDocumentOptions? options = null)
where TDocument : class
{
return Document<TDocument>().ReactiveAsync(options);
}
/// <summary>Creates a managed reactive document mirror synchronously by CLR type or same-schema alias.</summary>
public CultMeshReactiveDocument<TDocument> Reactive<TDocument>(
CultMeshReactiveDocumentOptions? options = null)
where TDocument : class
{
return Document<TDocument>().Reactive(options);
}
}
/// <summary>

View file

@ -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<T>` owns typed snapshot and watch access;
- an authoritative writer owns canonical replacement capability;
- a prediction writer owns candidate/prediction submission capability;
- `CultMeshReactiveDocument<T>` 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.

View file

@ -503,6 +503,22 @@ namespace GameCult.Networking
/// </summary>
public IReadOnlyList<CultNetClientAuthorityScope> ClientAuthorityScopes => _clientAuthorityScopes;
/// <summary>Gets whether this database instance can authoritatively replace the document at a key.</summary>
public bool CanWriteAuthoritatively<T>(CultRecordKey key) where T : class
{
ThrowIfDisposed();
var descriptor = _cache.Registry.GetRequired<T>();
return ResolveShardInternal(descriptor, key).IsPrimary;
}
/// <summary>Gets whether this database instance can submit a prediction for the document at a key.</summary>
public bool CanSubmitPrediction<T>(CultRecordKey key) where T : class
{
ThrowIfDisposed();
var descriptor = _cache.Registry.GetRequired<T>();
return _clientAuthorityScopes.Any(scope => scope.Matches(_runtimeId, descriptor, key));
}
/// <summary>
/// Gets mutation log entries for a shard after the supplied sequence.
/// </summary>

View file

@ -940,16 +940,17 @@ public sealed class CultMeshStreamingTests
CultMesh.Verse("starbridge", "pilot-a"));
var alias = handle.AsSchemaAlias<MeshNoteAliasDocument>();
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<MeshNoteAliasDocument>().Should().BeTrue();
catalog.CanSet<MeshNoteAliasDocument>().Should().BeTrue();
var predictionWriter = catalog.PredictionWriter<MeshNoteAliasDocument>();
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<MeshNoteAliasDocument>(
using var reactive = await catalog.PredictionWriter<MeshNoteAliasDocument>().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<MeshNoteDocument>();
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<MeshNoteDocument>();
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<MeshNoteDocument>());
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<Task> act = () => reactive.FlushAsync();
authoritative.Should().Throw<NotSupportedException>();
prediction.Should().Throw<NotSupportedException>();
}
await act.Should().ThrowAsync<NotSupportedException>();
[Test]
public async Task ObservedDocument_WatchesWithoutWriteCapabilityOrIdleTimer()
{
var subject = new Subject<MeshNoteDocument>();
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<MeshNoteDocument>)
.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<MeshNoteAliasDocument>(
.PredictionWriter<MeshNoteAliasDocument>()
.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<MeshNoteAliasDocument>()
.ReactiveAsync(new CultMeshReactiveDocumentOptions
{
FlushDelay = TimeSpan.FromMinutes(1)
});
.Watch(value => commanderSnapshot = value);
using var pilotReactive = await pilotHandle
.AsSchemaAlias<MeshNoteAliasDocument>()
.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<MeshNoteDocument>(key)!.Text.Should().Be("pilot-coalesced-order");
}
[Test]
public async Task NodeReactiveDocument_OpensSameSchemaAliasWithOneCall()
public async Task NodeDocument_OpensExplicitPredictionWriterForSameSchemaAlias()
{
var schemaId = CultDocumentRegistry.Shared.GetRequired<MeshNoteDocument>().SchemaId;
using var temp = new TemporaryDirectory();
@ -1630,10 +1666,10 @@ public sealed class CultMeshStreamingTests
Revision = 1
}, new CultRecordHandle<MeshNoteDocument>(key));
using var pilotReactive = await node.ReactiveDocumentAsync<MeshNoteAliasDocument>(
key,
CultMesh.Verse("starbridge", "pilot-a"),
new CultMeshReactiveDocumentOptions
using var pilotReactive = await node
.Document<MeshNoteAliasDocument>(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<MeshNoteAliasDocument>(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<MeshNoteDocument>(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<MeshNoteDocument>(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<MeshNoteDocument>(key.Value))
.ReactiveAsync<MeshNoteAliasDocument>();
.LatestAsync<MeshNoteAliasDocument>();
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<MeshNoteDocument>(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<MeshNoteAliasDocument>(key.Value)
.ReactiveAsync(new CultMeshReactiveDocumentOptions
{
FlushDelay = TimeSpan.FromMinutes(1)
});
reactive.Update(document => document.Text = "observer-should-not-write");
Func<Task> flush = () => reactive.FlushAsync();
await flush.Should().ThrowAsync<NotSupportedException>();
.Document<MeshNoteAliasDocument>(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<MeshNoteDocument>(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<MeshNoteDocument>(key)!.Revision.Should().Be(5);
}
[Test]
public async Task ReactiveDocument_MarkDirtyFlushesDirectMemberEdits()
public async Task ReactiveDocument_SnapshotMutationCannotPublish()
{
var subject = new Subject<MeshNoteDocument>();
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<MeshNoteDocument>();
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<MeshNoteDocument>();
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<MeshNoteDocument>();
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();
}