Merge CultCache publication Cut 3: the journal

AddJournal runs under the gate right after Apply, once per admission that
changed something, before observers. Soul: merge with follow-ups; the
dispose and catch-path fixes landed. Cut 4 must decide what a throwing
journal means for replicas and whether a journal needs the change origin.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

src/GameCult.Caching/Contracts/cultcache-store-composition.md

@@ -74,6 +74,22 @@ which route wrote it.
 - Not stale-protected: schema-alias handles (`AsSchemaAlias`) get no
   `Sequence`; removals do not advance a mirror's applied `Sequence`, and the
   mirrors do not surface removals.
+- `AddJournal` registers an ordered admission journal, for audit, replication or
+  a derived index. The cache calls each journal under its gate, right after an
+  admission is applied and before that admission is published, once per
+  admission that changed something, with that admission's changes (a variant
+  re-resolved by a base edit included) in `Sequence` order. Journals of one cache
+  run in registration order, and reading the cache from one sees the new state.
+  Because the gate serializes admissions, the calls of one cache see contiguous,
+  strictly increasing `Sequence` values: a journal is the one place where cache
+  order is a fact, and delivery order is not. A refused admission, or one a store
+  declines, is not journaled. Disposing the registration removes the journal.
+- A journal that throws does not undo the admission: the remaining journals
+  still run, the change is published, and the exception is rethrown to the
+  writer after publication, with the same aggregation as an observer's. The
+  journal stays registered. Any hold on the cache while a journal runs throws
+  `InvalidOperationException`: a journal must not write to the cache or
+  register another journal.
 - 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
@@ -83,10 +99,15 @@ which route wrote it.
   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.
+- An admission runs a record's `[CultName]` and `[CultIndex]` getters once per
+  landing record, before the store is written and before any `Sequence` is
+  minted. A conditional commit on a clean file store is judged a second time
+  against the file it merges onto whenever the cache holds variants in any store,
+  the batch lands a variant, or the file holds one. That second read runs the
+  getters again for the records the batch lands, for any variant it re-resolves,
+  and for the records other writers put in the file (the arriving set); that
+  judgement is also before the store is written. 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
@@ -100,6 +121,11 @@ which route wrote it.
   cache takes that cache's gate as its own lock, so a store's load callback into
   the cache cannot take the two out of order. A store must not call `Loaded`
   while holding its own lock outside the cache's hold.
+- A journal runs under the cache's gate, so a slow journal slows every writer of
+  that cache. It must not enter the cache, and must not wait for anything that
+  needs the cache or its gate: another thread's write, a lock a writer holds.
+  Its own leaf locks, taken only inside the journal, are the only locks that may
+  nest under the gate.
 - A thread holds one cache's gate at a time: entering a cache's hold inside
   another cache's hold throws.
 - A store doing I/O on behalf of its cache (pull, flush, commit) holds the gate,

src/GameCult.Caching/CultCache.cs

@@ -1813,8 +1813,11 @@ namespace GameCult.Caching
         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.
-        [ThreadStatic] private static List<(Change Change, bool Loaded)>? _held;
+        // What this thread's outermost hold admitted, published by that hold when it exits.
+        [ThreadStatic] private static Hold? _held;
+        // Journals run under the gate, in registration order; both fields are read and written only under it.
+        private readonly List<Action<IReadOnlyList<CultCacheDocumentChange<object>>>> _journals = new();
+        private bool _journaling;
         private bool _dirtyInMemory;
 
         // The codec is the store's encoding (CultCacheMessagePack.Create supplies it). A cache without one holds no variants:
@@ -1900,6 +1903,29 @@ namespace GameCult.Caching
                     change.Sequence));
         }
 
+        // Disposing removes the journal. A dispose the cache refuses (from inside a journal, or inside another cache's hold)
+        // throws and leaves the registration live and disposable; it counts as disposed only once the removal succeeded.
+        private sealed class JournalRegistration : IDisposable
+        {
+            private readonly CultCache _cache;
+            private readonly Action<IReadOnlyList<CultCacheDocumentChange<object>>> _journal;
+            private bool _removed;
+
+            public JournalRegistration(CultCache cache, Action<IReadOnlyList<CultCacheDocumentChange<object>>> journal)
+            {
+                _cache = cache;
+                _journal = journal;
+            }
+
+            public void Dispose()
+            {
+                if (_removed)
+                    return;
+                _cache.Held(() => _cache._journals.Remove(_journal));
+                _removed = true;
+            }
+        }
+
         private IDisposable Register(Observer<Change> observer)
         {
             lock (_observerGate)
@@ -1924,6 +1950,23 @@ namespace GameCult.Caching
             return Watch<T>().Where(change => change.Key.Equals(key));
         }
 
