diff --git a/src/GameCult.Caching/Contracts/cultcache-store-composition.md b/src/GameCult.Caching/Contracts/cultcache-store-composition.md index 83aaa61b..ca0abd61 100644 --- a/src/GameCult.Caching/Contracts/cultcache-store-composition.md +++ b/src/GameCult.Caching/Contracts/cultcache-store-composition.md @@ -44,21 +44,19 @@ which route wrote it. single file keeps staged writes dirty and does not drop loaded records. - A load publishes changes to `Watch` subscribers and fires `OnUpdate`. A local write or commit publishes to `Watch` only; `OnUpdate` fires for loads alone. +- Every change carries a `Sequence`: a per-cache, in-memory number assigned + when the cache admits the change, under its gate. It increases in admission + order and is not persisted. - Observers and `OnUpdate` run after the cache releases its gate, never under - it: an observer may read or write the cache from any thread. A change is - published after the cache releases its gate and before the call that made it - returns, on that caller's thread; no call publishes another call's changes. A - direct `PullAll` on an attached store is such a call. -- Changes are delivered in save order. Across threads, a cache delivers changes - in the order it admitted them, so a later change to a key never reaches an - observer before an earlier one. A call whose changes were admitted after - another call's waits, outside the gate, until that delivery finishes. -- A write made by an observer during delivery is not delivered before that - write returns. It is queued behind the delivery in progress, in save order, - and delivered on the same thread before the outermost call returns; its - `OnUpdate` exceptions are rethrown to that outermost caller. -- An observer must not block waiting on another thread's write to a cache: - that thread's delivery may be waiting for this one to finish. + it: an observer may read or write the cache from any thread. Each call + publishes exactly its own changes, on its own thread, after releasing the gate + and before it returns; no call publishes another call's changes or waits on + another call's delivery. A write made by an observer is its own call and + publishes its own changes before that write returns. A direct `PullAll` on an + attached store is such a call. +- Cross-thread delivery order is not guaranteed. A consumer that keeps a latest + value per record ignores a change whose `Sequence` is lower than the one it + already applied for that record. - An `OnUpdate` handler exception is rethrown to that caller after all of that call's changes are delivered (an `AggregateException` if several threw). If the call itself failed, its own exception is rethrown and handler exceptions diff --git a/src/GameCult.Caching/CultCache.cs b/src/GameCult.Caching/CultCache.cs index ddc5c7a1..da005717 100644 --- a/src/GameCult.Caching/CultCache.cs +++ b/src/GameCult.Caching/CultCache.cs @@ -1180,13 +1180,9 @@ namespace GameCult.Caching private readonly ConditionalWeakTable _handles = new(); private readonly Subject _changes = new(); private readonly object _gate = new(); - private readonly object _deliveryGate = new(); - private long _admitted; - private long _delivered; + private long _sequence; // The changes admitted by this thread's outermost hold, published by that hold when it exits. [ThreadStatic] private static List<(Change Change, bool Loaded)>? _held; - // Non-null while this thread delivers: tickets it has yet to deliver, in the order its holds closed. - [ThreadStatic] private static List<(CultCache Cache, long Ticket, List<(Change Change, bool Loaded)> Changes)>? _delivering; private bool _dirtyInMemory; public CultCache(CultDocumentRegistry? registry = null) @@ -1247,7 +1243,8 @@ namespace GameCult.Caching change.Kind, change.Stored.Key, change.Document as T, - change.Previous as T)); + change.Previous as T, + change.Sequence)); } public Observable> WatchRecord(CultRecordKey key) where T : class @@ -1583,11 +1580,12 @@ namespace GameCult.Caching return CultCommitOutcome.Committed; }); - // Every admission runs in a hold. A nested hold on the same cache adds to the outermost one, which takes a ticket - // under the gate and, after leaving it, publishes exactly its own changes in ticket order, on its caller's thread: - // observers never run under the gate and see changes in save order. A hold closed while its thread is delivering - // queues its ticket behind that delivery. Every change is delivered even if an OnUpdate handler throws; then the - // first handler exception (an AggregateException for several) is rethrown. If the body threw, its exception wins. + // Every admission runs in a hold. A nested hold on the same cache adds to the outermost one, which publishes + // exactly its own changes after it leaves the gate, before it returns, on its caller's thread: observers never + // run under the gate, and a write an observer makes is a new outermost hold. Cross-thread delivery order is not + // guaranteed; each change carries the Sequence it was admitted with. Every change is delivered even if an + // OnUpdate handler throws; then the first handler exception (an AggregateException for several) is rethrown. + // If the body threw, its exception wins. internal T Held(Func body) { if (Monitor.IsEntered(_gate)) @@ -1596,27 +1594,21 @@ namespace GameCult.Caching throw new InvalidOperationException( "A cache hold was entered inside another cache's hold; a thread holds one cache's gate at a time."); var mine = _held = new List<(Change Change, bool Loaded)>(); - var ticket = -1L; - T result = default!; - ExceptionDispatchInfo? thrown = null; - lock (_gate) + T result; + try { - try - { + lock (_gate) result = body(); - } - catch (Exception exception) - { - thrown = ExceptionDispatchInfo.Capture(exception); - } - + } + catch + { _held = null; - if (mine.Count > 0) - ticket = _admitted++; + Publish(mine); + throw; } - var failures = ticket < 0 ? new List() : Publish(ticket, mine); - thrown?.Throw(); + _held = null; + var failures = Publish(mine); if (failures.Count == 1) ExceptionDispatchInfo.Capture(failures[0]).Throw(); if (failures.Count > 1) @@ -1624,80 +1616,28 @@ namespace GameCult.Caching return result; } - private List Publish(long ticket, List<(Change Change, bool Loaded)> changes) + private List Publish(List<(Change Change, bool Loaded)> changes) { var failures = new List(); - if (_delivering != null) + foreach (var (change, loaded) in changes) { - _delivering.Add((this, ticket, changes)); - return failures; - } - - var queue = _delivering = new List<(CultCache Cache, long Ticket, List<(Change Change, bool Loaded)> Changes)> - { - (this, ticket, changes) - }; - try - { - // A queued ticket that is never delivered would stall every later writer of its cache, so none is skipped. - for (var i = 0; i < queue.Count; i++) + // R3 routes a throwing subscriber to its unhandled-exception handler; OnNext does not throw. + _changes.OnNext(change); + if (!loaded) + continue; + try { - try - { - queue[i].Cache.Deliver(queue[i].Ticket, queue[i].Changes, failures); - } - catch (Exception exception) - { - failures.Add(exception); - } + OnUpdate?.Invoke(change.Previous, change.Document); + } + catch (Exception exception) + { + failures.Add(exception); } - } - finally - { - _delivering = null; } return failures; } - // A thread waits for its turn holding no gate and no turn, and holds a turn only while delivering, which never - // waits: every wait is for a lower ticket, so deliveries cannot deadlock one another. - private void Deliver(long ticket, List<(Change Change, bool Loaded)> changes, List failures) - { - lock (_deliveryGate) - { - while (_delivered != ticket) - Monitor.Wait(_deliveryGate); - } - - try - { - foreach (var (change, loaded) in changes) - { - // R3 routes a throwing subscriber to its unhandled-exception handler; OnNext does not throw. - _changes.OnNext(change); - if (!loaded) - continue; - try - { - OnUpdate?.Invoke(change.Previous, change.Document); - } - catch (Exception exception) - { - failures.Add(exception); - } - } - } - finally - { - lock (_deliveryGate) - { - _delivered++; - Monitor.PulseAll(_deliveryGate); - } - } - } - private CacheBackingStore? Validate( IReadOnlyList admitted, IReadOnlyList evicted, @@ -1776,7 +1716,7 @@ namespace GameCult.Caching Unindex(existing); _entries.Remove(stored.Key.Value); _handles.Remove(existing.Document); - changes.Add(new Change(CultCacheDocumentChangeKind.Removed, existing, null, existing.Document)); + changes.Add(new Change(CultCacheDocumentChangeKind.Removed, existing, null, existing.Document, ++_sequence)); } foreach (var stored in admitted) @@ -1795,7 +1735,8 @@ namespace GameCult.Caching previous == null ? CultCacheDocumentChangeKind.Added : CultCacheDocumentChangeKind.Updated, stored, stored.Document, - previous?.Document)); + previous?.Document, + ++_sequence)); } if (source == null && _stores.Count == 0) @@ -1895,18 +1836,20 @@ namespace GameCult.Caching private sealed class Change { - public Change(CultCacheDocumentChangeKind kind, CultStoredDocument stored, object? document, object? previous) + public Change(CultCacheDocumentChangeKind kind, CultStoredDocument stored, object? document, object? previous, long sequence) { Kind = kind; Stored = stored; Document = document; Previous = previous; + Sequence = sequence; } public CultCacheDocumentChangeKind Kind { get; } public CultStoredDocument Stored { get; } public object? Document { get; } public object? Previous { get; } + public long Sequence { get; } } } diff --git a/src/GameCult.Caching/CultManagedDocument.cs b/src/GameCult.Caching/CultManagedDocument.cs index fcfeaee8..00a6b2c2 100644 --- a/src/GameCult.Caching/CultManagedDocument.cs +++ b/src/GameCult.Caching/CultManagedDocument.cs @@ -22,14 +22,19 @@ namespace GameCult.Caching CultCacheDocumentChangeKind kind, CultRecordKey key, T? document, - T? previousDocument) + T? previousDocument, + long sequence = 0) { Kind = kind; Key = key; Document = document; PreviousDocument = previousDocument; + Sequence = sequence; } + // Per-cache, in memory, assigned at admission. A consumer keeping a latest value ignores a lower one. + public long Sequence { get; } + public CultCacheDocumentChangeKind Kind { get; } public CultRecordKey Key { get; } diff --git a/src/GameCult.Mesh/CultMesh.cs b/src/GameCult.Mesh/CultMesh.cs index 08868ade..6f89966d 100644 --- a/src/GameCult.Mesh/CultMesh.cs +++ b/src/GameCult.Mesh/CultMesh.cs @@ -1520,19 +1520,18 @@ namespace GameCult.Mesh var sourceList = sources?.ToArray() ?? new[] { ProjectionSource(key.Value, descriptor.SchemaId, "CultCache record") }; var route = routeHint ?? new CultMeshRouteHint(CultMeshLocalityKind.InProcess, "CultCache document"); - return Document( + var feed = LiveFeed( ResolveDocumentId(documentId, key), - context, - _ => Task.FromResult(ReadRequired(cache, key)), - _ => cache.WatchRecord(key) + (_parameters, _context) => Task.FromResult(ReadRequired(cache, key)), + (_parameters, _context) => cache.WatchRecord(key) .Where(change => change.Document != null) .Select(change => change.Document!), - async value => - { - await cache.UpsertAsync(value, new CultRecordHandle(key)).ConfigureAwait(false); - }, sourceList, route); + return new CultMeshDocumentHandle( + BindLiveFeed(context, feed), + async value => await cache.UpsertAsync(value, new CultRecordHandle(key)).ConfigureAwait(false), + () => cache.WatchRecord(key)); } /// diff --git a/src/GameCult.Mesh/CultMeshPrimitives.cs b/src/GameCult.Mesh/CultMeshPrimitives.cs index a034e6e6..98ee230d 100644 --- a/src/GameCult.Mesh/CultMeshPrimitives.cs +++ b/src/GameCult.Mesh/CultMeshPrimitives.cs @@ -927,6 +927,8 @@ namespace GameCult.Mesh private readonly CultMeshBoundLiveFeed _feed; private readonly Func? _replace; private readonly Func? _submitPrediction; + // Set only over a CultCache record, whose changes carry the cache's Sequence. + private readonly Func>>? _recordChanges; /// Creates a document handle from a Verse-bound live feed. public CultMeshDocumentHandle( @@ -939,6 +941,15 @@ namespace GameCult.Mesh _submitPrediction = submitPrediction; } + internal CultMeshDocumentHandle( + CultMeshBoundLiveFeed feed, + Func replace, + Func>> recordChanges) + : this(feed, replace) + { + _recordChanges = recordChanges; + } + /// Gets the semantic document id. public string DocumentId => _feed.FeedId; @@ -997,6 +1008,13 @@ namespace GameCult.Mesh return Watch().Subscribe(onNext); } + // Latest-value mirrors: a cache-backed handle passes each snapshot's Sequence; other sources pass null. + internal IDisposable WatchSequenced(Action onNext) => _recordChanges == null + ? Watch(document => onNext(document, null)) + : _recordChanges() + .Where(change => change.Document != null) + .Subscribe(change => onNext(change.Document!, change.Sequence)); + /// /// Projects one typed document field into a stable reactive state pointer. The expression is /// structural metadata: only member access and constant collection indices are accepted. @@ -1203,6 +1221,7 @@ namespace GameCult.Mesh private readonly object _gate = new(); private IDisposable? _subscription; private TDocument _current; + private long _appliedSequence; private bool _disposed; internal CultMeshObservedDocument( @@ -1234,7 +1253,7 @@ namespace GameCult.Mesh internal void Start() { - _subscription = Document.Watch(ApplyCanonicalSnapshot); + _subscription = Document.WatchSequenced(ApplyCanonicalSnapshot); } /// Reads and adopts a fresh canonical snapshot. @@ -1264,15 +1283,17 @@ namespace GameCult.Mesh _subscription?.Dispose(); } - private void ApplyCanonicalSnapshot(TDocument canonical) + private void ApplyCanonicalSnapshot(TDocument canonical, long? sequence) { if (canonical == null) return; var next = CultMeshDocumentHandle.CloneDocument(canonical); lock (_gate) { - if (!_disposed) - _current = next; + if (_disposed || sequence <= _appliedSequence) + return; + _appliedSequence = sequence ?? _appliedSequence; + _current = next; } } @@ -1439,6 +1460,7 @@ namespace GameCult.Mesh private bool _flushQueued; private bool _flushing; private bool _disposed; + private long _appliedSequence; private int _reconciliationVersion; internal CultMeshReactiveDocument( @@ -1489,7 +1511,7 @@ namespace GameCult.Mesh internal void Start() { - _subscription = _document.Watch(ApplyCanonicalSnapshot); + _subscription = _document.WatchSequenced(ApplyCanonicalSnapshot); } /// Mutates the current value and schedules a coalesced prediction or replacement. @@ -1596,15 +1618,16 @@ namespace GameCult.Mesh _subscription?.Dispose(); } - private void ApplyCanonicalSnapshot(TDocument canonical) + private void ApplyCanonicalSnapshot(TDocument canonical, long? sequence) { if (canonical == null) return; lock (_gate) { - if (_disposed) + if (_disposed || sequence <= _appliedSequence) return; + _appliedSequence = sequence ?? _appliedSequence; TDocument? replacement = null; if (_dirty || _flushing) diff --git a/tests/GameCult.Caching.Tests/ConditionalCommitTests.cs b/tests/GameCult.Caching.Tests/ConditionalCommitTests.cs index 0e668450..5c36c713 100644 --- a/tests/GameCult.Caching.Tests/ConditionalCommitTests.cs +++ b/tests/GameCult.Caching.Tests/ConditionalCommitTests.cs @@ -757,65 +757,60 @@ namespace GameCult.Caching.Tests batch.Upsert(new Counter { Name = "counter", Value = value }, new CultRecordHandle(Key)); }); - // Delivered in save order: each change replaces the document the previously delivered change installed. + // One writer's delivery is parked in an observer; other writers, whose observers write the same cache, finish. + // The cache is not disposed on failure because a stuck delivery would hang disposal too. [Test] - public void ConcurrentWritersDeliverInSaveOrder() + public void ObserverWritingTheSameCacheDoesNotWaitOnOtherDeliveries() + { + var cache = new CultCache(Registry); + using var release = new ManualResetEventSlim(); + using var parked = new ManualResetEventSlim(); + using var park = cache.Watch().Subscribe(change => + { + if (change.Key.Value != "parked") return; + parked.Set(); + release.Wait(TimeSpan.FromSeconds(30)); + }); + using var tallies = cache.Watch().Subscribe(change => + cache.UpsertAsync(new Tally { Name = $"t:{change.Key.Value}" }, new CultRecordHandle(new CultRecordKey($"t:{change.Key.Value}")))); + + var parkedWriter = Task.Run(() => cache.UpsertAsync(new Counter { Name = "parked" }, new CultRecordHandle(new CultRecordKey("parked")))); + Assert.That(parked.Wait(TimeSpan.FromSeconds(5)), Is.True); + var writers = Enumerable.Range(0, 8).Select(thread => Task.Run(() => + { + for (var i = 0; i < 100; i++) + cache.UpsertAsync(new Counter { Name = $"c:{thread}", Value = i }, new CultRecordHandle(new CultRecordKey($"c:{thread}"))); + })).ToArray(); + var finished = Task.WaitAll(writers, TimeSpan.FromSeconds(10)); + release.Set(); + + Assert.That(finished, Is.True, "a writer waited on another thread's delivery"); + Assert.That(parkedWriter.Wait(TimeSpan.FromSeconds(10)), Is.True); + cache.Dispose(); + } + + // Ordered by Sequence, each change replaces the document installed by the change before it. + [Test] + public void SequenceIncreasesInAdmissionOrder() { using var cache = new CultCache(Registry); var key = new CultRecordKey("ordered"); - var observed = new System.Collections.Generic.List<(object? Document, object? Previous)>(); - using var subscription = cache.Watch().Subscribe(change => - { - lock (observed) - observed.Add((change.Document, change.PreviousDocument)); - }); + var observed = new ConcurrentBag>(); + using var subscription = cache.Watch().Subscribe(observed.Add); - var writers = Enumerable.Range(0, 8).Select(thread => Task.Run(() => + var writers = Enumerable.Range(0, 8).Select(_ => Task.Run(() => { for (var i = 0; i < 2000; i++) cache.UpsertAsync(new Counter { Name = "ordered", Value = i }, new CultRecordHandle(key)); })).ToArray(); Assert.That(Task.WaitAll(writers, TimeSpan.FromSeconds(60)), Is.True); - Assert.That(observed, Has.Count.EqualTo(16000)); - Assert.That(FirstOutOfOrder(observed), Is.EqualTo(-1), "a change was delivered out of save order"); - } - - // The cache is not disposed on failure because a deadlocked delivery would hang the test thread too. - [Test] - public void ObserverWritingTheSameCacheDuringConcurrentDeliveryDoesNotDeadlock() - { - var cache = new CultCache(Registry); - var tally = new CultRecordKey("tally"); - var observed = new System.Collections.Generic.List<(object? Document, object? Previous)>(); - using var counters = cache.Watch().Subscribe(_ => - cache.UpsertAsync(new Tally { Name = "tally" }, new CultRecordHandle(tally))); - using var tallies = cache.Watch().Subscribe(change => - { - lock (observed) - observed.Add((change.Document, change.PreviousDocument)); - }); - - var writers = Enumerable.Range(0, 8).Select(thread => Task.Run(() => - { - for (var i = 0; i < 500; i++) - cache.UpsertAsync(new Counter { Name = $"c:{thread}", Value = i }, new CultRecordHandle(new CultRecordKey($"c:{thread}"))); - })).ToArray(); - - Assert.That(Task.WaitAll(writers, TimeSpan.FromSeconds(60)), Is.True, "an observer writing the cache during delivery deadlocked"); - Assert.That(observed, Has.Count.EqualTo(4000)); - Assert.That(FirstOutOfOrder(observed), Is.EqualTo(-1), "an observer's write was delivered out of save order"); - cache.Dispose(); - } - - private static int FirstOutOfOrder(System.Collections.Generic.List<(object? Document, object? Previous)> observed) - { - for (var i = 1; i < observed.Count; i++) - { - if (!ReferenceEquals(observed[i].Previous, observed[i - 1].Document)) - return i; - } - return -1; + var ordered = observed.OrderBy(change => change.Sequence).ToArray(); + Assert.That(ordered, Has.Length.EqualTo(16000)); + Assert.That(ordered.Select(change => change.Sequence).Distinct().Count(), Is.EqualTo(16000), "a Sequence repeated"); + var broken = Enumerable.Range(1, ordered.Length - 1) + .FirstOrDefault(i => !ReferenceEquals(ordered[i].PreviousDocument, ordered[i - 1].Document)); + Assert.That(broken, Is.EqualTo(0), "a change's previous document is not the one installed by the next-lower Sequence"); } private string PathOf(string fileName) => System.IO.Path.Combine(_directory, fileName); diff --git a/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs b/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs index cd80f6e5..1dc938c9 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs +++ b/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs @@ -1387,6 +1387,38 @@ public sealed class CultMeshStreamingTests observed.Current.Revision.Should().Be(2); } + // The older of two racing writes is parked in an earlier observer until the newer one has been applied. + [Test] + public async Task WatchRecordIgnoresStaleSequence() + { + var cache = new CultCache(); + var key = new CultRecordKey("mesh-note:stale-sequence"); + var handle = new CultRecordHandle(key); + MeshNoteDocument Note(string text, int revision) => + new() { Schema = "tests.mesh_note.v1", Text = text, Revision = revision }; + await cache.UpsertAsync(Note("initial", 1), handle); + using var release = new System.Threading.ManualResetEventSlim(); + using var parked = new System.Threading.ManualResetEventSlim(); + using var park = cache.WatchRecord(key).Subscribe(change => + { + if (change.Document?.Revision != 2) return; + parked.Set(); + release.Wait(TimeSpan.FromSeconds(30)); + }); + using var observed = await CultMesh.Document(cache, key).ObserveAsync(); + + var older = Task.Run(() => cache.UpsertAsync(Note("older", 2), handle)); + parked.Wait(TimeSpan.FromSeconds(5)).Should().BeTrue(); + var newer = Task.Run(() => cache.UpsertAsync(Note("newer", 3), handle)); + var newerFinished = newer.Wait(TimeSpan.FromSeconds(10)); + release.Set(); + + newerFinished.Should().BeTrue("a writer waited on another thread's delivery"); + older.Wait(TimeSpan.FromSeconds(10)).Should().BeTrue(); + observed.Current.Text.Should().Be("newer"); + cache.Dispose(); + } + [Test] public async Task ReactiveDocument_DisposeSuppressesScheduledFlush() { diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs index ae37be98..6e915984 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -1820,6 +1820,78 @@ namespace GameCult.Networking.Tests Assert.That(removed.Active, Is.False); } + // A load's delivery reaches a subscription observer that needs the lifecycle gate while a subscribe's demand + // handler, holding that gate, writes the same cache. Nothing is disposed on failure: disposal would hang too. + [Test] + public async Task SubscriptionDemandHandlerWritingCacheDuringDeliveryDoesNotDeadlock() + { + var path = Path.Combine(Path.GetTempPath(), $"cultlib-demand-{Guid.NewGuid():N}.cc"); + using (var seed = new CultCache()) + { + seed.AddBackingStore(new SingleFileMessagePackBackingStore(path)); + await seed.UpsertAsync(new NetworkSchemaNote { Schema = "tests.networking_note.v1", Text = "loaded" }, + new CultRecordHandle(new CultRecordKey("demand:loaded"))); + seed.FlushAllBackingStores(); + } + + var cache = new CultCache(); + var delivering = new ManualResetEventSlim(); + cache.OnUpdate += (_, _) => delivering.Set(); + var database = new CultNetDatabase(cache); + var server = new RudpCultNetSchemaServer(new RudpCultNetSchemaServerOptions + { + RuntimeId = "demand-write-server", + Socket = BindUdpSocket() + }); + var subscriptions = new CultNetDatabaseSubscriptionServer(server, database); + var written = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + subscriptions.DemandChanged += demand => + { + if (!demand.Active) return; + _ = Task.Run(() => cache.AddBackingStore(new SingleFileMessagePackBackingStore(path))); + if (!delivering.Wait(TimeSpan.FromSeconds(5))) + { + written.TrySetException(new TimeoutException("the load never started delivering")); + return; + } + cache.UpsertAsync(new NetworkSchemaNote { Schema = "tests.networking_note.v1", Text = "demanded" }, + new CultRecordHandle(new CultRecordKey("demand:written"))).GetAwaiter().GetResult(); + written.TrySetResult(true); + }; + using var cancellation = new CancellationTokenSource(); + var serverThread = new Thread(() => + { + while (!cancellation.IsCancellationRequested) + { + _ = server.PollOnceAsync().GetAwaiter().GetResult(); + Thread.Sleep(1); + } + }) { IsBackground = true }; + serverThread.Start(); + + using var client = CultNetSchemaClients.CreateRudp("demand-write-client"); + client.Connect("127.0.0.1", server.LocalEndPoint.Port); + await WaitUntilAsync(() => client.Connected, TimeSpan.FromSeconds(2)); + client.SendCultNet(new CultNetDatabaseSubscribeMessage + { + MessageId = "subscribe-demand-write", + SubscriptionId = "demand-write", + RecordKeys = ["demand:other"], + IncludeSnapshot = false + }); + + var finished = await Task.WhenAny(written.Task, Task.Delay(TimeSpan.FromSeconds(10))) == written.Task; + cancellation.Cancel(); + Assert.That(finished, Is.True, "the demand handler's cache write deadlocked with the load's delivery"); + await written.Task; + serverThread.Join(); + subscriptions.Dispose(); + server.Dispose(); + database.Dispose(); + cache.Dispose(); + File.Delete(path); + } + [Test] public async Task DatabaseSubscriptionServer_ClassifiesLoopbackTcpDemandAsSameMachine() {