From ccd5a0e9cfde57599b4d9bb8716ff613eb02e751 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sat, 27 Jun 2026 19:19:09 +0100 Subject: [PATCH] Add managed CultMesh reactive documents --- src/GameCult.Mesh/CultMeshPrimitives.cs | 298 +++++++++++++++++- .../CultMeshStreamingTests.cs | 106 +++++++ 2 files changed, 403 insertions(+), 1 deletion(-) diff --git a/src/GameCult.Mesh/CultMeshPrimitives.cs b/src/GameCult.Mesh/CultMeshPrimitives.cs index 267b330a..789a0b24 100644 --- a/src/GameCult.Mesh/CultMeshPrimitives.cs +++ b/src/GameCult.Mesh/CultMeshPrimitives.cs @@ -1052,6 +1052,26 @@ namespace GameCult.Mesh 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) + { + var current = CloneDocument(await LatestAsync().ConfigureAwait(false)); + var reactive = new CultMeshReactiveDocument(this, current, options); + reactive.Start(); + return reactive; + } + + /// + /// Creates a managed reactive document mirror synchronously for host APIs that cannot be async. + /// + public CultMeshReactiveDocument Reactive(CultMeshReactiveDocumentOptions? options = null) + { + return ReactiveAsync(options).ConfigureAwait(false).GetAwaiter().GetResult(); + } + /// Creates a same-schema alias presentation for another CLR document type. public CultMeshDocumentHandle AsSchemaAlias() where TAlias : class { @@ -1085,7 +1105,17 @@ namespace GameCult.Mesh return new CultMeshDocumentHandle(aliasFeed.Bind(Context), replace, submitPrediction); } - private static TTarget ConvertDocument(TSource document) + internal static TDocumentValue CloneDocument(TDocumentValue document) + where TDocumentValue : class + { + if (document == null) throw new ArgumentNullException(nameof(document)); + var payload = CultDocumentMessagePackSerialization.SerializeUntyped(document, typeof(TDocumentValue)); + return (TDocumentValue)CultDocumentMessagePackSerialization.DeserializeUntyped( + typeof(TDocumentValue), + payload); + } + + internal static TTarget ConvertDocument(TSource document) where TSource : class where TTarget : class { @@ -1100,6 +1130,256 @@ namespace GameCult.Mesh } } + /// + /// Configures a managed CultMesh reactive document mirror. + /// + public sealed class CultMeshReactiveDocumentOptions + { + /// + /// Gets or sets the write coalescing window. The default approximates a frame boundary for non-Unity hosts. + /// + public TimeSpan FlushDelay { get; set; } = TimeSpan.FromMilliseconds(16); + + /// Gets or sets whether canonical snapshots replace dirty local predictions immediately. + public bool ReplaceDirtyCurrentOnCanonicalSnapshot { get; set; } + } + + /// + /// Captures a canonical snapshot that arrived while the local reactive document had an outstanding prediction. + /// + public sealed class CultMeshReactiveDocumentReconciliation + where TDocument : class + { + internal CultMeshReactiveDocumentReconciliation( + TDocument canonical, + TDocument predicted, + DateTimeOffset receivedAt) + { + Canonical = canonical; + Predicted = predicted; + ReceivedAt = receivedAt; + } + + /// Gets the authoritative canonical document snapshot. + public TDocument Canonical { get; } + + /// Gets the locally predicted document snapshot that was active when the canonical value arrived. + public TDocument Predicted { get; } + + /// Gets when the canonical snapshot was received. + public DateTimeOffset ReceivedAt { get; } + } + + /// + /// Managed typed document mirror that keeps an editable current value and coalesces local edits into CultMesh mutations. + /// + public sealed class CultMeshReactiveDocument : IDisposable + where TDocument : class + { + private readonly CultMeshDocumentHandle _document; + private readonly CultMeshReactiveDocumentOptions _options; + private readonly object _gate = new(); + private IDisposable? _subscription; + private Timer? _timer; + private bool _dirty; + private bool _flushQueued; + private bool _flushing; + private bool _disposed; + + internal CultMeshReactiveDocument( + CultMeshDocumentHandle document, + TDocument current, + CultMeshReactiveDocumentOptions? options) + { + _document = document ?? throw new ArgumentNullException(nameof(document)); + Current = current ?? throw new ArgumentNullException(nameof(current)); + _options = options ?? new CultMeshReactiveDocumentOptions(); + } + + /// Gets the underlying document handle. + public CultMeshDocumentHandle Document => _document; + + /// Gets the editable local document value. + public TDocument Current { get; private set; } + + /// Gets whether local edits are waiting to be sent. + public bool IsDirty + { + get + { + lock (_gate) + return _dirty; + } + } + + /// Gets the most recent reconciliation snapshot, when a canonical value arrived during local prediction. + public CultMeshReactiveDocumentReconciliation? Reconciliation { get; private set; } + + internal void Start() + { + _subscription = _document.Watch(ApplyCanonicalSnapshot); + } + + /// 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. + public TDocument Update(Action update) + { + if (update == null) throw new ArgumentNullException(nameof(update)); + ThrowIfDisposed(); + lock (_gate) + { + update(Current); + _dirty = true; + ScheduleFlushLocked(); + return Current; + } + } + + /// Replaces the current local value and schedules a coalesced prediction or replacement. + public TDocument SetCurrent(TDocument value) + { + if (value == null) throw new ArgumentNullException(nameof(value)); + ThrowIfDisposed(); + lock (_gate) + { + Current = CultMeshDocumentHandle.CloneDocument(value); + _dirty = true; + ScheduleFlushLocked(); + return Current; + } + } + + /// Reads a fresh canonical snapshot and adopts it as the current value. + public async Task RefreshAsync() + { + ThrowIfDisposed(); + var latest = CultMeshDocumentHandle.CloneDocument( + await _document.LatestAsync().ConfigureAwait(false)); + lock (_gate) + { + Current = latest; + _dirty = false; + Reconciliation = null; + return Current; + } + } + + /// Immediately sends the latest local dirty value, if any, through the document authority shape. + public async Task FlushAsync() + { + TDocument predicted; + lock (_gate) + { + ThrowIfDisposed(); + if (!_dirty) + return; + if (_flushing) + { + _flushQueued = true; + return; + } + + _timer?.Change(Timeout.InfiniteTimeSpan, Timeout.InfiniteTimeSpan); + _flushing = true; + _dirty = false; + predicted = CultMeshDocumentHandle.CloneDocument(Current); + } + + try + { + await _document.SetAsync(predicted).ConfigureAwait(false); + } + finally + { + var shouldFlushAgain = false; + lock (_gate) + { + _flushing = false; + shouldFlushAgain = _flushQueued || _dirty; + _flushQueued = false; + } + + if (shouldFlushAgain) + await FlushAsync().ConfigureAwait(false); + } + } + + /// Clears the most recent reconciliation metadata. + public void ClearReconciliation() + { + ThrowIfDisposed(); + Reconciliation = null; + } + + /// + public void Dispose() + { + lock (_gate) + { + if (_disposed) + return; + _disposed = true; + } + + _subscription?.Dispose(); + _timer?.Dispose(); + } + + private void ApplyCanonicalSnapshot(TDocument canonical) + { + if (canonical == null) + return; + + lock (_gate) + { + if (_disposed) + return; + + var nextCanonical = CultMeshDocumentHandle.CloneDocument(canonical); + if (_dirty || _flushing) + { + Reconciliation = new CultMeshReactiveDocumentReconciliation( + nextCanonical, + CultMeshDocumentHandle.CloneDocument(Current), + DateTimeOffset.UtcNow); + if (!_options.ReplaceDirtyCurrentOnCanonicalSnapshot) + return; + } + + Current = nextCanonical; + Reconciliation = null; + } + } + + private void ScheduleFlushLocked() + { + if (_options.FlushDelay <= TimeSpan.Zero) + { + _ = Task.Run(FlushAsync); + return; + } + + _timer ??= new Timer(_ => _ = Task.Run(FlushAsync)); + _timer.Change(_options.FlushDelay, Timeout.InfiniteTimeSpan); + } + + private void ThrowIfDisposed() + { + if (_disposed) + throw new ObjectDisposedException(nameof(CultMeshReactiveDocument)); + } + } + /// /// Schema-aware lookup table for typed CultMesh document handles. /// @@ -1285,6 +1565,22 @@ 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/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs b/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs index d228788a..48812792 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs +++ b/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs @@ -892,6 +892,112 @@ public sealed class CultMeshStreamingTests cache.Get(key)!.Revision.Should().Be(6); } + [Test] + public async Task ReactiveDocument_CoalescesLocalEditsIntoOnePrediction() + { + var subject = new Subject(); + var current = new MeshNoteDocument + { + Schema = "tests.mesh_note.v1", + Text = "initial", + Revision = 1 + }; + var predictions = new List(); + var handle = CultMesh.Document( + "mesh.note.reactive", + CultMesh.Verse("starbridge", "pilot-a"), + _ => Task.FromResult(current), + _ => subject, + value => + { + current = value; + subject.OnNext(value); + return Task.CompletedTask; + }, + value => + { + predictions.Add(value); + current = value; + subject.OnNext(value); + return Task.CompletedTask; + }); + var catalog = CultMesh.Documents(handle); + + using var reactive = await catalog.ReactiveAsync( + new CultMeshReactiveDocumentOptions + { + FlushDelay = TimeSpan.FromMinutes(1) + }); + + reactive.Current.Text.Should().Be("initial"); + reactive.Update(document => + { + document.Text = "first-local-edit"; + document.Revision++; + }); + reactive.Update(document => + { + document.Text = "second-local-edit"; + document.Revision++; + }); + + predictions.Should().BeEmpty(); + + await reactive.FlushAsync(); + + predictions.Should().ContainSingle(); + predictions[0].Text.Should().Be("second-local-edit"); + predictions[0].Revision.Should().Be(3); + current.Text.Should().Be("second-local-edit"); + } + + [Test] + public async Task ReactiveDocument_TracksCanonicalSnapshotDuringDirtyPrediction() + { + var subject = new Subject(); + var current = new MeshNoteDocument + { + Schema = "tests.mesh_note.v1", + Text = "initial", + Revision = 1 + }; + var handle = CultMesh.Document( + "mesh.note.reconciliation", + CultMesh.Verse("starbridge", "pilot-a"), + _ => Task.FromResult(current), + _ => subject, + value => + { + current = value; + subject.OnNext(value); + return Task.CompletedTask; + }); + + using var reactive = await handle.ReactiveAsync( + new CultMeshReactiveDocumentOptions + { + FlushDelay = TimeSpan.FromMinutes(1) + }); + + reactive.Update(document => + { + document.Text = "local-prediction"; + document.Revision = 2; + }); + + subject.OnNext(new MeshNoteDocument + { + Schema = "tests.mesh_note.v1", + Text = "canonical-correction", + Revision = 7 + }); + + reactive.Current.Text.Should().Be("local-prediction"); + reactive.Reconciliation.Should().NotBeNull(); + reactive.Reconciliation!.Canonical.Text.Should().Be("canonical-correction"); + reactive.Reconciliation.Predicted.Text.Should().Be("local-prediction"); + } + [Test] public async Task DocumentHandle_ReadsSchemaPublicationsFromSingleFileStores() {