diff --git a/src/GameCult.Caching.MessagePack/CultDocumentMessagePackSerialization.cs b/src/GameCult.Caching.MessagePack/CultDocumentMessagePackSerialization.cs index df7302de..b66ed716 100644 --- a/src/GameCult.Caching.MessagePack/CultDocumentMessagePackSerialization.cs +++ b/src/GameCult.Caching.MessagePack/CultDocumentMessagePackSerialization.cs @@ -548,6 +548,12 @@ public class SingleFileMessagePackBackingStore : SingleFileBackingStore return snapshot; } + protected override string? ReadFormatVersion(byte[] data) + { + var reader = new MessagePackReader(data); + return reader.ReadArrayHeader() > 0 ? reader.ReadString() : null; + } + protected override byte[] SerializePayload(object document) { return CultDocumentMessagePackSerialization.SerializeUntyped(document, document.GetType(), Registry); diff --git a/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs b/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs index 61e65208..436ec4f4 100644 --- a/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs +++ b/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs @@ -449,6 +449,14 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore } var snapshot = CultDocumentMessagePackSerialization.DeserializeSnapshot(manifestBytes); + if (CultPersistedStoreSnapshot.FormatNumber(snapshot.FormatVersion) is int number && number > 4) + { + // Every write path parses the manifest under its commit lease before it writes a page or the manifest, so a + // store another process has since stamped newer is refused here, untouched. + throw new NotSupportedException( + $"Directory store {_manifestFile.FullName} is {snapshot.FormatVersion}, a newer format than this runtime reads ({IndexedFormatVersion})."); + } + if (!string.Equals(snapshot.FormatVersion, IndexedFormatVersion, StringComparison.Ordinal)) { throw new InvalidOperationException( diff --git a/src/GameCult.Caching/CultCache.cs b/src/GameCult.Caching/CultCache.cs index 124bea21..f4ebe368 100644 --- a/src/GameCult.Caching/CultCache.cs +++ b/src/GameCult.Caching/CultCache.cs @@ -66,6 +66,24 @@ namespace GameCult.Caching public const string FormatV2 = "cultcache.store.v2"; public string FormatVersion { get; set; } = FormatV1; + + /// + /// The N of a header shaped exactly cultcache.store.v<N> or cultcache.store.v<N>.<suffix>; + /// null for any other header. A header whose N exceeds what a runtime reads is a newer format, which that runtime + /// refuses and must never rewrite. + /// + internal static int? FormatNumber(string? header) + { + const string prefix = "cultcache.store.v"; + if (header == null || !header.StartsWith(prefix, StringComparison.Ordinal)) + return null; + var end = prefix.Length; + while (end < header.Length && header[end] >= '0' && header[end] <= '9') + end++; + if (end == prefix.Length || end - prefix.Length > 6 || (end < header.Length && header[end] != '.')) + return null; + return int.Parse(header.Substring(prefix.Length, end - prefix.Length), System.Globalization.CultureInfo.InvariantCulture); + } public CultSchemaCatalogEntry[] SchemaCatalog { get; set; } = Array.Empty(); public CultPersistedRecord[] Records { get; set; } = Array.Empty(); } @@ -3354,6 +3372,14 @@ namespace GameCult.Caching protected FileInfo FileInfo { get; } protected abstract byte[] SerializeSnapshot(CultPersistedStoreSnapshot snapshot); protected abstract CultPersistedStoreSnapshot DeserializeSnapshot(byte[] data); + + /// + /// Reads only the format header of a stored file, or null when this store cannot tell. A plain flush rewrites the + /// whole file from this cache's view, so it first reads this header and refuses a file stamped with a newer + /// format than the store reads; a store that returns null here is not guarded. + /// + protected virtual string? ReadFormatVersion(byte[] data) => null; + protected abstract byte[] SerializePayload(object document); protected abstract object DeserializePayload(Type documentType, byte[] payload); @@ -3429,6 +3455,7 @@ namespace GameCult.Caching { using (AcquireLock(wait: true)) { + RefuseToRewriteANewerFormat(); WriteSnapshot( Entries.Values.Select(entry => ToPersistedRecord(entry, SerializePayload)), Entries.Values.Select(entry => entry.Descriptor.ToCatalogEntry())); @@ -3520,6 +3547,30 @@ namespace GameCult.Caching Judging!(arriving, departing); } + // A plain flush replaces the file without merging it, so under the lock it first reads the stamp: a file another + // process has since written in a newer format is refused and left as it is. An unreadable stamp (missing, empty, + // corrupt) is not newer, and the flush proceeds as it always has, so it can still heal a damaged store. + private void RefuseToRewriteANewerFormat() + { + FileInfo.Refresh(); + if (!FileInfo.Exists) + return; + string? header; + try + { + header = ReadFormatVersion(ReadAllBytesShared(FileInfo.FullName)); + } + catch (Exception) + { + return; + } + + if (FormatNumber(header) is int number && number > 2) + throw new NotSupportedException( + $"Store {FileInfo.FullName} is {header}, a newer format than this runtime reads " + + $"({CultPersistedStoreSnapshot.FormatV1} and {CultPersistedStoreSnapshot.FormatV2}); a flush would downgrade it, so none was written."); + } + private CultPersistedStoreSnapshot? ReadSnapshot() { FileInfo.Refresh(); diff --git a/tests/GameCult.Caching.Tests/PreCut2StoreFormatTests.cs b/tests/GameCult.Caching.Tests/PreCut2StoreFormatTests.cs index 0f64f3ff..393d1300 100644 --- a/tests/GameCult.Caching.Tests/PreCut2StoreFormatTests.cs +++ b/tests/GameCult.Caching.Tests/PreCut2StoreFormatTests.cs @@ -156,6 +156,100 @@ namespace GameCult.Caching.Tests } } + // Two processes: this one attached a v1 store, then a later build rewrote it as v3. A plain flush, a staged write + // and a staged delete must each refuse to downgrade it, leaving the file as the other process wrote it. + [Test] + public async Task SingleFileFlushRefusesAStoreAnotherProcessStampedNewerSinceAttach() + { + var root = CopyFixture("single-file-v1"); + try + { + var file = Path.Combine(root, "store.msgpack"); + using var cache = new CultCache(); + cache.AddBackingStore(new SingleFileMessagePackBackingStore(file)); + await cache.PullAllBackingStoresAsync(); + + var snapshot = CultDocumentMessagePackSerialization.DeserializeSnapshot(File.ReadAllBytes(file)); + snapshot.FormatVersion = "cultcache.store.v3"; + File.WriteAllBytes(file, CultDocumentMessagePackSerialization.SerializeSnapshot(snapshot)); + var before = File.ReadAllBytes(file); + + await cache.AddAsync(new PreCut2FixtureNote { Text = "staged after the other process wrote v3" }); + Assert.Throws(() => cache.FlushAllBackingStores()); + Assert.That(File.ReadAllBytes(file), Is.EqualTo(before)); + + cache.Remove(new CultRecordKey("item:anvil")); + Assert.Throws(() => cache.BackingStores.Single().PushAll()); + Assert.That(File.ReadAllBytes(file), Is.EqualTo(before)); + } + finally + { + if (Directory.Exists(root)) Directory.Delete(root, recursive: true); + } + } + + // An unreadable stamp is not a newer format: a flush still heals a damaged store. + [Test] + public async Task SingleFileFlushStillHealsAStoreWhoseStampIsUnreadable() + { + var root = CopyFixture("single-file-v1"); + try + { + var file = Path.Combine(root, "store.msgpack"); + using var cache = new CultCache(); + cache.AddBackingStore(new SingleFileMessagePackBackingStore(file)); + await cache.PullAllBackingStoresAsync(); + File.WriteAllBytes(file, new byte[] { 0xc1, 0xc1 }); + + await cache.AddAsync(new PreCut2FixtureNote { Text = "heals" }); + cache.FlushAllBackingStores(); + + Assert.That(CultDocumentMessagePackSerialization.DeserializeSnapshot(File.ReadAllBytes(file)).FormatVersion, + Is.EqualTo("cultcache.store.v1")); + } + finally + { + if (Directory.Exists(root)) Directory.Delete(root, recursive: true); + } + } + + [TestCase("cultcache.store.v5.directory-content-addressed-pages")] + [TestCase("cultcache.store.v9")] + public async Task DirectoryWritesRefuseAStoreAnotherProcessStampedNewerSinceAttach(string newer) + { + var root = CopyFixture("directory-v4"); + try + { + var manifest = Path.Combine(root, "store.cc"); + using var cache = new CultCache(); + cache.AddBackingStore(new DirectoryMessagePackBackingStore(manifest)); + await cache.PullAllBackingStoresAsync(); + + var snapshot = CultDocumentMessagePackSerialization.DeserializeSnapshot(File.ReadAllBytes(manifest)); + snapshot.FormatVersion = newer; + File.WriteAllBytes(manifest, CultDocumentMessagePackSerialization.SerializeSnapshot(snapshot)); + Func> files = () => Directory + .GetFiles(root, "*", SearchOption.AllDirectories) + .Where(path => !path.EndsWith(".lock", StringComparison.Ordinal)) + .ToDictionary(path => path, File.ReadAllBytes); + var before = files(); + + await cache.AddAsync(new PreCut2FixtureNote { Text = "staged after the other process wrote the newer format" }); + Assert.Throws(() => cache.FlushAllBackingStores()); + Assert.That(files().Keys, Is.EquivalentTo(before.Keys)); + foreach (var pair in before) Assert.That(File.ReadAllBytes(pair.Key), Is.EqualTo(pair.Value), pair.Key); + + cache.Remove(new CultRecordKey("item:anvil")); + Assert.Throws(() => cache.FlushAllBackingStores()); + Assert.That(files().Keys, Is.EquivalentTo(before.Keys)); + foreach (var pair in before) Assert.That(File.ReadAllBytes(pair.Key), Is.EqualTo(pair.Value), pair.Key); + } + finally + { + if (Directory.Exists(root)) Directory.Delete(root, recursive: true); + } + } + [Test] public void SingleFileRefusesExtraRecordSlotNamingTheRecord() {