diff --git a/src/GameCult.Caching/CultCache.cs b/src/GameCult.Caching/CultCache.cs index 35315c8b..c54884d5 100644 --- a/src/GameCult.Caching/CultCache.cs +++ b/src/GameCult.Caching/CultCache.cs @@ -2687,8 +2687,8 @@ namespace GameCult.Caching /// again, up to times. The write is durable when Update returns. /// /// - /// may run up to 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 + /// may run up to 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 (). Running out of attempts returns ; 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(CultUpdateOutcome.Declined, Get(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(); + 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. diff --git a/tests/GameCult.Caching.Tests/UpdateTests.cs b/tests/GameCult.Caching.Tests/UpdateTests.cs index 828f03bf..63462ba6 100644 --- a/tests/GameCult.Caching.Tests/UpdateTests.cs +++ b/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(); + using var subscription = cache.Watch().Subscribe(change => published.Add(change.Document!.Value)); + var calls = 0; + + var result = cache.Update(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(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(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(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().Subscribe(_ => changes++); + var calls = 0; + + var result = cache.Update(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 BaseHandle = new(Key); + + private static void RivalDerived(CultCache cache, int value) => + Assert.That(cache.Commit(batch => batch.Upsert(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(); + + var result = cache.Update(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(Key)!).Value, Is.EqualTo(2)); + } + + [Test] + public void RoutedSubtypeCreateLandsInItsHome() + { + using (var cache = OpenRouted()) + { + var calls = 0; + var result = cache.Update(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(Key), Is.InstanceOf()); + using (var inA = OpenOneRouted("a.cc", typeof(UpdateBase))) + Assert.That(inA.Get(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(() => + cache.Update(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(new UpdateDerived { Name = "counter", Value = 1 }, BaseHandle)), Is.True); + var calls = 0; + + var refusal = Assert.Throws(() => cache.Update(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; + } } }