+        // An ordered admission journal: called under the gate, right after an admission is applied, once per admission
+        // that changed something, with that admission's changes in Sequence order. Journals of one cache run in
+        // registration order. A journal sees the new state (reads work) and is the one place cache order is a fact: it
+        // must not write to the cache or wait for anything that needs it, and any hold on this cache while a journal runs
+        // throws. A journal's exception does not undo the admission: it is published, then the exception is rethrown to
+        // the writer, as an observer's is.
+        public IDisposable AddJournal(Action<IReadOnlyList<CultCacheDocumentChange<object>>> journal)
+        {
+            if (journal == null) throw new ArgumentNullException(nameof(journal));
+            Held(() =>
+            {
+                _journals.Add(journal);
+                return true;
+            });
+            return new JournalRegistration(this, journal);
+        }
+
         // Attaching reads the store; there is no interval in which it is attached but unread.
         public void AddBackingStore(CacheBackingStore store, params Type[] homes)
         {
@@ -2372,40 +2415,51 @@ namespace GameCult.Caching
             var outcome = land(home, plan.Admitted);
             if (outcome != CultCommitOutcome.Committed)
                 return outcome;
-            foreach (var change in Apply(plan, evicted, source))
-                _held.Add((change, source != null));
+            var applied = Apply(plan, evicted, source);
+            foreach (var change in applied)
+                _held.Changes.Add((change, source != null));
+            RunJournals(applied, _held.Failures);
             return CultCommitOutcome.Committed;
         });
 
         // 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 reaches every observer even
+        // guaranteed; each change carries the Sequence it was admitted with. Journals run before that, under the gate,
+        // in Sequence order (AddJournal); no hold may be entered on this cache while one runs. 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)
         {
             if (Monitor.IsEntered(_gate))
+            {
+                if (_journaling)
+                    throw new InvalidOperationException("A journal entered the cache it journals; a journal must not write to it.");
                 return body();
+            }
+
             if (_held != null)
                 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)>();
-            T result;
+            var mine = _held = new Hold();
+            var result = default(T)!;
             try
             {
                 lock (_gate)
                     result = body();
             }
-            catch
+            catch (Exception exception)
             {
                 _held = null;
                 Publish(mine);
-                throw;
+                if (mine.Failures.Count == 0)
+                    throw;
+                throw new AggregateException(mine.Failures.Prepend(exception));
             }
 
             _held = null;
-            var failures = Publish(mine);
+            var failures = mine.Failures;
+            failures.AddRange(Publish(mine));
             if (failures.Count == 1)
                 ExceptionDispatchInfo.Capture(failures[0]).Throw();
             if (failures.Count > 1)
@@ -2413,10 +2467,10 @@ namespace GameCult.Caching
             return result;
         }
 
