diff --git a/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs b/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs index 5a98ff08..75709c96 100644 --- a/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs +++ b/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs @@ -15,6 +15,8 @@ namespace GameCult.Caching.MessagePack; public sealed class DirectoryMessagePackBackingStore : CacheBackingStore { private const string IndexedFormatVersion = "cultcache.store.v4.directory-content-addressed-pages"; + // An unleased load that has not settled after this many attempts throws. + private const int UnleasedLoadAttempts = 5; private readonly FileInfo _manifestFile; private readonly DirectoryInfo _recordDirectory; private readonly ConcurrentDictionary _dirtyKeys = new(StringComparer.Ordinal); @@ -36,6 +38,9 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore public static string DefaultRecordDirectoryPath(string manifestPath) => manifestPath + ".records"; + // Test seam: runs between an unleased load's manifest read and its page reads. + internal Action? UnleasedManifestRead { get; set; } + public override string ToString() => _manifestFile.FullName; public override void PullAll() => Held(PullAllCore); @@ -57,18 +62,42 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore var reports = new List(); var loaded = new Dictionary(StringComparer.Ordinal); CultSchemaCatalogEntry[] catalog; - using (AcquireCommitLease(wait: true, create: false)) + for (var attempt = 1; ; attempt++) { - // Only pages named by the manifest read under the lease are loaded; orphaned pages are never loaded. - var manifest = ReadManifest(); + loaded.Clear(); + reports.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 and every page it named read back; otherwise it reloads. + using var lease = AcquireCommitLease(wait: true, create: false); + var manifestBytes = ReadManifestBytes(); + var manifest = ParseManifest(manifestBytes); Trace($"manifest records={manifest.Records.Length}"); - catalog = manifest.SchemaCatalog; - LoadRecordPages( - manifest.Records.OrderBy(record => record.Key, StringComparer.Ordinal).ToArray(), - manifest.SchemaCatalog, - loaded, - reports); - Trace($"indexed-pages loaded={loaded.Count}"); + try + { + if (lease == null) + UnleasedManifestRead?.Invoke(); + LoadRecordPages( + manifest.Records.OrderBy(record => record.Key, StringComparer.Ordinal).ToArray(), + manifest.SchemaCatalog, + loaded, + reports); + if (lease != null || SameBytes(manifestBytes, ReadManifestBytes())) + { + catalog = manifest.SchemaCatalog; + Trace($"indexed-pages loaded={loaded.Count}"); + break; + } + } + catch (Exception exception) when (lease == null && IsTornGeneration(exception)) + { + if (attempt == UnleasedLoadAttempts) + throw Unsettled(exception); + continue; + } + + if (attempt == UnleasedLoadAttempts) + throw Unsettled(null); } bool Staged(string key) => _dirtyKeys.ContainsKey(key) || _deletedKeys.ContainsKey(key); @@ -349,10 +378,26 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore Payload = HashPayload(pagePayload) }; + private CultPersistedStoreSnapshot ReadManifest() => ParseManifest(ReadManifestBytes()); + + private byte[]? ReadManifestBytes() => File.Exists(_manifestFile.FullName) ? ReadAllBytesShared(_manifestFile.FullName) : null; + + private static bool SameBytes(byte[]? first, byte[]? second) => + first == null ? second == null : second != null && first.SequenceEqual(second); + + // A page the manifest named that vanished or was replaced: a writer committed mid-load. + private static bool IsTornGeneration(Exception exception) => + exception is AggregateException aggregate + ? aggregate.Flatten().InnerExceptions.All(IsTornGeneration) + : exception is InvalidDataException or FileNotFoundException; + + private InvalidOperationException Unsettled(Exception? last) => new( + $"Directory store {_manifestFile.FullName} changed under every one of {UnleasedLoadAttempts} unlocked loads; it did not settle.", last); + // A missing manifest is an empty store. An existing manifest in any other format is refused. - private CultPersistedStoreSnapshot ReadManifest() + private CultPersistedStoreSnapshot ParseManifest(byte[]? manifestBytes) { - if (!File.Exists(_manifestFile.FullName)) + if (manifestBytes == null) { return new CultPersistedStoreSnapshot { @@ -362,7 +407,7 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore }; } - var snapshot = CultDocumentMessagePackSerialization.DeserializeSnapshot(ReadAllBytesShared(_manifestFile.FullName)); + var snapshot = CultDocumentMessagePackSerialization.DeserializeSnapshot(manifestBytes); if (!string.Equals(snapshot.FormatVersion, IndexedFormatVersion, StringComparison.Ordinal)) { throw new InvalidOperationException( @@ -470,8 +515,9 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore return buffer.ToArray(); } - // Only a writer creates the records folder and the lock. A reader opens an existing lock; with none, no writer has ever - // committed here, so there is nothing to exclude and opening the store creates nothing. + // Only a writer creates the records folder and the lock, so opening the store creates nothing. A reader opens an + // existing lock and returns null when there is none; that does not mean no writer committed (a store may be copied + // without its lock, and a first writer may create it mid-load), so an unleased load checks that it settled. private FileStream? AcquireCommitLease(bool wait, bool create) { if (create) diff --git a/src/GameCult.Caching.MessagePack/GameCult.Caching.MessagePack.csproj b/src/GameCult.Caching.MessagePack/GameCult.Caching.MessagePack.csproj index 794936bd..c2bd1c1b 100644 --- a/src/GameCult.Caching.MessagePack/GameCult.Caching.MessagePack.csproj +++ b/src/GameCult.Caching.MessagePack/GameCult.Caching.MessagePack.csproj @@ -8,5 +8,6 @@ + diff --git a/src/GameCult.Caching/Contracts/cultcache-store-composition.md b/src/GameCult.Caching/Contracts/cultcache-store-composition.md index dd8a20d3..8c4fc529 100644 --- a/src/GameCult.Caching/Contracts/cultcache-store-composition.md +++ b/src/GameCult.Caching/Contracts/cultcache-store-composition.md @@ -195,6 +195,12 @@ replaces; writers bump a minted timestamp by one tick when it is not later. flush first. - Single-file stores lock a sidecar `.lock` opened exclusively. Directory stores use their existing commit lease. +- A directory store with no `.commit.lock` (no writer has created it yet, or + the store was copied without it) gives a pull no lease, and opening it + creates none. Such a pull re-reads the manifest after loading its pages and + reloads when the manifest changed or a page it named was missing or did not + match its content address, at most five times; a pull that has not settled + by then throws. **A plain flush and an unconditional commit are last-writer-wins: per file for single-file stores, per key for directory stores.** Both run diff --git a/tests/GameCult.Caching.Tests/BackingStoreTests.cs b/tests/GameCult.Caching.Tests/BackingStoreTests.cs index be868959..c8be4b64 100644 --- a/tests/GameCult.Caching.Tests/BackingStoreTests.cs +++ b/tests/GameCult.Caching.Tests/BackingStoreTests.cs @@ -619,6 +619,58 @@ namespace GameCult.Caching.Tests } } + // A store copied without its lock gives a reader no lease; a writer then commits between the reader's manifest read + // and its page reads, either replacing a page the reader named (it vanishes) or only adding a key (the manifest moves). + [TestCase(true)] + [TestCase(false)] + public async Task DirectoryMessagePackBackingStore_UnlockedLoad_Reloads_When_A_Writer_Commits_Mid_Load(bool replaceReadPage) + { + var filePath = Path.Combine(Path.GetTempPath(), $"cultlib-tests-{Guid.NewGuid():N}.cc"); + var recordsPath = DirectoryMessagePackBackingStore.DefaultRecordDirectoryPath(filePath); + + try + { + CultRecordKey seeded; + using (var seed = await CultCacheMessagePack.OpenAsync(filePath, new CultCacheOpenOptions { UseDirectoryStore = true })) + { + seeded = (await seed.UpsertAsync(new NamedTestEntry { Name = "copied", Value = "before" })).Key; + await seed.FlushAsync(); + } + + // Attached while the lock exists, so the only unleased load is the pull below. + var readerStore = new DirectoryMessagePackBackingStore(filePath, recordsPath); + var reader = new CultCache(); + reader.AddBackingStore(readerStore); + await reader.PullAllBackingStoresAsync(); + File.Delete(Path.Combine(recordsPath, ".commit.lock")); + var committed = new System.Collections.Generic.List(); + // The writer is another cache, so it commits from another thread: a thread holds one cache's gate at a time. + readerStore.UnleasedManifestRead = () => + { + if (committed.Count > 0) return; + Task.Run(async () => + { + using var writer = CultCacheMessagePack.Create(filePath, new CultCacheOpenOptions { UseDirectoryStore = true }); + var document = new NamedTestEntry { Name = "writer", Value = "after" }; + committed.Add(replaceReadPage + ? await writer.UpsertAsync(typeof(NamedTestEntry), document, seeded) + : (await writer.UpsertAsync(document)).Key); + await writer.FlushAsync(); + }).GetAwaiter().GetResult(); + }; + + await reader.PullAllBackingStoresAsync(); + + Assert.That(committed, Has.Count.EqualTo(1), "the writer committed mid-load"); + Assert.That(reader.Get(committed[0])?.Value, Is.EqualTo("after"), "the reader reloaded the writer's generation"); + } + finally + { + if (File.Exists(filePath)) File.Delete(filePath); + if (Directory.Exists(recordsPath)) Directory.Delete(recordsPath, recursive: true); + } + } + [Test] public async Task DirectoryMessagePackBackingStore_Deletion_Commits_While_Locked_Old_Page_Remains_An_Orphan() {