Caching: Update reads every candidate home of T for its version and copy, refuses a two-store key and a cross-home move; pin the unpinned rules

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

src/GameCult.Caching/CultCache.cs

@@ -2687,8 +2687,8 @@ namespace GameCult.Caching
         /// <paramref name="apply"/> again, up to <paramref name="maxAttempts"/> times. The write is durable when Update returns.
         /// </summary>
         /// <remarks>
-        /// <paramref name="apply"/> may run up to <paramref name="maxAttempts"/> times. It gets the record as its home store holds it now,
-        /// decoded afresh for this attempt (null when the key holds nothing), which it may change and return. A change made in place to an
+        /// <paramref name="apply"/> may run up to <paramref name="maxAttempts"/> times. It gets the record as the store that holds it has it now, read from every home
+        /// of T and its routed subtypes and decoded afresh for this attempt (null when the key holds nothing), which it may change and return. A record held by two stores, or an apply that changes the record's home store, throws. A change made in place to an
         /// instance the cache holds is neither seen nor written. It must not write to the cache or have other side effects. Return null to write
         /// nothing. A store with staged blind writes refuses Update until it is flushed, as it refuses any conditional commit. A cache with no
         /// store attached copies its held instance and needs a codec; a store kind outside the library cannot serve Update (<see cref="NotSupportedException"/>). Running out of attempts returns <see cref="CultUpdateOutcome.LostRace"/>; every other failure throws.