-        private List<Exception> Publish(List<(Change Change, bool Loaded)> changes)
+        private List<Exception> Publish(Hold hold)
         {
             var failures = new List<Exception>();
-            foreach (var (change, loaded) in changes)
+            foreach (var (change, loaded) in hold.Changes)
             {
                 // 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.
@@ -2447,6 +2501,36 @@ namespace GameCult.Caching
             return failures;
         }
 
+        // Runs under the gate. A journal that throws joins the hold's failures and the remaining journals still run.
+        private void RunJournals(List<Change> applied, List<Exception> failures)
+        {
+            if (_journals.Count == 0 || applied.Count == 0)
+                return;
+            var changes = applied
+                .Select(change => new CultCacheDocumentChange<object>(
+                    change.Kind, change.Stored.Key, change.Document, change.Previous, change.Sequence))
+                .ToArray();
+            _journaling = true;
+            try
+            {
+                foreach (var journal in _journals)
+                {
+                    try
+                    {
+                        journal(changes);
+                    }
+                    catch (Exception exception)
+                    {
+                        failures.Add(exception);
+                    }
+                }
+            }
+            finally
+            {
+                _journaling = false;
+            }
+        }
+
         private CacheBackingStore? Validate(
             IReadOnlyList<CultStoredDocument> admitted,
             IReadOnlyList<CultStoredDocument> evicted,
@@ -2992,6 +3076,13 @@ namespace GameCult.Caching
             public CultRecordKey Key { get; }
         }
 
+        // What one outermost hold admitted, and what its journals threw.
+        private sealed class Hold
+        {
+            public List<(Change Change, bool Loaded)> Changes { get; } = new();
+            public List<Exception> Failures { get; } = new();
+        }
+
         private sealed class Change
         {
             public Change(CultCacheDocumentChangeKind kind, CultStoredDocument stored, object? document, object? previous, long sequence)

tests/GameCult.Caching.Tests/CultCacheJournalTests.cs

@@ -0,0 +1,402 @@
+#nullable enable
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Linq;
+using System.Threading.Tasks;
+using GameCult.Caching;
+using GameCult.Caching.MessagePack;
+using MessagePack;
+using NUnit.Framework;
+using R3;
+
+namespace GameCult.Caching.Tests
+{
+    // Cut 3 of docs/cultcache-publication-cut.md: a journal is called under the cache gate, right after an admission is
+    // applied, once per admission, with that admission's changes in Sequence order. It is the one place cache order is a fact.
+    [NonParallelizable]
+    public class CultCacheJournalTests
+    {
+        private static readonly CultDocumentRegistry Registry = CultDocumentRegistry.ForTypes(new[] { typeof(JournalNote) });
+
+        private string _directory = string.Empty;
+
+        [SetUp]
+        public void Prepare()
+        {
+            _directory = Path.Combine(Path.GetTempPath(), $"cultlib-journal-{Guid.NewGuid():N}");
+            Directory.CreateDirectory(_directory);
+            JournalNote.Armed = false;
+        }
+
+        [TearDown]
+        public void Cleanup()
+        {
+            JournalNote.Armed = false;
+            if (Directory.Exists(_directory))
+                Directory.Delete(_directory, recursive: true);
+        }
+
+        [CultDocument("tests.journal_note", "tests.journal_note.v1")]
+        [MessagePackObject(AllowPrivate = true)]
+        internal sealed class JournalNote
+        {
+            public static volatile bool Armed;
+            [IgnoreMember] private string _name = string.Empty;
+
+            [Key(0)]
+            [CultName]
+            public string Name
+            {
+                get
+                {
+                    if (Armed && _name == "boom")
+                        throw new InvalidOperationException("name getter boom");
+                    return _name;
+                }
+                set => _name = value;
+            }
+
+            [Key(1)]
+            public int Power;
+        }
+
+        private static CultRecordKey KeyOf(string text) => new($"journal:{text}");
+
+        private static CultCache Memory() => new(Registry, CultCacheMessagePack.CreateCodec(Registry));
+
+        private static void Put(CultCache cache, string name, int power = 1) =>
+            cache.Commit(batch => batch.Upsert(typeof(JournalNote), new JournalNote { Name = name, Power = power }, KeyOf(name)));
+
+        [Test]
+        public void EveryAdmissionIsJournaledInSequenceOrderAndItsChangesArriveTogether()
+        {
+            using var cache = Memory();
+            var calls = new List<long[]>();
+            using var journal = cache.AddJournal(changes => calls.Add(changes.Select(change => change.Sequence).ToArray()));
+            const int threads = 8;
+            const int writes = 1000;
+
+            Parallel.For(0, threads, new ParallelOptions { MaxDegreeOfParallelism = threads }, thread =>
+            {
+                for (var i = 0; i < writes; i++)
+                    cache.Commit(batch =>
+                    {
+                        batch.Upsert(typeof(JournalNote), new JournalNote { Name = $"{thread}-{i}-a" }, KeyOf($"{thread}-{i}-a"));
+                        batch.Upsert(typeof(JournalNote), new JournalNote { Name = $"{thread}-{i}-b" }, KeyOf($"{thread}-{i}-b"));
+                    });
+            });
+
+            Assert.That(calls, Has.Count.EqualTo(threads * writes), "one call per admission");
+            var expected = 1L;
+            foreach (var call in calls)
+            {
+                Assert.That(call, Is.EqualTo(new[] { expected, expected + 1 }), "contiguous, and one call's changes together");
+                expected += 2;
+            }
+        }
+
+        [Test]
+        public void AJournalRunsAfterTheAdmissionIsAppliedAndBeforeAnyObserver()
+        {
+            using var cache = Memory();
+            var order = new List<string>();
+            using var observer = cache.Watch<JournalNote>().Subscribe(change => order.Add($"observer:{change.Sequence}"));
+            var seenInside = new List<string?>();
+            using var journal = cache.AddJournal(changes =>
+            {
+                order.Add($"journal:{changes[0].Sequence}");
+                seenInside.Add(cache.Get<JournalNote>(KeyOf("one"))?.Name);
+                Assert.That(changes[0].Kind, Is.EqualTo(order.Count == 1 ? CultCacheDocumentChangeKind.Added : CultCacheDocumentChangeKind.Updated));
+                Assert.That(((JournalNote?)changes[0].Document)?.Power, Is.EqualTo(order.Count == 1 ? 1 : 2));
+                Assert.That(((JournalNote?)changes[0].PreviousDocument)?.Power, Is.EqualTo(order.Count == 1 ? null : 1));
+            });
+
+            Put(cache, "one", 1);
+            Put(cache, "one", 2);
+
+            Assert.That(order, Is.EqualTo(new[] { "journal:1", "observer:1", "journal:2", "observer:2" }));
+            Assert.That(seenInside, Is.EqualTo(new[] { "one", "one" }), "the journal reads the new state");
+        }
+
+        [Test]
+        public void AnAdmissionThatChangesNothingIsNotJournaled()
+        {
+            using var cache = Memory();
+            var calls = 0;
+            using var journal = cache.AddJournal(_ => calls++);
+
+            // Removing a record the cache does not hold is a commit that admits nothing.
+            cache.Commit(batch => batch.Remove(KeyOf("absent")));
+
+            Assert.That(calls, Is.Zero);
+        }
+
+        [Test]
+        public void JournalsRunInRegistrationOrderAndADisposedJournalStopsRunning()
+        {
+            using var cache = Memory();
+            var order = new List<string>();
+            using var first = cache.AddJournal(_ => order.Add("first"));
+            var second = cache.AddJournal(_ => order.Add("second"));
+            using var third = cache.AddJournal(_ => order.Add("third"));
+            Put(cache, "one");
+
+            second.Dispose();
+            order.Add("|");
+            Put(cache, "two");
+
+            Assert.That(order, Is.EqualTo(new[] { "first", "second", "third", "|", "first", "third" }));
+        }
+
+        [Test]
+        public void AJournalThatWritesTheCacheThrowsAndTheAdmissionStands()
+        {
+            using var cache = Memory();
+            var seen = new List<long>();
+            using var observer = cache.Watch<JournalNote>().Subscribe(change => seen.Add(change.Sequence));
+            using var journal = cache.AddJournal(_ => Put(cache, "nested"));
+
+            var thrown = Assert.Throws<InvalidOperationException>(() => Put(cache, "outer"))!;
+
+            Assert.Multiple(() =>
+            {
+                Assert.That(thrown.Message, Does.Contain("journal"));
+                Assert.That(cache.Get<JournalNote>(KeyOf("outer")), Is.Not.Null, "the admission stands");
+                Assert.That(cache.Get<JournalNote>(KeyOf("nested")), Is.Null, "the journal's own write did not land");
+                Assert.That(seen, Is.EqualTo(new[] { 1L }), "and it was published");
+            });
+        }
+
+        [Test]
+        public void AJournalThatRegistersAnotherJournalThrows()
+        {
+            using var cache = Memory();
+            using var journal = cache.AddJournal(_ => cache.AddJournal(_ => { }));
+
+            var thrown = Assert.Throws<InvalidOperationException>(() => Put(cache, "outer"))!;
+
+            Assert.That(thrown.Message, Does.Contain("journal"));
+            Assert.That(cache.Get<JournalNote>(KeyOf("outer")), Is.Not.Null);
+        }
+
+        [Test]
+        public void TheCacheIsWritableAgainOnceAJournalHasReturned()
+        {
+            using var cache = Memory();
+            using var journal = cache.AddJournal(_ => { });
+
+            Put(cache, "one");
+            cache.Commit(batch => batch.Upsert(typeof(JournalNote), new JournalNote { Name = "two" }, KeyOf("two")));
+            using var late = cache.AddJournal(_ => { });
+
+            Assert.That(cache.Get<JournalNote>(KeyOf("two")), Is.Not.Null);
+        }
+
+        [Test]
+        public void ARefusedAdmissionIsNotJournaled()
+        {
+            using var cache = Memory();
+            var calls = 0;
+            using var journal = cache.AddJournal(_ => calls++);
+            JournalNote.Armed = true;
+
+            Assert.Throws<InvalidOperationException>(() => Put(cache, "boom"));
+
+            JournalNote.Armed = false;
+            Assert.That(calls, Is.Zero);
+        }
+
+        [Test]
+        public void AnAdmissionTheStoreDeclinesIsNotJournaled()
+        {
+            using var cache = Memory();
+            Put(cache, "seed");
+            var calls = 0;
+            using var journal = cache.AddJournal(_ => calls++);
+
+            var outcome = cache.TryCommit(batch =>
+            {
+                batch.Expect(KeyOf("seed"), null);
+                batch.Upsert(typeof(JournalNote), new JournalNote { Name = "two" }, KeyOf("two"));
+            });
+
+            Assert.That(outcome, Is.EqualTo(CultCommitOutcome.Mismatch));
+            Assert.That(calls, Is.Zero);
+            Assert.That(cache.Get<JournalNote>(KeyOf("two")), Is.Null);
+        }
+
+        [Test]
+        public void AVariantDependentIsJournaledInTheSameCallAsItsBaseEdit()
+        {
+            using var cache = Memory();
+            Put(cache, "base", 10);
+            cache.Commit(batch => batch.UpsertVariant(KeyOf("big"), KeyOf("base"), new[]
+            {
+                cache.Override<JournalNote>(nameof(JournalNote.Name), "big"),
+            }));
+            var calls = new List<string[]>();
+            using var journal = cache.AddJournal(changes => calls.Add(changes.Select(change => change.Key.Value).ToArray()));
+
+            Put(cache, "base", 20);
+
+            Assert.That(calls, Has.Count.EqualTo(1));
+            Assert.That(calls[0], Is.EquivalentTo(new[] { KeyOf("base").Value, KeyOf("big").Value }));
+        }
+
+        [Test]
+        public void AThrowingJournalDoesNotStopTheOthersOrTheAdmissionAndTheWriterHearsOfItAfterPublication()
+        {
+            using var cache = Memory();
+            var order = new List<string>();
+            using var observer = cache.Watch<JournalNote>().Subscribe(_ => order.Add("observer"));
+            using var a = cache.AddJournal(_ =>
+            {
+                order.Add("a");
+                throw new InvalidOperationException("a");
+            });
+            using var b = cache.AddJournal(_ => order.Add("b"));
+            using var c = cache.AddJournal(_ =>
+            {
+                order.Add("c");
+                throw new InvalidOperationException("c");
+            });
+
+            var thrown = Assert.Throws<AggregateException>(() => Put(cache, "one"))!;
+
+            Assert.Multiple(() =>
+            {
+                Assert.That(thrown.InnerExceptions.Select(exception => exception.Message), Is.EqualTo(new[] { "a", "c" }));
+                Assert.That(order, Is.EqualTo(new[] { "a", "b", "c", "observer" }), "every journal ran, then the change published");
+                Assert.That(cache.Get<JournalNote>(KeyOf("one")), Is.Not.Null, "the admission stands");
+            });
+
+            using var quiet = cache.Watch<JournalNote>().Subscribe(_ => { });
+            Assert.Throws<AggregateException>(() => Put(cache, "two"), "a throwing journal stays registered");
+        }
+
+        [Test]
+        public void ADisposeTheCacheRefusesLeavesTheJournalRegisteredAndALaterDisposeRemovesIt()
+        {
+            using var cache = Memory();
+            var calls = 0;
+            IDisposable? registration = null;
+            registration = cache.AddJournal(_ =>
+            {
+                calls++;
+                registration!.Dispose();
+            });
+
+            var thrown = Assert.Throws<InvalidOperationException>(() => Put(cache, "one"))!;
+            Assert.That(thrown.Message, Does.Contain("journal"), "disposing from inside a journal is refused");
+
+            registration.Dispose();
+            Put(cache, "two");
+
+            Assert.That(calls, Is.EqualTo(1), "the outside dispose removed it, though the first attempt threw");
+            registration.Dispose();
+        }
+
+        [Test]
+        public void ALoadIsJournaledWithItsChanges()
+        {
+            var path = Path.Combine(_directory, "notes.cc");
+            using (var writer = CultCacheMessagePack.Create(path, new CultCacheOpenOptions { Registry = Registry }))
+            {
+                Put(writer, "a");
+                Put(writer, "b");
+            }
+
+            using var cache = Memory();
+            var journaled = new List<IReadOnlyList<CultCacheDocumentChange<object>>>();
+            using var journal = cache.AddJournal(changes => journaled.Add(changes));
+
+            cache.AddBackingStore(new SingleFileMessagePackBackingStore(path));
+
+            Assert.Multiple(() =>
+            {
+                Assert.That(journaled, Has.Count.EqualTo(1), "one admission, one call");
+                Assert.That(journaled[0].Select(change => change.Key).OrderBy(key => key.Value),
+                    Is.EqualTo(new[] { KeyOf("a"), KeyOf("b") }));
+            });
+        }
+
+        // A store that admits a load and then fails: the hold's body throws after the journal has already failed.
+        private sealed class LoadThenFailStore : CacheBackingStore
+        {
+            public bool Armed;
+
+            public override void PullAll() => Held(() =>
+            {
+                if (!Armed)
+                    return;
+                var record = new CultStoredDocument(
+                    KeyOf("loaded"), "2026-01-01T00:00:00.0000000+00:00", Registry.GetRequired(typeof(JournalNote)), new JournalNote { Name = "loaded" });
+                Loaded!(new[] { record }, Array.Empty<CultStoredDocument>());
+                throw new InvalidOperationException("store failed after the load");
+            });
+
+            public override void Push(CultStoredDocument entry) { }
+            public override void Delete(CultStoredDocument entry) { }
+            public override void PushAll() { }
+            public override CultCommitOutcome CommitBatch(CultCommitRequest request, bool wait) => CultCommitOutcome.Committed;
+        }
+
+        [Test]
+        public void AHoldWhoseBodyThrowsSurfacesTheBodyFirstThenTheJournalFailures()
+        {
+            using var cache = Memory();
+            var store = new LoadThenFailStore();
+            cache.AddBackingStore(store);
+            using var a = cache.AddJournal(_ => throw new InvalidOperationException("journal a"));
+            using var b = cache.AddJournal(_ => throw new InvalidOperationException("journal b"));
+            store.Armed = true;
+
+            var thrown = Assert.Throws<AggregateException>(() => store.PullAll())!;
+
+            Assert.Multiple(() =>
+            {
+                Assert.That(thrown.InnerExceptions.Select(exception => exception.Message),
+                    Is.EqualTo(new[] { "store failed after the load", "journal a", "journal b" }));
+                Assert.That(cache.Get<JournalNote>(KeyOf("loaded")), Is.Not.Null, "the load landed");
+            });
+        }
+
+        [Test]
+        public void AHoldWhoseBodyThrowsAloneRethrowsItAsItself()
+        {
+            using var cache = Memory();
+            var store = new LoadThenFailStore();
+            cache.AddBackingStore(store);
+            using var journal = cache.AddJournal(_ => { });
+            store.Armed = true;
+
+            var thrown = Assert.Throws<InvalidOperationException>(() => store.PullAll())!;
+
+            Assert.That(thrown.Message, Is.EqualTo("store failed after the load"));
+        }
+
+        [Test]
+        public void AJournalThatThrowsDuringALoadLeavesTheStoreAttachedAndTheLoadAdopted()
+        {
+            var path = Path.Combine(_directory, "notes.cc");
+            using (var writer = CultCacheMessagePack.Create(path, new CultCacheOpenOptions { Registry = Registry }))
+                Put(writer, "stored");
+
+            var cache = Memory();
+            var seen = new List<long>();
+            using var observer = cache.Watch<JournalNote>().Subscribe(change => seen.Add(change.Sequence));
+            using var journal = cache.AddJournal(_ => throw new InvalidOperationException("journal boom"));
+
+            var thrown = Assert.Throws<InvalidOperationException>(() => cache.AddBackingStore(new SingleFileMessagePackBackingStore(path)))!;
+
+            Assert.Multiple(() =>
+            {
+                Assert.That(thrown.Message, Is.EqualTo("journal boom"));
+                Assert.That(cache.BackingStores, Has.Count.EqualTo(1), "the store stays attached: it and the cache agree");
+                Assert.That(cache.Get<JournalNote>(KeyOf("stored")), Is.Not.Null);
+                Assert.That(seen, Is.EqualTo(new[] { 1L }));
+            });
+        }
+    }
+}

tests/GameCult.Caching.Tests/CultCacheObserverTests.cs

@@ -2,6 +2,7 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
+using System.Runtime.CompilerServices;
 using GameCult.Caching;
 using GameCult.Caching.MessagePack;
 using MessagePack;
@@ -122,6 +123,87 @@ namespace GameCult.Caching.Tests
             });
         }
 
