diff --git a/src/GameCult.Caching.MessagePack/CultDocumentMessagePackSerialization.cs b/src/GameCult.Caching.MessagePack/CultDocumentMessagePackSerialization.cs index 23c889f5..55fd7584 100644 --- a/src/GameCult.Caching.MessagePack/CultDocumentMessagePackSerialization.cs +++ b/src/GameCult.Caching.MessagePack/CultDocumentMessagePackSerialization.cs @@ -802,9 +802,9 @@ public class SingleFileMessagePackBackingStore : SingleFileBackingStore return store; } - protected override byte[] SerializePayload(object document) + protected override byte[] SerializePayload(object document, Type documentType) { - return CultDocumentMessagePackSerialization.SerializeUntyped(document, document.GetType(), Registry); + return CultDocumentMessagePackSerialization.SerializeUntyped(document, documentType, Registry); } protected override object DeserializePayload(Type documentType, byte[] payload) diff --git a/src/GameCult.Caching/CultCache.cs b/src/GameCult.Caching/CultCache.cs index e94f6ac3..180d1d90 100644 --- a/src/GameCult.Caching/CultCache.cs +++ b/src/GameCult.Caching/CultCache.cs @@ -3591,6 +3591,13 @@ namespace GameCult.Caching protected void Held(Action body) => Held(() => { body(); return true; }); + private protected CultDocumentRegistry? SwapRegistry(CultDocumentRegistry? registry) + { + var previous = _registry; + _registry = registry; + return previous; + } + internal void AttachRegistry(CultDocumentRegistry registry) { _registry = registry ?? throw new ArgumentNullException(nameof(registry)); @@ -4041,7 +4048,7 @@ namespace GameCult.Caching _catalog.TryGetValue(entry, out var bytes) ? bytes : _owner.SerializeCatalogEntry(entry); } - protected abstract byte[] SerializePayload(object document); + protected abstract byte[] SerializePayload(object document, Type documentType); protected abstract object DeserializePayload(Type documentType, byte[] payload); public override string ToString() => FileInfo.FullName; @@ -4212,7 +4219,7 @@ namespace GameCult.Caching { if (!catalogEntries.TryGetValue(document.Descriptor, out var catalogEntry)) catalogEntries[document.Descriptor] = catalogEntry = document.Descriptor.ToCatalogEntry(); - return new CultStagedRecord(ToPersistedRecord(document, SerializePayload), catalogEntry, document.HoldsIds); + return new CultStagedRecord(ToPersistedRecord(document, payload => SerializePayload(payload, payload.GetType())), catalogEntry, document.HoldsIds); }).ToList(); return (upserts, removals.ToList()); } @@ -4228,7 +4235,7 @@ namespace GameCult.Caching /// The record to store. /// The schema catalog entry for the record's schema id. /// Whether the record's payload holds element ids, or null when the caller cannot tell. - /// The registry the payload was encoded with. Given, the write decides the header for copied records that registry can read; omitted, it decides nothing for them. + /// The registry the payload was encoded with. Given, the write decides the header for copied records that registry can read; omitted, it decides nothing for them. It serves this call only: the store keeps no registry. /// The store is attached to a cache; write through the cache. /// The file at the path is not a store this runtime can read. public void WriteRecord(CultPersistedRecord record, CultSchemaCatalogEntry catalogEntry, bool? holdsIds = null, CultDocumentRegistry? registry = null) @@ -4236,27 +4243,22 @@ namespace GameCult.Caching if (record == null) throw new ArgumentNullException(nameof(record)); if (catalogEntry == null) throw new ArgumentNullException(nameof(catalogEntry)); RefuseWhenAttached(); - if (registry != null) - AttachRegistry(registry); - Held(() => - { - using (AcquireLock(wait: true)) - _ = ApplyWriteSet(ReadSnapshot(), new[] { new CultStagedRecord(record, catalogEntry, holdsIds) }, Array.Empty(), Array.Empty()); - }); + ForTheCall(registry, () => WriteOne(record, catalogEntry, holdsIds)); } /// /// Writes one typed document to the store file without a cache, as a cache write would. Element ids are decided before anything /// is written: an unset id is minted, a list a cache write refuses refuses the write, and the file is marked as holding element - /// ids when the document holds one. A write that fails gives the minted ids back. Then the document is encoded with - /// and stored as stores a record: every other record and catalog entry - /// is copied forward as stored, under the store's lock, and the file is replaced atomically. + /// ids when the document holds one. A write that fails gives the minted ids back. Then the document is encoded as + /// , the type its schema id and ids are taken from, with , and stored as + /// stores a record: every other record and catalog entry is copied forward as stored, under the store's + /// lock, and the file is replaced atomically. /// - /// The registered document type. + /// The registered document type. A derived instance is stored as this type. /// The record key the document is stored under. /// The document to store. /// The stored-at stamp, or null for now. - /// The registry that describes and encodes the document, or null for . + /// The registry that describes and encodes the document, or null for . It serves this call only: the store keeps no registry. /// The store is attached to a cache; write through the cache. /// The file at the path is not a store this runtime can read. /// The document holds an element id or list a cache write refuses. @@ -4267,49 +4269,51 @@ namespace GameCult.Caching RefuseWhenAttached(); var resolvedRegistry = registry ?? CultDocumentRegistry.Shared; var catalogEntry = resolvedRegistry.GetRequired().ToCatalogEntry(); - AttachRegistry(resolvedRegistry); - var ids = CultElementIds.Plan(typeof(TDocument), document, key.Value, deterministic: false); - ids.Apply(); - try + ForTheCall(resolvedRegistry, () => { - WriteRecord( - new CultPersistedRecord - { - Key = key.Value, - SchemaId = catalogEntry.SchemaId, - StoredAt = string.IsNullOrWhiteSpace(storedAt) ? DateTimeOffset.UtcNow.ToString("O") : storedAt!, - Payload = SerializePayload(document) - }, - catalogEntry, - CultElementIds.Holds(typeof(TDocument), document)); - } - catch - { - ids.Undo(); - throw; - } + var ids = CultElementIds.Plan(typeof(TDocument), document, key.Value, deterministic: false); + ids.Apply(); + try + { + WriteOne( + new CultPersistedRecord + { + Key = key.Value, + SchemaId = catalogEntry.SchemaId, + StoredAt = string.IsNullOrWhiteSpace(storedAt) ? DateTimeOffset.UtcNow.ToString("O") : storedAt!, + Payload = SerializePayload(document, typeof(TDocument)) + }, + catalogEntry, + CultElementIds.Holds(typeof(TDocument), document)); + } + catch + { + ids.Undo(); + throw; + } + }); } - /// - /// Decodes a document payload read from this store file, as a cache load decodes it: the ids a load mints are minted from the key - /// and their path, so every reader of one store mints the same ids, and a payload a cache write would refuse is refused here too. - /// - /// The registered document type. - /// The record key the payload was stored under. - /// The record's payload. - /// The registry that describes and decodes the document, or null for . - /// The store is attached to a cache; read through the cache. - /// The payload holds an element id or list a cache load refuses. - public TDocument DecodeDocument(CultRecordKey key, byte[] payload, CultDocumentRegistry? registry = null) - where TDocument : class + // The registry is the call's argument, not store state: it is attached for the call under the store's gate and the store's + // previous registry (none, for a store no cache ever attached) comes back whatever the call does. + private void ForTheCall(CultDocumentRegistry? registry, Action body) => + Held(() => + { + var previous = SwapRegistry(registry); + try + { + body(); + } + finally + { + SwapRegistry(previous); + } + }); + + private void WriteOne(CultPersistedRecord record, CultSchemaCatalogEntry catalogEntry, bool? holdsIds) { - if (payload == null) throw new ArgumentNullException(nameof(payload)); - if (Cache != null) - throw new InvalidOperationException("The store is attached to a cache, read through the cache."); - AttachRegistry(registry ?? CultDocumentRegistry.Shared); - var document = (TDocument)DeserializePayload(typeof(TDocument), payload); - CultElementIds.Plan(typeof(TDocument), document, key.Value, deterministic: true).Apply(); - return document; + using (AcquireLock(wait: true)) + _ = ApplyWriteSet(ReadSnapshot(), new[] { new CultStagedRecord(record, catalogEntry, holdsIds) }, Array.Empty(), Array.Empty()); } private void RefuseWhenAttached() diff --git a/src/GameCult.Caching/CultVariants.cs b/src/GameCult.Caching/CultVariants.cs index ee5cc523..48909c19 100644 --- a/src/GameCult.Caching/CultVariants.cs +++ b/src/GameCult.Caching/CultVariants.cs @@ -28,6 +28,24 @@ namespace GameCult.Caching // (document payload, slot and encoded value pairs) public Func>, byte[]> Overlay { get; } + + /// + /// Decodes a document payload read from a store file, as a cache load decodes it: the ids a load mints are minted from the key + /// and their path, so every reader of one store mints the same ids, and a payload a cache write would refuse is refused here too. + /// It reads no file and holds no state. + /// + /// The registered document type. + /// The record key the payload was stored under. + /// The record's payload. + /// The payload holds an element id or list a cache load refuses. + public TDocument DecodeDocument(CultRecordKey key, byte[] payload) + where TDocument : class + { + if (payload == null) throw new ArgumentNullException(nameof(payload)); + var document = (TDocument)Deserialize(typeof(TDocument), typeof(TDocument), payload)!; + CultElementIds.Plan(typeof(TDocument), document, key.Value, deterministic: true).Apply(); + return document; + } } public enum CultOverrideOp diff --git a/src/GameCult.Mesh/CultMeshSingleFileDocuments.cs b/src/GameCult.Mesh/CultMeshSingleFileDocuments.cs index 39a087ba..37de5226 100644 --- a/src/GameCult.Mesh/CultMeshSingleFileDocuments.cs +++ b/src/GameCult.Mesh/CultMeshSingleFileDocuments.cs @@ -112,7 +112,7 @@ namespace GameCult.Mesh var descriptor = resolvedRegistry.GetRequired(); var payload = ReadSingleFileDocumentPayload(path, key, descriptor.SchemaId); // A load refuses what a write refuses, and mints the ids a cache load of the same file would mint. - return StoreAt(path).DecodeDocument(key, payload, resolvedRegistry); + return CultCacheMessagePack.CreateCodec(resolvedRegistry).DecodeDocument(key, payload); } /// diff --git a/tests/GameCult.Caching.ConsumerBuild/Program.cs b/tests/GameCult.Caching.ConsumerBuild/Program.cs index 3280b35f..60376068 100644 --- a/tests/GameCult.Caching.ConsumerBuild/Program.cs +++ b/tests/GameCult.Caching.ConsumerBuild/Program.cs @@ -19,7 +19,7 @@ if (!committed) return 1; // A store file is written and read without a cache through the public members: this project has no InternalsVisibleTo, so -// building it pins that ReadStoreSnapshot, WriteRecord, WriteDocument and DecodeDocument are public API. +// building it pins that ReadStoreSnapshot, WriteRecord and WriteDocument, and CultCodec.DecodeDocument, are public API. var storePath = System.IO.Path.Combine(System.IO.Path.GetTempPath(), "consumer-build-" + System.Guid.NewGuid().ToString("N") + ".cc"); try { @@ -32,7 +32,7 @@ try var typedRegistry = CultDocumentRegistry.ForTypes(new[] { typeof(ConsumerFaction) }); store.WriteDocument(new CultRecordKey("f"), new ConsumerFaction { Name = "typed" }, registry: typedRegistry); var typed = store.ReadStoreSnapshot()!.Records.Single(record => record.Key == "f"); - if (store.DecodeDocument(new CultRecordKey("f"), typed.Payload, typedRegistry).Name != "typed") + if (GameCult.Caching.MessagePack.CultCacheMessagePack.CreateCodec(typedRegistry).DecodeDocument(new CultRecordKey("f"), typed.Payload).Name != "typed") return 1; } finally diff --git a/tests/GameCult.Caching.Tests/CachelessSingleFileTests.cs b/tests/GameCult.Caching.Tests/CachelessSingleFileTests.cs index 0997b969..ea153a96 100644 --- a/tests/GameCult.Caching.Tests/CachelessSingleFileTests.cs +++ b/tests/GameCult.Caching.Tests/CachelessSingleFileTests.cs @@ -2,14 +2,16 @@ using System; using System.IO; using System.Linq; +using System.Threading; using System.Threading.Tasks; using GameCult.Caching.MessagePack; +using MessagePack; using NUnit.Framework; using static GameCult.Caching.Tests.ElementIdTests; namespace GameCult.Caching.Tests { - // A single-file store file is read and written without a cache through ReadStoreSnapshot, WriteRecord, WriteDocument and DecodeDocument: the same reader, lock, + // A single-file store file is read and written without a cache through ReadStoreSnapshot, WriteRecord and WriteDocument: the same reader, lock, // atomic replace and header rule a cache's store uses. CultMesh's single-file document helpers are built on these members. public class CachelessSingleFileTests { @@ -33,6 +35,136 @@ namespace GameCult.Caching.Tests Directory.Delete(_directory, true); } + private static CultCodec Codec(CultDocumentRegistry? registry = null) => + CultCacheMessagePack.CreateCodec(registry ?? CultDocumentRegistry.Shared); + + [CultDocument("tests.cacheless.base", "tests.cacheless.base.v1")] + [MessagePackObject] + public class DeclaredBase + { + [Key(0)] [CultName] public string Name { get; set; } = ""; + } + + [MessagePackObject] + public sealed class UndeclaredDerived : DeclaredBase + { + [Key(1)] public string Extra { get; set; } = ""; + } + + // Reads the store file as a SingleFileMessagePackBackingStore does, and holds the first reader inside ReadStore so a test can + // see whether a second one gets in beside it. + private sealed class GatedStore : SingleFileBackingStore + { + private int _held; + private int _inside; + private int _mostInside; + public readonly ManualResetEventSlim Entered = new(); + public readonly ManualResetEventSlim Release = new(); + + public GatedStore(string path) : base(path) { } + + public int MostInside => Volatile.Read(ref _mostInside); + + protected override byte[] SerializeStore(string formatVersion, System.Collections.Generic.IReadOnlyList catalogEntries, System.Collections.Generic.IReadOnlyList records) => + CultDocumentMessagePackSerialization.SerializeStore(formatVersion, catalogEntries, records); + + protected override byte[] SerializeRecord(CultPersistedRecord record) => + CultDocumentMessagePackSerialization.SerializePersistedRecord(record); + + protected override byte[] SerializeCatalogEntry(CultSchemaCatalogEntry entry) => + CultDocumentMessagePackSerialization.SerializeCatalogEntry(entry); + + protected override (CultPersistedStoreSnapshot Snapshot, byte[][] CatalogBytes, byte[][] RecordBytes) ReadStore(byte[] data) + { + var inside = Interlocked.Increment(ref _inside); + int seen; + while ((seen = MostInside) < inside && Interlocked.CompareExchange(ref _mostInside, inside, seen) != seen) { } + Entered.Set(); + try + { + if (Interlocked.Exchange(ref _held, 1) == 0) + Release.Wait(TimeSpan.FromSeconds(30)); + var store = CultDocumentMessagePackSerialization.ReadStore(data); + CultDocumentMessagePackSerialization.RequireSingleFileFormat(store.Snapshot); + return store; + } + finally + { + Interlocked.Decrement(ref _inside); + } + } + + protected override byte[] SerializePayload(object document, Type documentType) => + CultDocumentMessagePackSerialization.SerializeUntyped(document, documentType, Registry); + + protected override object DeserializePayload(Type documentType, byte[] payload) => + CultDocumentMessagePackSerialization.DeserializeUntyped(documentType, payload, Registry); + } + + // ReadStoreSnapshot takes the store's gate: while one read is inside the reader, a second does not enter beside it. + [Test] + public async Task ReadStoreSnapshotTakesTheStoresGate() + { + new SingleFileMessagePackBackingStore(_path).WriteRecord(Record("a", One, 1), One); + var store = new GatedStore(_path); + + var first = Task.Run(() => store.ReadStoreSnapshot()); + Assert.That(store.Entered.Wait(TimeSpan.FromSeconds(10)), Is.True, "the first read reached the reader"); + var second = Task.Run(() => store.ReadStoreSnapshot()); + await Task.Delay(500); + var mostInside = store.MostInside; + store.Release.Set(); + await Task.WhenAll(first, second); + + Assert.That(mostInside, Is.EqualTo(1), "a second read waits at the gate while the first is inside the reader"); + Assert.That(first.Result!.Records, Has.Length.EqualTo(1)); + Assert.That(second.Result!.Records, Has.Length.EqualTo(1)); + } + + // WriteDocument stores the document as T: the schema id, the id plan, the header and the bytes all come from the one + // declared type, as they did when CultMesh encoded typeof(T). + [Test] + public void WriteDocumentEncodesTheDeclaredTypeNotTheRuntimeType() + { + var registry = CultDocumentRegistry.ForTypes(new[] { typeof(DeclaredBase) }); + + new SingleFileMessagePackBackingStore(_path).WriteDocument(new CultRecordKey("d"), new UndeclaredDerived { Name = "n", Extra = "x" }, registry: registry); + + var record = new SingleFileMessagePackBackingStore(_path).ReadStoreSnapshot()!.Records.Single(); + var asBase = CultDocumentMessagePackSerialization.SerializeUntyped(new DeclaredBase { Name = "n" }, typeof(DeclaredBase), registry); + Assert.That(record.SchemaId, Is.EqualTo(registry.GetRequired().SchemaId)); + Assert.That(record.Payload, Is.EqualTo(asBase), "base bytes under the base schema: one slot, not the derived type's two"); + } + + // The registry is the call's argument. A typed call on a store instance leaves nothing behind, so a later WriteRecord that + // gives none decides the header as a fresh instance does. + [Test] + public async Task ATypedCallLeavesNoRegistryOnTheStore() + { + var registry = CultDocumentRegistry.ForTypes(new[] { typeof(IdDeck) }); + using (var cache = new CultCache(registry)) + { + cache.AddBackingStore(new SingleFileMessagePackBackingStore(_path)); + await cache.AddAsync(new IdDeck { Name = "deck", Reels = { new IdReel { Label = "reel" } } }); + cache.FlushAllBackingStores(); + } + void Unmark() + { + var written = CultDocumentMessagePackSerialization.DeserializeSnapshot(File.ReadAllBytes(_path)); + written.FormatVersion = CultPersistedStoreSnapshot.FormatV1; + File.WriteAllBytes(_path, CultDocumentMessagePackSerialization.SerializeSnapshot(written)); + } + Unmark(); + var store = new SingleFileMessagePackBackingStore(_path); + + store.WriteDocument(new CultRecordKey("again"), new IdDeck { Name = "again" }, registry: registry); + Assert.That(store.ReadStoreSnapshot()!.FormatVersion, Is.EqualTo(CultPersistedStoreSnapshot.FormatV3), "the typed call, given a registry, read the copied ids"); + Unmark(); + store.WriteRecord(Record("other", One, 1), One, holdsIds: false, registry: null); + + Assert.That(store.ReadStoreSnapshot()!.FormatVersion, Is.EqualTo(CultPersistedStoreSnapshot.FormatV1), "a write given no registry decides nothing for copied records"); + } + private static CultPersistedRecord Record(string key, CultSchemaCatalogEntry entry, params byte[] payload) => new() { Key = key, SchemaId = entry.SchemaId, StoredAt = "2026-10-22T00:00:00.0000000+00:00", Payload = payload }; @@ -124,7 +256,7 @@ namespace GameCult.Caching.Tests { var store = new SingleFileMessagePackBackingStore(_path); Assert.Throws(() => store.WriteDocument(new CultRecordKey("deck"), null!)); - Assert.Throws(() => store.DecodeDocument(new CultRecordKey("deck"), null!)); + Assert.Throws(() => Codec().DecodeDocument(new CultRecordKey("deck"), null!)); } [Test] @@ -158,9 +290,8 @@ namespace GameCult.Caching.Tests Assert.That(Written("one.cc"), Is.Not.EqualTo(Written("two.cc")), "a write mints random ids"); var unset = CultDocumentMessagePackSerialization.SerializeUntyped(new IdDeck { Name = "deck", Reels = { new IdReel { Label = "reel" } } }, typeof(IdDeck), registry); - var store = new SingleFileMessagePackBackingStore(_path); - var first = store.DecodeDocument(new CultRecordKey("deck"), unset, registry).Reels[0].Id; - var second = new SingleFileMessagePackBackingStore(_path).DecodeDocument(new CultRecordKey("deck"), unset, registry).Reels[0].Id; + var first = Codec(registry).DecodeDocument(new CultRecordKey("deck"), unset).Reels[0].Id; + var second = Codec(registry).DecodeDocument(new CultRecordKey("deck"), unset).Reels[0].Id; Assert.That(first, Does.Match("^[0-9a-f]{12}$")); Assert.That(second, Is.EqualTo(first), "a load mints from the key and path, so every reader mints the same id"); } @@ -222,7 +353,7 @@ namespace GameCult.Caching.Tests Assert.That(snapshot.FormatVersion, Is.EqualTo(CultPersistedStoreSnapshot.FormatV3)); Assert.That(snapshot.Records.Select(record => record.Key), Is.EquivalentTo(new[] { "other", "deck" })); Assert.That(deck.Reels[0].Id, Does.Match("^[0-9a-f]{12}$"), "the id the file holds was minted on the document"); - var read = new SingleFileMessagePackBackingStore(_path).DecodeDocument(new CultRecordKey("deck"), snapshot.Records.Single(record => record.Key == "deck").Payload, registry); + var read = Codec(registry).DecodeDocument(new CultRecordKey("deck"), snapshot.Records.Single(record => record.Key == "deck").Payload); Assert.That(read.Reels[0].Id, Is.EqualTo(deck.Reels[0].Id)); } @@ -240,7 +371,7 @@ namespace GameCult.Caching.Tests } [Test] - public void WriteDocumentAndDecodeDocumentRefuseOnAStoreAttachedToACache() + public void WriteDocumentRefusesOnAStoreAttachedToACache() { var registry = CultDocumentRegistry.ForTypes(new[] { typeof(IdDeck) }); var store = new SingleFileMessagePackBackingStore(_path); @@ -248,7 +379,6 @@ namespace GameCult.Caching.Tests cache.AddBackingStore(store); Assert.That(Assert.Throws(() => store.WriteDocument(new CultRecordKey("deck"), new IdDeck { Name = "d" }))!.Message, Does.Contain("write through the cache")); - Assert.That(Assert.Throws(() => store.DecodeDocument(new CultRecordKey("deck"), new byte[] { 0x90 }))!.Message, Does.Contain("read through the cache")); Assert.That(File.Exists(_path), Is.False); } } diff --git a/tests/GameCult.Caching.Tests/PreCut2StoreFormatTests.cs b/tests/GameCult.Caching.Tests/PreCut2StoreFormatTests.cs index 4580a185..df7ce6a4 100644 --- a/tests/GameCult.Caching.Tests/PreCut2StoreFormatTests.cs +++ b/tests/GameCult.Caching.Tests/PreCut2StoreFormatTests.cs @@ -372,8 +372,8 @@ namespace GameCult.Caching.Tests return store; } - protected override byte[] SerializePayload(object document) => - CultDocumentMessagePackSerialization.SerializeUntyped(document, document.GetType(), Registry); + protected override byte[] SerializePayload(object document, Type documentType) => + CultDocumentMessagePackSerialization.SerializeUntyped(document, documentType, Registry); protected override object DeserializePayload(Type documentType, byte[] payload) => CultDocumentMessagePackSerialization.DeserializeUntyped(documentType, payload, Registry);