diff --git a/src/GameCult.Caching/Contracts/cultcache-store-composition.md b/src/GameCult.Caching/Contracts/cultcache-store-composition.md index ab028fca..6cf6cdfa 100644 --- a/src/GameCult.Caching/Contracts/cultcache-store-composition.md +++ b/src/GameCult.Caching/Contracts/cultcache-store-composition.md @@ -77,14 +77,21 @@ which route wrote it. - 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 - are dropped. A `Watch` subscriber exception is not rethrown; it follows R3's - unhandled-exception handling. A throwing handler cannot undo the store's - adoption of a load; the store and cache already agree when publication starts. + are dropped. Every `Watch` subscriber receives every change: R3 routes a + subscriber's exception to its unhandled-exception handler, and if that handler + rethrows (fail-fast), the exception is treated as an `OnUpdate` handler + exception. The subscriber stays subscribed and the commit stands. A throwing + handler cannot undo the store's adoption of a load; the store and cache + already agree when publication starts. +- A record's `[CultName]` and `[CultIndex]` getters run once per landing record, + before the store is written and before any `Sequence` is minted. A getter that + throws refuses the whole change set, loads included: nothing is written, + minted or left in memory. - Pulling all stores pulls every attached store even if a handler throws during one store's load, then rethrows. - Hydration failure on open is loud: a corrupt store file, or a record whose - schema the registry cannot resolve, makes the open throw and leaves the file - byte-identical. Consumers never delete and rewrite a store they failed to + schema the registry cannot resolve or whose name or index getter throws, makes + the open throw and leaves the file byte-identical. Consumers never delete and rewrite a store they failed to open. ## Locking diff --git a/src/GameCult.Caching/CultCache.cs b/src/GameCult.Caching/CultCache.cs index 8af4360c..2a9de06e 100644 --- a/src/GameCult.Caching/CultCache.cs +++ b/src/GameCult.Caching/CultCache.cs @@ -1806,7 +1806,11 @@ namespace GameCult.Caching // Keys holding a variant: derived from the entries by Apply, never decided anywhere else. private readonly HashSet _variantKeys = new(StringComparer.Ordinal); private readonly ConditionalWeakTable _handles = new(); - private readonly Subject _changes = new(); + // The cache's own observers, copy-on-write under _observerGate: publication reads one snapshot per change and + // catches per observer, so no observer decides whether another receives a change. + private Observer[] _observers = Array.Empty>(); + private bool _observersCompleted; + private readonly object _observerGate = new(); private readonly object _gate = new(); private long _sequence; // The changes admitted by this thread's outermost hold, published by that hold when it exits. @@ -1886,7 +1890,7 @@ namespace GameCult.Caching public Observable> Watch() where T : class { - return _changes + return Observable.Create(Register) .Where(change => typeof(T).IsAssignableFrom(change.Stored.Descriptor.DocumentType)) .Select(change => new CultCacheDocumentChange( change.Kind, @@ -1896,6 +1900,25 @@ namespace GameCult.Caching change.Sequence)); } + private IDisposable Register(Observer observer) + { + lock (_observerGate) + { + if (!_observersCompleted) + { + _observers = _observers.Append(observer).ToArray(); + return Disposable.Create(() => + { + lock (_observerGate) + _observers = _observers.Where(held => !ReferenceEquals(held, observer)).ToArray(); + }); + } + } + + observer.OnCompleted(); + return Disposable.Empty; + } + public Observable> WatchRecord(CultRecordKey key) where T : class { return Watch().Where(change => change.Key.Equals(key)); @@ -2151,7 +2174,31 @@ namespace GameCult.Caching foreach (var store in BackingStores) store.Dispose(); - _changes.Dispose(); + Observer[] completing; + lock (_observerGate) + { + completing = _observers; + _observers = Array.Empty>(); + _observersCompleted = true; + } + + var failures = new List(); + foreach (var observer in completing) + { + try + { + observer.OnCompleted(); + } + catch (Exception exception) + { + failures.Add(exception); + } + } + + if (failures.Count == 1) + ExceptionDispatchInfo.Capture(failures[0]).Throw(); + if (failures.Count > 1) + throw new AggregateException(failures); } internal static void RequireInstanceOf(Type documentType, object document) @@ -2333,8 +2380,8 @@ namespace GameCult.Caching // 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. + // guaranteed; each change carries the Sequence it was admitted with. Every change reaches every observer even + // if an observer or an OnUpdate handler throws; then the first exception (an AggregateException for several) is rethrown. // If the body threw, its exception wins. internal T Held(Func body) { @@ -2371,8 +2418,20 @@ namespace GameCult.Caching var failures = new List(); foreach (var (change, loaded) in changes) { - // R3 routes a throwing subscriber to its unhandled-exception handler; OnNext does not throw. - _changes.OnNext(change); + // R3 routes a throwing subscriber to its unhandled-exception handler, and a fail-fast handler rethrows: every + // observer still runs, the exception joins the failures, and the observer stays subscribed. + foreach (var observer in Volatile.Read(ref _observers)) + { + try + { + observer.OnNext(change); + } + catch (Exception exception) + { + failures.Add(exception); + } + } + if (!loaded) continue; try diff --git a/tests/GameCult.Caching.Tests/CultCacheObserverTests.cs b/tests/GameCult.Caching.Tests/CultCacheObserverTests.cs new file mode 100644 index 00000000..8cb3992d --- /dev/null +++ b/tests/GameCult.Caching.Tests/CultCacheObserverTests.cs @@ -0,0 +1,139 @@ +#nullable enable +using System; +using System.Collections.Generic; +using System.Linq; +using GameCult.Caching; +using GameCult.Caching.MessagePack; +using MessagePack; +using NUnit.Framework; +using R3; + +namespace GameCult.Caching.Tests +{ + // Cut 2 of docs/cultcache-publication-cut.md: the cache keeps its own observers and catches per observer, so no + // observer decides whether another receives a change. R3 routes a subscriber's exception to the unhandled handler; + // these tests install a fail-fast (rethrowing) handler, the only way an exception reaches the cache. + [NonParallelizable] + public class CultCacheObserverTests + { + private static readonly CultDocumentRegistry Registry = CultDocumentRegistry.ForTypes(new[] { typeof(ObservedPing) }); + + private Action _previousHandler = _ => { }; + + [SetUp] + public void FailFast() + { + _previousHandler = ObservableSystem.GetUnhandledExceptionHandler(); + ObservableSystem.RegisterUnhandledExceptionHandler(exception => throw exception); + } + + [TearDown] + public void Restore() => ObservableSystem.RegisterUnhandledExceptionHandler(_previousHandler); + + [CultDocument("tests.observed_ping", "tests.observed_ping.v1")] + [MessagePackObject(AllowPrivate = true)] + internal sealed class ObservedPing + { + [Key(0)] + public string Text = string.Empty; + } + + private static CultRecordKey KeyOf(string text) => new($"observed:{text}"); + + private static void Send(CultCache cache, params string[] texts) => + cache.Commit(batch => + { + foreach (var text in texts) + batch.Upsert(typeof(ObservedPing), new ObservedPing { Text = text }, KeyOf(text)); + }); + + [Test] + public void AnEscapedObserverExceptionReachesTheWriterAfterEveryObserverSawEveryChange() + { + using var cache = new CultCache(Registry); + var thrower = new List(); + var later = new List(); + using var first = cache.Watch().Subscribe(change => + { + thrower.Add(change.Document!.Text); + if (change.Document.Text == "boom") + throw new InvalidOperationException("observer boom"); + }); + using var second = cache.Watch().Subscribe(change => later.Add(change.Document!.Text)); + + var escaped = Assert.Throws(() => Send(cache, "boom", "after"))!; + + Assert.Multiple(() => + { + Assert.That(escaped.Message, Is.EqualTo("observer boom")); + Assert.That(later, Is.EqualTo(new[] { "boom", "after" }), "the observer after the thrower saw every change of the hold"); + Assert.That(thrower, Is.EqualTo(new[] { "boom", "after" }), "the thrower kept receiving after its own exception"); + Assert.That(cache.Get(KeyOf("boom")), Is.Not.Null, "the commit stands"); + Assert.That(cache.Get(KeyOf("after")), Is.Not.Null); + }); + + Send(cache, "later"); + Assert.That(thrower, Is.EqualTo(new[] { "boom", "after", "later" }), "the thrower stayed subscribed"); + Assert.That(later, Is.EqualTo(new[] { "boom", "after", "later" })); + } + + [Test] + public void SeveralEscapedObserverExceptionsArriveTogether() + { + using var cache = new CultCache(Registry); + using var one = cache.Watch().Subscribe(_ => throw new InvalidOperationException("one")); + using var two = cache.Watch().Subscribe(_ => throw new InvalidOperationException("two")); + + var escaped = Assert.Throws(() => Send(cache, "boom"))!; + + Assert.That(escaped.InnerExceptions.Select(exception => exception.Message), Is.EqualTo(new[] { "one", "two" })); + } + + [Test] + public void ObserversMayJoinAndLeaveDuringADeliveryWithoutBreakingTheOthers() + { + using var cache = new CultCache(Registry); + var a = new List(); + var b = new List(); + var c = new List(); + var d = new List(); + IDisposable? leaving = null; + IDisposable? joined = null; + using var first = cache.Watch().Subscribe(change => + { + a.Add(change.Sequence); + if (a.Count != 1) + return; + leaving!.Dispose(); + joined = cache.Watch().Subscribe(next => c.Add(next.Sequence)); + }); + leaving = cache.Watch().Subscribe(change => b.Add(change.Sequence)); + using var last = cache.Watch().Subscribe(change => d.Add(change.Sequence)); + + Send(cache, "one", "two"); + joined?.Dispose(); + + Assert.Multiple(() => + { + Assert.That(a, Is.EqualTo(new[] { 1L, 2L })); + Assert.That(d, Is.EqualTo(new[] { 1L, 2L }), "an observer that did not move still saw everything"); + Assert.That(b, Does.Not.Contain(2L), "a disposed observer stops receiving"); + Assert.That(c, Is.EqualTo(new[] { 2L }), "an observer that joined mid-hold receives from the next change"); + }); + } + + [Test] + public void DisposeCompletesObserversAndACompletedCacheCompletesNewOnes() + { + var cache = new CultCache(Registry); + var completed = 0; + using var before = cache.Watch().Subscribe(_ => { }, _ => completed++); + + cache.Dispose(); + Assert.That(completed, Is.EqualTo(1)); + + using var after = cache.Watch().Subscribe(_ => { }, _ => completed++); + Assert.That(completed, Is.EqualTo(2), "subscribing to a disposed cache completes at once"); + } + } +} diff --git a/tests/GameCult.Networking.Tests/CultCachePublicationIsolationTests.cs b/tests/GameCult.Networking.Tests/CultCachePublicationIsolationTests.cs index 9ba78efa..33d1100d 100644 --- a/tests/GameCult.Networking.Tests/CultCachePublicationIsolationTests.cs +++ b/tests/GameCult.Networking.Tests/CultCachePublicationIsolationTests.cs @@ -59,5 +59,40 @@ namespace GameCult.Networking.Tests Assert.That(database.LastWriteSequence(SchemaId, after), Is.Not.Null, "a later commit was never logged"); }); } + + // One observer's escaped exception (a fail-fast R3 handler) does not stop the database observer after it, and the writer gets it. + [Test] + public async Task AnEscapedObserverExceptionDoesNotStopTheDatabaseOrLaterChanges() + { + var previous = ObservableSystem.GetUnhandledExceptionHandler(); + ObservableSystem.RegisterUnhandledExceptionHandler(exception => throw exception); + try + { + var cache = new CultCache(); + using var thrower = cache.Watch().Subscribe(change => + { + if (change.Document?.Text == "boom") throw new InvalidOperationException("observer boom"); + }); + var database = new CultNetDatabase(cache); + var seen = new ConcurrentQueue(); + using var watch = database.Watch().Subscribe(change => seen.Enqueue(change.Document?.Text)); + var boom = new CultRecordKey("pub:boom"); + var escaped = Assert.ThrowsAsync(() => cache.UpsertAsync(Note("boom"), new CultRecordHandle(boom))); + + ObservableSystem.RegisterUnhandledExceptionHandler(previous); + var after = new CultRecordKey("pub:after"); + await database.PutAsync(after, Note("after")); + Assert.Multiple(() => + { + Assert.That(escaped!.Message, Is.EqualTo("observer boom")); + Assert.That(seen, Is.EqualTo(new[] { "boom", "after" }), "the database observer after the thrower saw both changes"); + Assert.That(database.LastWriteSequence(SchemaId, after), Is.Not.Null, "a later commit was never logged"); + }); + } + finally + { + ObservableSystem.RegisterUnhandledExceptionHandler(previous); + } + } } }