+        [Test]
+        public void AnUnsubscribedObserverIsReleasedByTheCache()
+        {
+            using var cache = new CultCache(Registry);
+            var kept = new List<IDisposable>();
+            var live = Subscribe(cache, kept, dispose: false);
+            var gone = Subscribe(cache, kept, dispose: true);
+            Send(cache, "one");
+            CollectGarbage();
+
+            Assert.Multiple(() =>
+            {
+                Assert.That(live.IsAlive, Is.True, "a live subscription is held by the cache");
+                Assert.That(gone.IsAlive, Is.False, "an unsubscribed observer is no longer held");
+            });
+            GC.KeepAlive(kept);
+            GC.KeepAlive(cache);
+        }
+
+        // A disposed R3 subscription ignores OnNext, so a stale entry in the cache is invisible to a receiver; it shows as
+        // a leak: the cache keeps the subscriber alive. The test frame must not hold the subscription itself.
+        [MethodImpl(MethodImplOptions.NoInlining)]
+        private static WeakReference Subscribe(CultCache cache, List<IDisposable> kept, bool dispose)
+        {
+            var target = new object();
+            var subscription = cache.Watch<ObservedPing>().Subscribe(_ => GC.KeepAlive(target));
+            if (dispose)
+                subscription.Dispose();
+            else
+                kept.Add(subscription);
+            return new WeakReference(target);
+        }
+
+        private static void CollectGarbage()
+        {
+            GC.Collect();
+            GC.WaitForPendingFinalizers();
+            GC.Collect();
+        }
+
+        // A store that admits a load and then fails: the hold's body throws after the changes were minted.
+        private sealed class LoadThenFailStore : CacheBackingStore
+        {
+            public bool Armed;
+
+            public override void PullAll() => Held(() =>
+            {
+                if (!Armed)
+                    return;
+                var record = new CultStoredDocument(
+                    KeyOf("loaded"), "2026-01-01T00:00:00.0000000+00:00", Registry.GetRequired(typeof(ObservedPing)), new ObservedPing { Text = "loaded" });
+                Loaded!(new[] { record }, Array.Empty<CultStoredDocument>());
+                throw new InvalidOperationException("store failed after the load");
+            });
+
+            public override void Push(CultStoredDocument entry) { }
+            public override void Delete(CultStoredDocument entry) { }
+            public override void PushAll() { }
+            public override CultCommitOutcome CommitBatch(CultCommitRequest request, bool wait) => CultCommitOutcome.Committed;
+        }
+
+        [Test]
+        public void AHoldWhoseBodyThrowsStillPublishesTheChangesItMinted()
+        {
+            using var cache = new CultCache(Registry);
+            var store = new LoadThenFailStore();
+            cache.AddBackingStore(store);
+            var seen = new List<long>();
+            using var subscription = cache.Watch<ObservedPing>().Subscribe(change => seen.Add(change.Sequence));
+            store.Armed = true;
+
+            var thrown = Assert.Throws<InvalidOperationException>(() => store.PullAll())!;
+
+            Assert.Multiple(() =>
+            {
+                Assert.That(thrown.Message, Is.EqualTo("store failed after the load"), "the body's own exception wins");
+                Assert.That(cache.Get<ObservedPing>(KeyOf("loaded")), Is.Not.Null, "the load landed");
+                Assert.That(seen, Is.EqualTo(new[] { 1L }), "and reached its observers");
+            });
+        }
+
         [Test]
         public void DisposeCompletesObserversAndACompletedCacheCompletesNewOnes()
         {