diff --git a/docs/start/README.md b/docs/start/README.md index e9dc0b3f..4430a37a 100644 --- a/docs/start/README.md +++ b/docs/start/README.md @@ -72,8 +72,8 @@ public static class FirstCall ## C# update with retry -Change a record you may not be the only one changing. `Update` reads the record, hands your function a private -copy, and commits the result only if the record is still the one it read. If another writer got there first, it +Change a record you may not be the only one changing. `Update` reads the record as its store holds it, hands your function a fresh +copy of it, and commits the result only if the record is still the one it read. If another writer got there first, it reads again and runs your function again, up to five times. The source is [`tests/GameCult.Caching.Tests/UpdateCallTests.cs`](../../tests/GameCult.Caching.Tests/UpdateCallTests.cs); the snippet reuses `Item` from the first call. diff --git a/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs b/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs index 6f6a4daf..6e0ce6cd 100644 --- a/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs +++ b/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs @@ -64,48 +64,7 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore var reports = new List(); var loaded = new Dictionary(StringComparer.Ordinal); var foreign = new List(); - for (var attempt = 1; ; attempt++) - { - loaded.Clear(); - reports.Clear(); - foreign.Clear(); - // Only pages named by the manifest read under the lease are loaded; orphaned pages are never loaded. With no - // lock to lease (none created yet, or a store copied without it) a writer may commit mid-load, so an unleased - // load stands only if its manifest is still current; if the manifest moved it reloads, under the lease the - // writer created. A page that fails under an unmoved manifest is corruption, not a race. - using var lease = AcquireCommitLease(wait: true, create: false); - var manifestBytes = ReadManifestBytes(); - var (manifest, _) = ParseManifest(manifestBytes); - Trace($"manifest records={manifest.Records.Length}"); - try - { - if (lease == null) - UnleasedManifestRead?.Invoke(); - LoadRecordPages( - manifest.Records.OrderBy(record => record.Key, StringComparer.Ordinal).ToArray(), - manifest.SchemaCatalog, - loaded, - reports, - foreign); - if (lease != null || !ManifestMoved(manifestBytes)) - { - Trace($"indexed-pages loaded={loaded.Count}"); - break; - } - } - catch (Exception exception) when (lease == null && IsTornGeneration(exception)) - { - // The re-read stays out of the filter: an exception thrown inside a filter is swallowed as false. - if (!ManifestMoved(manifestBytes)) - throw; - if (attempt == UnleasedLoadAttempts) - throw Unsettled(exception); - continue; - } - - if (attempt == UnleasedLoadAttempts) - throw Unsettled(null); - } + LoadManifestPages(null, loaded, reports, foreign, Trace); bool Staged(string key) => _dirtyKeys.ContainsKey(key) || _deletedKeys.ContainsKey(key); var arrived = loaded @@ -136,6 +95,70 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore Trace("publish"); } + // The one load of manifest pages, for the whole store (key null) or one record: under the lease, with the moved-manifest retry. + // It reads only; adopting what it loaded is the caller's. + private void LoadManifestPages( + string? key, + Dictionary loaded, + List reports, + List foreign, + Action trace) + { + for (var attempt = 1; ; attempt++) + { + loaded.Clear(); + reports.Clear(); + foreign.Clear(); + // Only pages named by the manifest read under the lease are loaded; orphaned pages are never loaded. With no + // lock to lease (none created yet, or a store copied without it) a writer may commit mid-load, so an unleased + // load stands only if its manifest is still current; if the manifest moved it reloads, under the lease the + // writer created. A page that fails under an unmoved manifest is corruption, not a race. + using var lease = AcquireCommitLease(wait: true, create: false); + var manifestBytes = ReadManifestBytes(); + var (manifest, _) = ParseManifest(manifestBytes); + trace($"manifest records={manifest.Records.Length}"); + try + { + if (lease == null) + UnleasedManifestRead?.Invoke(); + LoadRecordPages( + manifest.Records.Where(record => key == null || string.Equals(record.Key, key, StringComparison.Ordinal)) + .OrderBy(record => record.Key, StringComparer.Ordinal).ToArray(), + manifest.SchemaCatalog, + loaded, + reports, + foreign); + if (lease != null || !ManifestMoved(manifestBytes)) + { + trace($"indexed-pages loaded={loaded.Count}"); + break; + } + } + catch (Exception exception) when (lease == null && IsTornGeneration(exception)) + { + // The re-read stays out of the filter: an exception thrown inside a filter is swallowed as false. + if (!ManifestMoved(manifestBytes)) + throw; + if (attempt == UnleasedLoadAttempts) + throw Unsettled(exception); + continue; + } + + if (attempt == UnleasedLoadAttempts) + throw Unsettled(null); + } + } + + // The record at a key as its page holds it now, decoded fresh and stamped with VersionOf its manifest payload, the version + // CommitBatch compares. Null when none is there or it is foreign. Touches no entry, hydration, report or foreign list of the + // store, and raises nothing. + internal override CultStoredDocument? ReadDurable(string key) => Held(() => + { + var loaded = new Dictionary(StringComparer.Ordinal); + LoadManifestPages(key, loaded, new List(), new List(), _ => { }); + return loaded.TryGetValue(key, out var stored) ? stored : null; + }); + // Variants wait for the directory store's own cut; until then it refuses one rather than persist a resolved view as a plain page. private static void RefuseVariant(CultStoredDocument entry) { diff --git a/src/GameCult.Caching/CultCache.cs b/src/GameCult.Caching/CultCache.cs index dc0c8b99..303d3737 100644 --- a/src/GameCult.Caching/CultCache.cs +++ b/src/GameCult.Caching/CultCache.cs @@ -2682,34 +2682,66 @@ namespace GameCult.Caching } /// - /// Reads the record at , applies to a private copy of it, and commits the result only - /// if the record is still the one that was read; when another writer got there first it re-reads and runs - /// again, up to times. The write is durable when Update returns. + /// Reads the record at as its home store holds it, applies to a fresh copy of it, and + /// commits the result only if the record is still the one that was read; when another writer got there first it re-reads and runs + /// again, up to times. The write is durable when Update returns. /// /// - /// may run up to times. It gets a private copy (null when the key holds - /// nothing), which it may change and return. 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. The cache - /// needs a codec to copy with. Running out of attempts returns ; every other failure throws. + /// 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. /// public CultUpdateResult Update(CultRecordKey key, Func apply, int maxAttempts = 5) where T : class { if (apply == null) throw new ArgumentNullException(nameof(apply)); if (maxAttempts < 1) throw new ArgumentOutOfRangeException(nameof(maxAttempts), maxAttempts, "At least one attempt is required."); - var codec = _codec ?? throw new InvalidOperationException("This cache has no codec; create it with CultCacheMessagePack.Create."); for (var attempt = 1; ; attempt++) { - var (version, copy) = Held(() => + var (version, copy, source) = Held(() => { - if (!_entries.TryGetValue(key.Value, out var held)) - return ((string?)null, (T?)null); - if (held.Variant != null || held.Document is not T) + 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, (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), 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; + } + + if (durable == 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."); - return (held.StoredVersion, (T?)codec.Clone(held.Document, held.Descriptor.DocumentType)); + CultElementIds.Plan(durable.Descriptor.DocumentType, durable.Document, key.Value, deterministic: true).Apply(); + 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 { @@ -2728,9 +2760,11 @@ namespace GameCult.Caching return new CultUpdateResult(CultUpdateOutcome.Updated, next); if (outcome != CultCommitOutcome.Mismatch) throw new InvalidOperationException($"Update of {key.Value} was refused: {outcome}."); - Held(() => { Home(next.GetType())?.PullAll(); return true; }); if (attempt >= maxAttempts) + { + Held(() => { Home(next.GetType())?.PullAll(); return true; }); return new CultUpdateResult(CultUpdateOutcome.LostRace, Get(key)); + } } } @@ -3751,6 +3785,18 @@ namespace GameCult.Caching return changes; } + // Every store a record of this type, or of a registered subtype of it, can live in. + 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(); + } + // The most derived routed home type assignable from the document type, else the untyped store. private CacheBackingStore? Home(Type documentType) { @@ -4027,6 +4073,15 @@ namespace GameCult.Caching internal void UseRegistry(CultDocumentRegistry? registry) => _registry = registry; public abstract void PullAll(); + + /// + /// The record at as this store holds it now, decoded as a load decodes it by this call (a new instance nobody + /// else holds) and stamped with the version this store's commit compares. Null when the store holds nothing there or holds it foreign. + /// It changes none of the store's entries, staged writes or reports and raises nothing. Internal because only this library can stamp a version. + /// + internal virtual CultStoredDocument? ReadDurable(string key) => + throw new NotSupportedException($"{this} cannot read one record, which Update needs."); + public abstract void Push(CultStoredDocument entry); public abstract void Delete(CultStoredDocument entry); public void PushAll() => PushAll(soft: false); @@ -5173,6 +5228,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.OutsideStoreTests/HeldInstanceUpdateTests.cs b/tests/GameCult.Caching.OutsideStoreTests/HeldInstanceUpdateTests.cs new file mode 100644 index 00000000..6b02b72f --- /dev/null +++ b/tests/GameCult.Caching.OutsideStoreTests/HeldInstanceUpdateTests.cs @@ -0,0 +1,131 @@ +using System; +using System.Collections.Generic; +using System.IO; +using GameCult.Caching.MessagePack; +using MessagePack; +using NUnit.Framework; + +namespace GameCult.Caching.OutsideStoreTests +{ + // variants:follow_up:update-copy-from-stored-bytes: Update takes its version and its copy from one durable read of the home + // store, never from the held instance. Consumer side: public API only, so a change made in place to an instance the cache + // hands out can never be committed as if it had been read. + public class HeldInstanceUpdateTests + { + [CultDocument("tests.held_counter", "tests.held_counter.v1")] + [MessagePackObject] + public sealed class HeldCounter + { + [Key(0)] [CultName] public string Name { get; set; } = string.Empty; + [Key(1)] public int Value { get; set; } + } + + private static readonly CultDocumentRegistry Registry = CultDocumentRegistry.ForTypes(new[] { typeof(HeldCounter) }); + private static readonly CultRecordKey Key = new("counter"); + + private string _directory = string.Empty; + + [SetUp] + public void CreateDirectory() + { + _directory = Path.Combine(Path.GetTempPath(), $"cultlib-held-update-{Guid.NewGuid():N}"); + Directory.CreateDirectory(_directory); + } + + [TearDown] + public void DeleteDirectory() + { + if (Directory.Exists(_directory)) + Directory.Delete(_directory, recursive: true); + } + + private static CultCache Open(string path, bool directory) + { + var cache = new CultCache(Registry, CultCacheMessagePack.CreateCodec(Registry)); + cache.AddBackingStore(directory ? new DirectoryMessagePackBackingStore(path) : new SingleFileMessagePackBackingStore(path)); + cache.PullAllBackingStoresAsync().GetAwaiter().GetResult(); + return cache; + } + + private string Seed(bool directory, int value) + { + var path = Path.Combine(_directory, "held.cc"); + using var cache = Open(path, directory); + Assert.That(cache.Commit(batch => batch.Upsert(new HeldCounter { Name = "counter", Value = value }, new CultRecordHandle(Key))), Is.True); + return path; + } + + private static int StoredValue(string path, bool directory) + { + using var reader = Open(path, directory); + return reader.Get(Key)!.Value; + } + + private static HeldCounter Plus(HeldCounter? counter, int by) + { + counter!.Value += by; + return counter; + } + + [TestCase(false)] + [TestCase(true)] + public void InPlaceChangeToHeldInstanceIsNeitherAppliedNorPersisted(bool directory) + { + var path = Seed(directory, 1); + using var cache = Open(path, directory); + cache.Get(Key)!.Value = 999; + var seen = new List(); + + var result = cache.Update(Key, counter => { seen.Add(counter!.Value); return Plus(counter, 1); }); + + Assert.That(result.Outcome, Is.EqualTo(CultUpdateOutcome.Updated)); + Assert.That(seen, Is.EqualTo(new[] { 1 }), "apply saw the in-place change"); + Assert.That(result.Document!.Value, Is.EqualTo(2)); + Assert.That(StoredValue(path, directory), Is.EqualTo(2), "the in-place 999 was persisted"); + } + + [TestCase(false)] + [TestCase(true)] + public void InPlaceChangeToAPreviousResultIsNotCommitted(bool directory) + { + var path = Seed(directory, 1); + using var cache = Open(path, directory); + var first = cache.Update(Key, counter => Plus(counter, 1)); + Assert.That(first.Outcome, Is.EqualTo(CultUpdateOutcome.Updated)); + first.Document!.Value = 999; + var seen = new List(); + + var second = cache.Update(Key, counter => { seen.Add(counter!.Value); return Plus(counter, 1); }); + + Assert.That(second.Outcome, Is.EqualTo(CultUpdateOutcome.Updated)); + Assert.That(seen, Is.EqualTo(new[] { 2 }), "apply ran more than once or saw the in-place change"); + Assert.That(StoredValue(path, directory), Is.EqualTo(3)); + } + + [TestCase(false)] + [TestCase(true)] + public void DeclineAfterInPlaceChangeWritesNothing(bool directory) + { + var path = Seed(directory, 1); + using var cache = Open(path, directory); + cache.Get(Key)!.Value = 999; + var before = Snapshot(path); + + Assert.That(cache.Update(Key, _ => null).Outcome, Is.EqualTo(CultUpdateOutcome.Declined)); + + Assert.That(Snapshot(path), Is.EqualTo(before)); + Assert.That(StoredValue(path, directory), Is.EqualTo(1)); + } + + // Every file of the store: the single file, or the manifest and its pages. + private static Dictionary Snapshot(string path) + { + var files = new Dictionary { [path] = File.ReadAllBytes(path) }; + var pages = path + ".records"; + if (Directory.Exists(pages)) + foreach (var file in Directory.GetFiles(pages, "*", SearchOption.AllDirectories)) + files[file] = File.ReadAllBytes(file); + return files; + } + } +} diff --git a/tests/GameCult.Caching.OutsideStoreTests/OutsideStoreJudgeTests.cs b/tests/GameCult.Caching.OutsideStoreTests/OutsideStoreJudgeTests.cs index bd2fe047..86949fa0 100644 --- a/tests/GameCult.Caching.OutsideStoreTests/OutsideStoreJudgeTests.cs +++ b/tests/GameCult.Caching.OutsideStoreTests/OutsideStoreJudgeTests.cs @@ -364,5 +364,18 @@ namespace GameCult.Caching.OutsideStoreTests Assert.That(() => cache.Commit(batch => batch.Remove(K)), Throws.InstanceOf(), "removal"); Assert.That(disk.Records["k"], Is.EqualTo(before)); } + + // Only the library can stamp a version on a decoded record, so a store kind outside it cannot read one honestly: Update refuses before apply. + [Test] + public void UpdateOverAnOutsideStoreIsNotSupported() + { + var disk = new Disk(); + PutOnDisk(disk, AB, "k", new OutA { Name = "k", Value = 1 }); + using var cache = Open(disk, AB); + var calls = 0; + + Assert.That(() => cache.Update(K, a => { calls++; return a; }), Throws.TypeOf()); + Assert.That(calls, Is.EqualTo(0), "apply ran for a store that cannot serve Update"); + } } } diff --git a/tests/GameCult.Caching.Tests/ElementIdTests.cs b/tests/GameCult.Caching.Tests/ElementIdTests.cs index a6c68c94..d40ffe25 100644 --- a/tests/GameCult.Caching.Tests/ElementIdTests.cs +++ b/tests/GameCult.Caching.Tests/ElementIdTests.cs @@ -224,6 +224,28 @@ namespace GameCult.Caching.Tests Assert.That(AllIds(reread), Is.EqualTo(first), "the ids minted at load are the ids written, and stay through repeated writes"); } + // Update reads the store's bytes, which hold no ids; it mints the ids the load minted, so the write keeps the ids the cache showed. + [Test] + public void UpdateOfAPreIdRecordPersistsTheIdsTheLoadMinted() + { + var path = PathOf("old.cc"); + WritePreIdStore(path, null, ("a", "a", 2)); + var key = new CultRecordKey("a"); + + string[] shown; + using (var cache = Open(path)) + { + shown = AllIds(cache.Get(key)!); + Assert.That(shown, Has.All.Not.Empty); + var result = cache.Update(key, deck => { deck!.Name = "a2"; return deck; }); + Assert.That(result.Outcome, Is.EqualTo(CultUpdateOutcome.Updated)); + } + + var written = MessagePackSerializer.Deserialize(DiskRecord(path, "a").Payload); + Assert.That(written.Name, Is.EqualTo("a2")); + Assert.That(AllIds(written), Is.EqualTo(shown), "the write replaced the ids the cache showed with random ones"); + } + [Test] public void MintElementIdsRewritesEveryPreIdRecordAndNothingElse() { diff --git a/tests/GameCult.Caching.Tests/UpdateTests.cs b/tests/GameCult.Caching.Tests/UpdateTests.cs index 0dd96bad..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; @@ -78,16 +78,22 @@ namespace GameCult.Caching.Tests using var a = Open(path, directory); using var b = Open(path, directory); var applied = 0; + // Each round starts both writers together: an attempt reads fresh from the store, so only overlapping read-commit windows lose. + var round = new Barrier(2); var racers = new[] { a, b }.Select(cache => Task.Run(() => { var results = new List(); for (var i = 0; i < n; i++) + { + round.SignalAndWait(); results.Add(cache.Update(Key, counter => { Interlocked.Increment(ref applied); return Plus(counter, 1); }, maxAttempts: 200).Outcome); + } + return results; })).ToArray(); Assert.That(Task.WaitAll(racers, TimeSpan.FromMinutes(2)), Is.True, "the racers did not finish"); @@ -172,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() { @@ -219,8 +412,19 @@ namespace GameCult.Caching.Tests var calls = 0; Func apply = counter => { calls++; return Plus(counter, 1); }; - using (var codecLess = Open(path, false, codec: false)) - Assert.That(() => codecLess.Update(Key, apply), Throws.InvalidOperationException); + // A cache with no store copies its held instance, so it needs a codec. + using (var storeLess = new CultCache(Registry)) + { + storeLess.Commit(batch => batch.Upsert(new UpdateCounter { Name = "counter", Value = 1 }, new CultRecordHandle(Key))); + Assert.That(() => storeLess.Update(Key, apply), Throws.InvalidOperationException.With.Message.Contains("no codec")); + } + + // Stores are attached, but none is home to the type: refused before apply. + using (var noHome = new CultCache(Registry, CultCacheMessagePack.CreateCodec(Registry))) + { + noHome.AddBackingStore(new SingleFileMessagePackBackingStore(PathOf("nohome.cc")), typeof(UpdateTally)); + Assert.That(() => noHome.Update(Key, apply), Throws.InvalidOperationException.With.Message.Contains("names its home store")); + } using (var cache = Open(path, false)) { Assert.That(() => cache.Update(Key, apply, maxAttempts: 0), Throws.TypeOf()); @@ -241,7 +445,7 @@ namespace GameCult.Caching.Tests before = File.ReadAllBytes(path); var gearCalls = 0; using (var cache = Open(path, false)) - Assert.That(() => cache.Update(new CultRecordKey("gear-big"), gear => { gearCalls++; return gear; }), Throws.InvalidOperationException); + Assert.That(() => cache.Update(new CultRecordKey("gear-big"), gear => { gearCalls++; return gear; }), Throws.InvalidOperationException.With.Message.Contains("Update cannot copy it")); Assert.That(gearCalls, Is.EqualTo(0), "apply ran for a variant"); Assert.That(calls, Is.EqualTo(0), "apply ran for a refused call"); Assert.That(File.ReadAllBytes(path), Is.EqualTo(before)); @@ -479,6 +683,73 @@ namespace GameCult.Caching.Tests Assert.That(cache.Get(Key)!.Value, Is.EqualTo(200)); } + [TestCase(false)] + [TestCase(true)] + public void StaleCacheFirstAttemptSeesTheStore(bool directory) + { + var path = PathOf("stale.cc"); + Seed(path, directory, 1); + using var a = Open(path, directory); + using var b = Open(path, directory); + Assert.That(b.Get(Key)!.Value, Is.EqualTo(1)); + Rival(a, 7); + var seen = new List(); + + var result = b.Update(Key, counter => { seen.Add(counter!.Value); return Plus(counter, 1); }); + + Assert.That(result.Outcome, Is.EqualTo(CultUpdateOutcome.Updated)); + Assert.That(seen, Is.EqualTo(new[] { 7 }), "the first attempt read the cache's stale copy"); + Assert.That(StoredValue(path, directory), Is.EqualTo(8)); + } + + // The durable read is not a load: it leaves the store's view of what the cache holds alone, so a later pull still delivers the rival's write. + [TestCase(false)] + [TestCase(true)] + public void ADeclinedUpdateDoesNotHideTheRivalsWriteFromALaterPull(bool directory) + { + var path = PathOf("hidden.cc"); + Seed(path, directory, 1); + using var a = Open(path, directory); + using var b = Open(path, directory); + Assert.That(b.Get(Key)!.Value, Is.EqualTo(1)); + Rival(a, 7); + + Assert.That(b.Update(Key, _ => null).Outcome, Is.EqualTo(CultUpdateOutcome.Declined)); + Assert.That(b.Get(Key)!.Value, Is.EqualTo(1), "a declined Update changed what the cache holds"); + b.PullAllBackingStoresAsync().GetAwaiter().GetResult(); + + Assert.That(b.Get(Key)!.Value, Is.EqualTo(7)); + } + + [Test] + public void CodecLessStoreBackedCacheUpdates() + { + var path = PathOf("codecless.cc"); + Seed(path, false, 1); + var calls = 0; + using (var cache = Open(path, false, codec: false)) + Assert.That(cache.Update(Key, counter => { calls++; return Plus(counter, 1); }).Outcome, Is.EqualTo(CultUpdateOutcome.Updated)); + + Assert.That(calls, Is.EqualTo(1)); + Assert.That(StoredValue(path, false), Is.EqualTo(2)); + } + + [TestCase(false)] + [TestCase(true)] + public void AStoreHoldingNothingAtTheKeyIsAbsentEvenIfTheCacheHoldsIt(bool directory) + { + var path = PathOf("gone.cc"); + Seed(path, directory, 5); + using var a = Open(path, directory); + using var b = Open(path, directory); + Assert.That(b.Get(Key), Is.Not.Null); + Assert.That(a.Commit(batch => batch.Remove(Key)), Is.True); + var seen = new List(); + + Assert.That(b.Update(Key, counter => { seen.Add(counter); return null; }).Outcome, Is.EqualTo(CultUpdateOutcome.Declined)); + Assert.That(seen, Is.EqualTo(new UpdateCounter?[] { null })); + } + [CultDocument("tests.update_counter", "tests.update_counter.v1")] [MessagePackObject(AllowPrivate = true)] internal sealed class UpdateCounter @@ -511,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; + } } }