CultCache: own the observer list and isolate every observer

The cache no longer fans changes out through an R3 Subject. It keeps a
copy-on-write observer array and Publish catches per observer: every observer
receives every change, an exception that escapes (a fail-fast R3 handler) joins
the OnUpdate failures rethrown to the writer after delivery, and the observer
stays subscribed. Dispose completes observers, and a disposed cache completes
new ones at once. Deletes the false "OnNext does not throw" comment.

The composition contract now states the observer policy and that a throwing
name or index getter refuses a change set, loads included.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

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

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<string> _variantKeys = new(StringComparer.Ordinal);
         private readonly ConditionalWeakTable<object, KeyBox> _handles = new();
-        private readonly Subject<Change> _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<Change>[] _observers = Array.Empty<Observer<Change>>();
+        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<CultCacheDocumentChange<T>> Watch<T>() where T : class
         {
-            return _changes
+            return Observable.Create<Change>(Register)
                 .Where(change => typeof(T).IsAssignableFrom(change.Stored.Descriptor.DocumentType))
                 .Select(change => new CultCacheDocumentChange<T>(
                     change.Kind,
@@ -1896,6 +1900,25 @@ namespace GameCult.Caching
                     change.Sequence));
         }
 
+        private IDisposable Register(Observer<Change> 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<CultCacheDocumentChange<T>> WatchRecord<T>(CultRecordKey key) where T : class
         {
             return Watch<T>().Where(change => change.Key.Equals(key));
@@ -2151,7 +2174,31 @@ namespace GameCult.Caching
             foreach (var store in BackingStores)
                 store.Dispose();
 
-            _changes.Dispose();
+            Observer<Change>[] completing;
+            lock (_observerGate)
+            {
+                completing = _observers;
+                _observers = Array.Empty<Observer<Change>>();
+                _observersCompleted = true;
+            }
+
+            var failures = new List<Exception>();
+            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<T>(Func<T> body)
         {
@@ -2371,8 +2418,20 @@ namespace GameCult.Caching
             var failures = new List<Exception>();
             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

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<Exception> _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<string>();
+            var later = new List<string>();
+            using var first = cache.Watch<ObservedPing>().Subscribe(change =>
+            {
+                thrower.Add(change.Document!.Text);
+                if (change.Document.Text == "boom")
+                    throw new InvalidOperationException("observer boom");
+            });
+            using var second = cache.Watch<ObservedPing>().Subscribe(change => later.Add(change.Document!.Text));
+
+            var escaped = Assert.Throws<InvalidOperationException>(() => 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<ObservedPing>(KeyOf("boom")), Is.Not.Null, "the commit stands");
+                Assert.That(cache.Get<ObservedPing>(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<ObservedPing>().Subscribe(_ => throw new InvalidOperationException("one"));
+            using var two = cache.Watch<ObservedPing>().Subscribe(_ => throw new InvalidOperationException("two"));
+
+            var escaped = Assert.Throws<AggregateException>(() => 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<long>();
+            var b = new List<long>();
+            var c = new List<long>();
+            var d = new List<long>();
+            IDisposable? leaving = null;
+            IDisposable? joined = null;
+            using var first = cache.Watch<ObservedPing>().Subscribe(change =>
+            {
+                a.Add(change.Sequence);
+                if (a.Count != 1)
+                    return;
+                leaving!.Dispose();
+                joined = cache.Watch<ObservedPing>().Subscribe(next => c.Add(next.Sequence));
+            });
+            leaving = cache.Watch<ObservedPing>().Subscribe(change => b.Add(change.Sequence));
+            using var last = cache.Watch<ObservedPing>().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<ObservedPing>().Subscribe(_ => { }, _ => completed++);
+
+            cache.Dispose();
+            Assert.That(completed, Is.EqualTo(1));
+
+            using var after = cache.Watch<ObservedPing>().Subscribe(_ => { }, _ => completed++);
+            Assert.That(completed, Is.EqualTo(2), "subscribing to a disposed cache completes at once");
+        }
+    }
+}

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<NetworkSchemaNote>().Subscribe(change =>
+                {
+                    if (change.Document?.Text == "boom") throw new InvalidOperationException("observer boom");
+                });
+                var database = new CultNetDatabase(cache);
+                var seen = new ConcurrentQueue<string?>();
+                using var watch = database.Watch<NetworkSchemaNote>().Subscribe(change => seen.Enqueue(change.Document?.Text));
+                var boom = new CultRecordKey("pub:boom");
+                var escaped = Assert.ThrowsAsync<InvalidOperationException>(() => cache.UpsertAsync(Note("boom"), new CultRecordHandle<NetworkSchemaNote>(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);
+            }
+        }
     }
 }