GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

Carry order as data: per-cache change Sequence, no delivery scheduler

Browse source 
bcae483 made the cache a delivery scheduler (tickets, _deliveryGate turn
waits, per-thread delivery queues). It deadlocked inside CultLib: a load's
delivery reaching a CultNetDatabaseSubscriptionServer observer that needs
_lifecycleGate, while a subscribe's demand handler holding that gate writes
the cache and waits for the delivery's turn.

The cache now owns order as data. Every admitted change takes the next value
of a per-cache in-memory sequence under the gate, published as
CultCacheDocumentChange<T>.Sequence. Delivery is the 544c087/b653f68 rule:
each call publishes its own changes on its own thread after releasing the
gate; nothing waits on another call's delivery. Cross-thread delivery order is
not guaranteed.

Deleted: ticket counter, _delivered, _deliveryGate, Monitor.Wait turns,
per-thread delivery queues, per-ticket catch-all, and the save-order contract
text with its observer-must-not-block rule.

Consumer: a cache-backed CultMesh document handle hands its observed and
reactive mirrors each snapshot's Sequence; ApplyCanonicalSnapshot drops one
not newer than the last applied.

Tests: SequenceIncreasesInAdmissionOrder,
ObserverWritingTheSameCacheDoesNotWaitOnOtherDeliveries,
SubscriptionDemandHandlerWritingCacheDuringDeliveryDoesNotDeadlock,
WatchRecordIgnoresStaleSequence. The last three time out on bcae483.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-13 20:43:11 +02:00
commit 456234018f
8 changed files with 233 additions and 166 deletions
Download patch file Download diff file Expand all files Collapse all files

26
src/GameCult.Caching/Contracts/cultcache-store-composition.md
View file

@ -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

131
src/GameCult.Caching/CultCache.cs
View file

@ -1180,13 +1180,9 @@ namespace GameCult.Caching
private readonly ConditionalWeakTable<object, KeyBox> _handles = new();
private readonly Subject<Change> _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<CultCacheDocumentChange<T>> WatchRecord<T>(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<T>(Func<T> 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<Exception>() : 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<Exception> Publish(long ticket, List<(Change Change, bool Loaded)> changes)
private List<Exception> Publish(List<(Change Change, bool Loaded)> changes)
{
var failures = new List<Exception>();
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<Exception> 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<CultStoredDocument> admitted,
IReadOnlyList<CultStoredDocument> 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; }
}
}

7
src/GameCult.Caching/CultManagedDocument.cs
View file

@ -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; }

15
src/GameCult.Mesh/CultMesh.cs
View file

@ -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<TDocument>(
var feed = LiveFeed<CultMeshDocumentQueryParameters, TDocument>(
ResolveDocumentId(documentId, key),
context,
_ => Task.FromResult(ReadRequired<TDocument>(cache, key)),
_ => cache.WatchRecord<TDocument>(key)
(_parameters, _context) => Task.FromResult(ReadRequired<TDocument>(cache, key)),
(_parameters, _context) => cache.WatchRecord<TDocument>(key)
.Where(change => change.Document != null)
.Select(change => change.Document!),
async value =>
{
await cache.UpsertAsync(value, new CultRecordHandle<TDocument>(key)).ConfigureAwait(false);
},
sourceList,
route);
return new CultMeshDocumentHandle<TDocument>(
BindLiveFeed(context, feed),
async value => await cache.UpsertAsync(value, new CultRecordHandle<TDocument>(key)).ConfigureAwait(false),
() => cache.WatchRecord<TDocument>(key));
}
/// <summary>

37
src/GameCult.Mesh/CultMeshPrimitives.cs
View file

@ -927,6 +927,8 @@ namespace GameCult.Mesh
private readonly CultMeshBoundLiveFeed<CultMeshDocumentQueryParameters, TDocument> _feed;
private readonly Func<TDocument, Task>? _replace;
private readonly Func<TDocument, Task>? _submitPrediction;
// Set only over a CultCache record, whose changes carry the cache's Sequence.
private readonly Func<Observable<CultCacheDocumentChange<TDocument>>>? _recordChanges;
/// <summary>Creates a document handle from a Verse-bound live feed.</summary>
public CultMeshDocumentHandle(
@ -939,6 +941,15 @@ namespace GameCult.Mesh
_submitPrediction = submitPrediction;
}
internal CultMeshDocumentHandle(
CultMeshBoundLiveFeed<CultMeshDocumentQueryParameters, TDocument> feed,
Func<TDocument, Task> replace,
Func<Observable<CultCacheDocumentChange<TDocument>>> recordChanges)
: this(feed, replace)
{
_recordChanges = recordChanges;
}
/// <summary>Gets the semantic document id.</summary>
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<TDocument, long?> onNext) => _recordChanges == null
? Watch(document => onNext(document, null))
: _recordChanges()
.Where(change => change.Document != null)
.Subscribe(change => onNext(change.Document!, change.Sequence));
/// <summary>
/// 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);
}
/// <summary>Reads and adopts a fresh canonical snapshot.</summary>
@ -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<TDocument>.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);
}
/// <summary>Mutates the current value and schedules a coalesced prediction or replacement.</summary>
@ -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)

91
tests/GameCult.Caching.Tests/ConditionalCommitTests.cs
View file

@ -757,65 +757,60 @@ namespace GameCult.Caching.Tests
batch.Upsert(new Counter { Name = "counter", Value = value }, new CultRecordHandle<Counter>(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<Counter>().Subscribe(change =>
{
if (change.Key.Value != "parked") return;
parked.Set();
release.Wait(TimeSpan.FromSeconds(30));
});
using var tallies = cache.Watch<Counter>().Subscribe(change =>
cache.UpsertAsync(new Tally { Name = $"t:{change.Key.Value}" }, new CultRecordHandle<Tally>(new CultRecordKey($"t:{change.Key.Value}"))));
var parkedWriter = Task.Run(() => cache.UpsertAsync(new Counter { Name = "parked" }, new CultRecordHandle<Counter>(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<Counter>(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<Counter>().Subscribe(change =>
{
lock (observed)
observed.Add((change.Document, change.PreviousDocument));
});
var observed = new ConcurrentBag<CultCacheDocumentChange<Counter>>();
using var subscription = cache.Watch<Counter>().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<Counter>(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<Counter>().Subscribe(_ =>
cache.UpsertAsync(new Tally { Name = "tally" }, new CultRecordHandle<Tally>(tally)));
using var tallies = cache.Watch<Tally>().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<Counter>(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);

32
tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs
View file

@ -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<MeshNoteDocument>(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<MeshNoteDocument>(key).Subscribe(change =>
{
if (change.Document?.Revision != 2) return;
parked.Set();
release.Wait(TimeSpan.FromSeconds(30));
});
using var observed = await CultMesh.Document<MeshNoteDocument>(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()
{

72
tests/GameCult.Networking.Tests/NetworkingTests.cs
View file

@ -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<NetworkSchemaNote>(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<bool>(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<NetworkSchemaNote>(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()
{