@@ -2699,33 +2699,49 @@ namespace GameCult.Caching
             if (maxAttempts < 1) throw new ArgumentOutOfRangeException(nameof(maxAttempts), maxAttempts, "At least one attempt is required.");
             for (var attempt = 1; ; attempt++)
             {
-                var (version, copy) = Held(() =>
+                var (version, copy, source) = Held(() =>
                 {
-                    _entries.TryGetValue(key.Value, out var held);
                     if (_stores.Count == 0)
                     {
                         // No store: the held entry is the record.
+                        _entries.TryGetValue(key.Value, out var held);
                         var codec = _codec ?? throw new InvalidOperationException("This cache has no codec; create it with CultCacheMessagePack.Create.");
                         if (held == null)
-                            return ((string?)null, (T?)null);
+                            return ((string?)null, (T?)null, (CacheBackingStore?)null);
                         if (held.Variant != null || held.Document is not T)
                             throw new InvalidOperationException($"Record {key.Value} is a variant or not a {typeof(T).Name}; Update cannot copy it.");
-                        return (held.StoredVersion, (T?)codec.Clone(held.Document, held.Descriptor.DocumentType));
+                        return (held.StoredVersion, (T?)codec.Clone(held.Document, held.Descriptor.DocumentType), null);
+                    }
+
+                    // The store with the key is among the homes of T and its routed subtypes; the cache's own view of the key may be stale, so it is not asked.
+                    var homes = CandidateHomes(typeof(T));
+                    if (homes.Length == 0)
+                        throw new InvalidOperationException("A conditional batch names its home store through a record it upserts or removes.");
+                    CacheBackingStore? holder = null;
+                    CultStoredDocument? durable = null;
+                    foreach (var candidate in homes)
+                    {
+                        if (candidate.ReadDurable(key.Value) is not { } found)
+                            continue;
+                        if (holder != null)
+                            throw new InvalidOperationException($"Record {key.Value} is held by two stores, {holder} and {candidate}; Update cannot choose.");
+                        holder = candidate;
+                        durable = found;
                     }
 
-                    var home = Home(held?.Descriptor.DocumentType ?? typeof(T))
-                               ?? throw new InvalidOperationException("A conditional batch names its home store through a record it upserts or removes.");
-                    var durable = home.ReadDurable(key.Value);
                     if (durable == null)
-                        return ((string?)null, (T?)null);
+                        return ((string?)null, (T?)null, (CacheBackingStore?)null);
                     if (durable.Variant != null || durable.Document is not T typed)
                         throw new InvalidOperationException($"Record {key.Value} is a variant or not a {typeof(T).Name}; Update cannot copy it.");
                     CultElementIds.Plan(durable.Descriptor.DocumentType, durable.Document, key.Value, deterministic: true).Apply();
-                    return (durable.StoredVersion, typed);
+                    return (durable.StoredVersion, typed, holder);
                 });
                 var next = apply(copy);
                 if (next == null)
                     return new CultUpdateResult<T>(CultUpdateOutcome.Declined, Get<T>(key));
+                if (source != null && Held(() => Home(next.GetType())) is var landing && landing != source)
+                    throw new InvalidOperationException(
+                        $"Update of {key.Value} would move the record from {source} to {landing?.ToString() ?? "memory"}; remove it and write it at its new home.");
                 CultCommitOutcome outcome;
                 try
                 {
@@ -3770,6 +3786,17 @@ namespace GameCult.Caching
         }
 
         // The most derived routed home type assignable from the document type, else the untyped store.
+        private CacheBackingStore[] CandidateHomes(Type type)
+        {
+            var homes = new HashSet<CacheBackingStore>();
+            if (Home(type) is { } own)
+                homes.Add(own);
+            foreach (var descriptor in _registry.AllDescriptors)
+                if (type.IsAssignableFrom(descriptor.DocumentType) && Home(descriptor.DocumentType) is { } routed)
+                    homes.Add(routed);
+            return _stores.Select(entry => entry.Store).Where(homes.Contains).ToArray();
+        }
+
         private CacheBackingStore? Home(Type documentType)
         {
             CacheBackingStore? untyped = null;
@@ -5192,8 +5219,6 @@ namespace GameCult.Caching
 
         // The durable record at a key decoded as a load decodes it, at its durable version: null when the store holds none there or
         // holds it foreign. The one decode both refusals reload through.
-        internal override CultStoredDocument? ReadDurable(string key) => Held(() => DurableAt(ReadSnapshot(), key));
-
         private CultStoredDocument? DurableAt(Durable? durable, string key)
         {
             var record = durable?.Snapshot.Records.FirstOrDefault(candidate => string.Equals(candidate.Key, key, StringComparison.Ordinal));
@@ -5202,6 +5227,9 @@ namespace GameCult.Caching
             return ToStoredDocument(record, durable.Version(key), durable.Snapshot.SchemaCatalog, DeserializeMarked, out _);
         }
 
+        // Update's read: the durable record at the key, decoded afresh, stamped with the version CommitBatch compares.
+        internal override CultStoredDocument? ReadDurable(string key) => Held(() => DurableAt(ReadSnapshot(), key));
+
         // The file now holds every record written, under the id and at the storedAt the write gave it, and the version of those bytes:
         // the store's entries and the cache's say so. A copied record is on disk as the file had it, so nothing learns anything new about it: in particular the
         // ids it minted at load are still in memory only.

tests/GameCult.Caching.Tests/UpdateTests.cs

@@ -19,7 +19,7 @@ namespace GameCult.Caching.Tests
     {
         private static readonly CultRecordKey Key = new("counter");
         private static readonly CultDocumentRegistry Registry =
-            CultDocumentRegistry.ForTypes(new[] { typeof(UpdateCounter), typeof(UpdateTally), typeof(UpdateGear) });
+            CultDocumentRegistry.ForTypes(new[] { typeof(UpdateCounter), typeof(UpdateTally), typeof(UpdateGear), typeof(UpdateBase), typeof(UpdateDerived) });
 
         private string _directory = string.Empty;
 
@@ -178,6 +178,193 @@ namespace GameCult.Caching.Tests
             Assert.That(StoredValue(path, false), Is.EqualTo(1005));
         }
 
+        // No publication in the middle of an Update: a rival's write that cost this cache an attempt is never published on its own,
+        // only the value this Update commits (a pull after every lost attempt would publish both).
+        [TestCase(false)]
+        [TestCase(true)]
+        public void LostAttemptPublishesNothingMidUpdate(bool directory)
+        {
+            var path = PathOf("midupdate.cc");
+            Seed(path, directory, 1);
+            using var cache = Open(path, directory);
+            using var rival = Open(path, directory);
+            var published = new List<int>();
+            using var subscription = cache.Watch<UpdateCounter>().Subscribe(change => published.Add(change.Document!.Value));
+            var calls = 0;
+
+            var result = cache.Update<UpdateCounter>(Key, counter =>
+            {
+                if (++calls == 1)
+                    Rival(rival, 5);
+                return Plus(counter, 1);
+            });
+
+            Assert.That(result.Outcome, Is.EqualTo(CultUpdateOutcome.Updated));
+            Assert.That(calls, Is.EqualTo(2));
+            Assert.That(published, Is.EqualTo(new[] { 6 }), "the rival's 5 was published in the middle of the Update");
+        }
+
+        // The directory store reads only the key's own page, so a damaged page of another record does not stop an Update.
+        [Test]
+        public void DirectoryUpdateReadsOnlyItsOwnPage()
+        {
+            var path = PathOf("pages.cc");
+            var other = new CultRecordKey("other");
+            using (var seed = Open(path, true))
+                Assert.That(seed.Commit(batch => batch.Upsert(new UpdateCounter { Name = "other", Value = 9 }, new CultRecordHandle<UpdateCounter>(other))), Is.True);
+            var otherPages = Directory.GetFiles(path + ".records", "*.msgpack");
+            Assert.That(otherPages, Has.Length.EqualTo(1));
+            Seed(path, true, 1);
+            using var cache = Open(path, true);
+            File.WriteAllBytes(otherPages[0], new byte[] { 1, 2, 3, 4, 5 });
+
+            var result = cache.Update<UpdateCounter>(Key, counter => Plus(counter, 1));
+
+            Assert.That(result.Outcome, Is.EqualTo(CultUpdateOutcome.Updated));
+            Assert.That(result.Document!.Value, Is.EqualTo(2));
+        }
+
+        // The durable read is not a load: it leaves the store's foreign list, its migration reports and what the cache publishes alone.
+        [TestCase(false)]
+        [TestCase(true)]
+        public void ReadDurableLeavesTheStoreViewAlone(bool directory)
+        {
+            var path = PathOf("view.cc");
+            Seed(path, directory, 1);
+            using (var writer = Open(path, directory))
+                Assert.That(writer.Commit(batch => batch.Upsert(new UpdateGear { Name = "gear", Power = 3 }, new CultRecordHandle<UpdateGear>(new CultRecordKey("gear")))), Is.True);
+            var narrow = CultDocumentRegistry.ForTypes(new[] { typeof(UpdateCounter) });
+            using var cache = new CultCache(narrow);
+            cache.AddBackingStore(directory ? new DirectoryMessagePackBackingStore(path) : new SingleFileMessagePackBackingStore(path));
+            using var rival = Open(path, directory);
+            var store = cache.BackingStores.Single();
+            var foreign = store.ForeignRecords;
+            var reports = store.LastSchemaMigrationReports;
+            Assert.That(foreign.Select(record => record.Key), Is.EqualTo(new[] { "gear" }), "the fixture holds a foreign record");
+            Assert.That(reports, Is.Not.Empty, "the fixture holds a migration report");
+            var foreignKeys = foreign.Select(record => (record.Key, record.SchemaId, record.StoredAt)).ToArray();
+            var reportKinds = reports.Select(report => (report.PersistedSchemaName, report.Kind)).ToArray();
+            var changes = 0;
+            using var subscription = cache.Watch<UpdateCounter>().Subscribe(_ => changes++);
+            var calls = 0;
+
+            var result = cache.Update<UpdateCounter>(Key, counter =>
+            {
+                if (++calls == 1)
+                    Rival(rival, 5);
+                return Plus(counter, 1);
+            });
+
+            Assert.That(result.Outcome, Is.EqualTo(CultUpdateOutcome.Updated));
+            Assert.That(calls, Is.EqualTo(2), "the Update reread the key");
+            Assert.That(store.ForeignRecords, Is.SameAs(foreign));
+            Assert.That(store.LastSchemaMigrationReports, Is.SameAs(reports));
+            Assert.That(store.ForeignRecords.Select(record => (record.Key, record.SchemaId, record.StoredAt)), Is.EqualTo(foreignKeys));
+            Assert.That(store.LastSchemaMigrationReports.Select(report => (report.PersistedSchemaName, report.Kind)), Is.EqualTo(reportKinds));
+            Assert.That(changes, Is.EqualTo(1), "only the committed write was published");
+        }
+
+        // Routing: store A is home to UpdateBase, store B to UpdateDerived : UpdateBase. Land judges a derived record at B.
+        private CultCache OpenRouted()
+        {
+            var cache = new CultCache(Registry);
+            cache.AddBackingStore(new SingleFileMessagePackBackingStore(PathOf("a.cc")), typeof(UpdateBase));
+            cache.AddBackingStore(new SingleFileMessagePackBackingStore(PathOf("b.cc")), typeof(UpdateDerived));
+            return cache;
+        }
+
+        private CultCache OpenOneRouted(string file, Type home)
+        {
+            var cache = new CultCache(Registry);
+            cache.AddBackingStore(new SingleFileMessagePackBackingStore(PathOf(file)), home);
+            return cache;
+        }
+
+        private byte[]? BytesOf(string file) => File.Exists(PathOf(file)) ? File.ReadAllBytes(PathOf(file)) : null;
+
+        private static readonly CultRecordHandle<UpdateBase> BaseHandle = new(Key);
+
+        private static void RivalDerived(CultCache cache, int value) =>
+            Assert.That(cache.Commit(batch => batch.Upsert<UpdateBase>(new UpdateDerived { Name = "counter", Value = value }, BaseHandle)), Is.True);
+
+        [Test]
+        public void RoutedSubtypeStaleCacheConverges()
+        {
+            using var cache = OpenRouted();
+            using var rival = OpenRouted();
+            RivalDerived(rival, 1);
+            var seen = new List<int?>();
+
+            var result = cache.Update<UpdateBase>(Key, document =>
+            {
+                seen.Add((document as UpdateDerived)?.Value);
+                var derived = (UpdateDerived)document!;
+                derived.Value += 1;
+                return derived;
+            });
+
+            Assert.That(seen, Is.EqualTo(new int?[] { 1 }), "apply did not see the rival's record in its home store, once");
+            Assert.That(result.Outcome, Is.EqualTo(CultUpdateOutcome.Updated));
+            using var reader = OpenRouted();
+            Assert.That(((UpdateDerived)reader.Get<UpdateBase>(Key)!).Value, Is.EqualTo(2));
+        }
+
+        [Test]
+        public void RoutedSubtypeCreateLandsInItsHome()
+        {
+            using (var cache = OpenRouted())
+            {
+                var calls = 0;
+                var result = cache.Update<UpdateBase>(Key, document =>
+                {
+                    calls++;
+                    Assert.That(document, Is.Null);
+                    return new UpdateDerived { Name = "counter", Value = 1 };
+                });
+                Assert.That(calls, Is.EqualTo(1));
+                Assert.That(result.Outcome, Is.EqualTo(CultUpdateOutcome.Updated));
+            }
+
+            using (var inB = OpenOneRouted("b.cc", typeof(UpdateDerived)))
+                Assert.That(inB.Get<UpdateBase>(Key), Is.InstanceOf<UpdateDerived>());
+            using (var inA = OpenOneRouted("a.cc", typeof(UpdateBase)))
+                Assert.That(inA.Get<UpdateBase>(Key), Is.Null, "the record landed in the wrong store");
+        }
+
+        [Test]
+        public void MovingTheRecordToAnotherHomeRefuses()
+        {
+            using (var rival = OpenRouted())
+                RivalDerived(rival, 1);
+            var before = (A: BytesOf("a.cc"), B: BytesOf("b.cc"));
+            using var cache = OpenRouted();
+            var calls = 0;
+
+            var refusal = Assert.Throws<InvalidOperationException>(() =>
+                cache.Update<UpdateBase>(Key, _ => { calls++; return new UpdateBase { Name = "counter" }; }))!;
+
+            Assert.That(calls, Is.EqualTo(1), "a refusal was retried");
+            Assert.That(refusal.Message, Does.Contain("a.cc").And.Contain("b.cc").And.Contain("move"));
+            Assert.That(BytesOf("a.cc"), Is.EqualTo(before.A));
+            Assert.That(BytesOf("b.cc"), Is.EqualTo(before.B));
+        }
+
+        [Test]
+        public void ARecordHeldByTwoStoresRefusesBeforeApply()
+        {
+            using var cache = OpenRouted();
+            using (var inA = OpenOneRouted("a.cc", typeof(UpdateBase)))
+                Assert.That(inA.Commit(batch => batch.Upsert(new UpdateBase { Name = "counter" }, BaseHandle)), Is.True);
+            using (var inB = OpenOneRouted("b.cc", typeof(UpdateDerived)))
+                Assert.That(inB.Commit(batch => batch.Upsert<UpdateBase>(new UpdateDerived { Name = "counter", Value = 1 }, BaseHandle)), Is.True);
+            var calls = 0;
+
+            var refusal = Assert.Throws<InvalidOperationException>(() => cache.Update<UpdateBase>(Key, document => { calls++; return document; }))!;
+
+            Assert.That(calls, Is.EqualTo(0));
+            Assert.That(refusal.Message, Does.Contain("held by two stores"));
+        }
+
         [Test]
         public void NullFromApplyDeclines()
         {
@@ -595,5 +782,22 @@ namespace GameCult.Caching.Tests
             [Key(1)]
             public int Power;
         }
+
+        [CultDocument("tests.update_base", "tests.update_base.v1")]
+        [MessagePackObject(AllowPrivate = true)]
+        internal class UpdateBase
+        {
+            [Key(0)]
+            [CultName]
+            public string Name = string.Empty;
+        }
+
+        [CultDocument("tests.update_derived", "tests.update_derived.v1")]
+        [MessagePackObject(AllowPrivate = true)]
+        internal sealed class UpdateDerived : UpdateBase
+        {
+            [Key(1)]
+            public int Value;
+        }
     }
 }