diff --git a/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs b/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs index 191febef..9d9584a5 100644 --- a/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs +++ b/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs @@ -17,7 +17,12 @@ namespace GameCult.Caching.MessagePack; /// public sealed class DirectoryMessagePackBackingStore : CacheBackingStore { - private const string IndexedFormatVersion = "cultcache.store.v2.directory-indexed"; + /// Fault-probe stage emitted before each immutable record page write. + public const string BeforeRecordPageWriteStage = "before-record-page-write"; + /// Fault-probe stage emitted after all pages exist and before manifest finality. + public const string BeforeManifestCommitStage = "before-manifest-commit"; + private const string LegacyIndexedFormatVersion = "cultcache.store.v2.directory-indexed"; + private const string IndexedFormatVersion = "cultcache.store.v3.directory-immutable-pages"; private readonly FileInfo _manifestFile; private readonly DirectoryInfo _recordDirectory; private readonly ConcurrentDictionary _dirtyKeys = new(StringComparer.Ordinal); @@ -27,6 +32,7 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore private Dictionary _legacyInlineRecords = new(StringComparer.Ordinal); private CultSchemaCatalogEntry[] _durableCatalog = Array.Empty(); private bool _needsIndexUpgrade; + private bool _manifestUsesImmutablePages; /// /// Gets or sets the predicate that selects indexed record payloads for hydration. @@ -34,6 +40,9 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore /// public Func? HydrationFilter { get; set; } + /// Optional deterministic fault/progress hook for durability tests and host diagnostics. + public Action? FlushStageProbe { get; set; } + /// /// Creates a paged MessagePack backing store. /// @@ -81,10 +90,12 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore _durableCatalog = manifest.SchemaCatalog; var reports = new List(); var loaded = new Dictionary(StringComparer.Ordinal); - var indexed = string.Equals(manifest.FormatVersion, IndexedFormatVersion, StringComparison.Ordinal); + var indexed = string.Equals(manifest.FormatVersion, IndexedFormatVersion, StringComparison.Ordinal) || + string.Equals(manifest.FormatVersion, LegacyIndexedFormatVersion, StringComparison.Ordinal); if (indexed) { - _needsIndexUpgrade = false; + _manifestUsesImmutablePages = string.Equals(manifest.FormatVersion, IndexedFormatVersion, StringComparison.Ordinal); + _needsIndexUpgrade = !_manifestUsesImmutablePages; _durableIndex = manifest.Records .Where(record => !string.IsNullOrWhiteSpace(record.Key)) .GroupBy(record => record.Key, StringComparer.Ordinal) @@ -148,12 +159,15 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore { if (selector == null) throw new ArgumentNullException(nameof(selector)); var manifest = ReadManifest(); - if (!string.Equals(manifest.FormatVersion, IndexedFormatVersion, StringComparison.Ordinal)) + if (!string.Equals(manifest.FormatVersion, IndexedFormatVersion, StringComparison.Ordinal) && + !string.Equals(manifest.FormatVersion, LegacyIndexedFormatVersion, StringComparison.Ordinal)) { PullAll(); return; } + _manifestUsesImmutablePages = string.Equals(manifest.FormatVersion, IndexedFormatVersion, StringComparison.Ordinal); + _needsIndexUpgrade = !_manifestUsesImmutablePages; _durableCatalog = manifest.SchemaCatalog; _durableIndex = manifest.Records .Where(record => !string.IsNullOrWhiteSpace(record.Key)) @@ -207,13 +221,12 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore Directory.CreateDirectory(_recordDirectory.FullName); var currentManifest = ReadManifest(); - var currentIndex = string.Equals(currentManifest.FormatVersion, IndexedFormatVersion, StringComparison.Ordinal) + var currentManifestIsIndexed = string.Equals(currentManifest.FormatVersion, IndexedFormatVersion, StringComparison.Ordinal) || + string.Equals(currentManifest.FormatVersion, LegacyIndexedFormatVersion, StringComparison.Ordinal); + var currentManifestUsesImmutablePages = string.Equals(currentManifest.FormatVersion, IndexedFormatVersion, StringComparison.Ordinal); + var currentIndex = currentManifestIsIndexed ? currentManifest.Records.ToDictionary(record => record.Key, record => record, StringComparer.Ordinal) : new Dictionary(_durableIndex, StringComparer.Ordinal); - var precommitIndex = currentIndex.Values - .Select(ToIndexRecord) - .OrderBy(record => record.Key, StringComparer.Ordinal) - .ToArray(); foreach (var key in _deletedKeys.Keys) currentIndex.Remove(key); foreach (var key in _dirtyKeys.Keys) @@ -233,18 +246,10 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore .Where(entry => usedSchemaIds.Contains(entry.SchemaId)) .OrderBy(entry => entry.SchemaName, StringComparer.Ordinal) .ToArray(); - var precommitCatalog = currentManifest.SchemaCatalog - .Concat(targetCatalog) - .GroupBy(entry => entry.SchemaId, StringComparer.Ordinal) - .Select(group => group.Last()) - .OrderBy(entry => entry.SchemaName, StringComparer.Ordinal) - .ToArray(); - - WriteManifest(precommitCatalog, precommitIndex); - var keysToWrite = (_needsIndexUpgrade ? Entries.Keys : _dirtyKeys.Keys) .OrderBy(value => value, StringComparer.Ordinal) .ToArray(); + var pageWriteIndex = 0; foreach (var key in keysToWrite) { if (!Entries.TryGetValue(key, out var stored)) @@ -257,7 +262,23 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore document, stored.Descriptor.DocumentType, Registry)); - WriteFileAtomically(RecordPath(key), CultDocumentMessagePackSerialization.SerializePersistedRecord(record)); + FlushStageProbe?.Invoke(BeforeRecordPageWriteStage, pageWriteIndex++); + WriteFileAtomically( + ImmutableRecordPath(ToIndexRecord(record)), + CultDocumentMessagePackSerialization.SerializePersistedRecord(record)); + } + + if (!currentManifestUsesImmutablePages) + { + foreach (var metadata in currentManifest.Records + .Where(record => !_deletedKeys.ContainsKey(record.Key) && !_dirtyKeys.ContainsKey(record.Key)) + .OrderBy(record => record.Key, StringComparer.Ordinal)) + { + var sourcePath = LegacyRecordPath(metadata.Key); + var targetPath = ImmutableRecordPath(ToIndexRecord(metadata)); + if (File.Exists(sourcePath) && !File.Exists(targetPath)) + WriteFileAtomically(targetPath, ReadAllBytesShared(sourcePath)); + } } foreach (var pair in _legacyInlineRecords.OrderBy(pair => pair.Key, StringComparer.Ordinal)) @@ -265,23 +286,18 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore if (_deletedKeys.ContainsKey(pair.Key) || _dirtyKeys.ContainsKey(pair.Key)) continue; WriteFileAtomically( - RecordPath(pair.Key), + ImmutableRecordPath(ToIndexRecord(pair.Value)), CultDocumentMessagePackSerialization.SerializePersistedRecord(pair.Value)); } - foreach (var key in _deletedKeys.Keys.OrderBy(value => value, StringComparer.Ordinal)) - { - var path = RecordPath(key); - if (File.Exists(path)) - { - File.Delete(path); - } - } - + FlushStageProbe?.Invoke(BeforeManifestCommitStage, pageWriteIndex); WriteManifest(targetCatalog, currentIndex.Values .OrderBy(record => record.Key, StringComparer.Ordinal) .ToArray()); + _manifestUsesImmutablePages = true; + DeleteUnreferencedRecordPages(currentIndex.Values); + _durableCatalog = targetCatalog; _durableIndex = currentIndex; _legacyInlineRecords.Clear(); @@ -367,7 +383,9 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore { var started = tracePages ? Stopwatch.GetTimestamp() : 0L; var metadata = records[index]; - var path = RecordPath(metadata.Key); + var path = _manifestUsesImmutablePages + ? ImmutableRecordPath(metadata) + : LegacyRecordPath(metadata.Key); if (!File.Exists(path)) return; var pagePayload = ReadAllBytesShared(path); @@ -607,7 +625,35 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore : ""; } - private string RecordPath(string key) => Path.Combine(_recordDirectory.FullName, $"{HashKey(key)}.msgpack"); + private string LegacyRecordPath(string key) => Path.Combine(_recordDirectory.FullName, $"{HashKey(key)}.msgpack"); + + private string ImmutableRecordPath(CultPersistedRecord metadata) => Path.Combine( + _recordDirectory.FullName, + $"{HashKey(metadata.Key + "\0" + metadata.SchemaId + "\0" + metadata.StoredAt)}.msgpack"); + + private void DeleteUnreferencedRecordPages(IEnumerable durableRecords) + { + var referenced = durableRecords + .Select(ImmutableRecordPath) + .ToHashSet(StringComparer.OrdinalIgnoreCase); + foreach (var file in _recordDirectory.EnumerateFiles("*.msgpack")) + { + if (referenced.Contains(file.FullName)) + continue; + try + { + File.Delete(file.FullName); + } + catch (IOException) + { + // An orphan is harmless: no committed manifest references it. + } + catch (UnauthorizedAccessException) + { + // Cleanup is best effort and never changes the committed generation. + } + } + } private static string HashKey(string key) { @@ -681,6 +727,6 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore } } - File.Copy(sourcePath, destinationPath, overwrite: true); + File.Replace(sourcePath, destinationPath, null); } } diff --git a/tests/GameCult.Caching.Tests/BackingStoreTests.cs b/tests/GameCult.Caching.Tests/BackingStoreTests.cs index 3910946a..66078f26 100644 --- a/tests/GameCult.Caching.Tests/BackingStoreTests.cs +++ b/tests/GameCult.Caching.Tests/BackingStoreTests.cs @@ -434,7 +434,7 @@ namespace GameCult.Caching.Tests } var manifest = CultDocumentMessagePackSerialization.DeserializeSnapshot(File.ReadAllBytes(filePath)); - Assert.That(manifest.FormatVersion, Is.EqualTo("cultcache.store.v2.directory-indexed")); + Assert.That(manifest.FormatVersion, Is.EqualTo("cultcache.store.v3.directory-immutable-pages")); Assert.That(manifest.Records, Has.Length.EqualTo(2)); Assert.That(manifest.Records.All(record => record.Payload.Length == 0), Is.True); @@ -599,7 +599,7 @@ namespace GameCult.Caching.Tests } [Test] - public async Task DirectoryMessagePackBackingStore_Precommits_Catalog_And_Remains_Dirty_When_Record_Write_Fails() + public async Task DirectoryMessagePackBackingStore_Does_Not_Expose_New_Pages_Before_Manifest_Commit() { var filePath = Path.Combine(Path.GetTempPath(), $"cultlib-tests-{Guid.NewGuid():N}.cc"); var recordsPath = DirectoryMessagePackBackingStore.DefaultRecordDirectoryPath(filePath); @@ -624,18 +624,21 @@ namespace GameCult.Caching.Tests alternateDescriptor, new AlternateNamedTestEntry { Name = "new" })); - using (new FileStream(recordPath, FileMode.Open, FileAccess.Read, FileShare.Read)) + store.FlushStageProbe = (stage, _) => { - Assert.That(() => store.PushAll(), Throws.Exception); - } + if (string.Equals(stage, DirectoryMessagePackBackingStore.BeforeManifestCommitStage, StringComparison.Ordinal)) + throw new IOException("injected crash before manifest commit"); + }; + Assert.That(() => store.PushAll(), Throws.TypeOf()); var precommit = CultDocumentMessagePackSerialization.DeserializeSnapshot(File.ReadAllBytes(filePath)); var alternateSchema = alternateDescriptor.SchemaId; Assert.That(precommit.SchemaCatalog.Select(entry => entry.SchemaId), Does.Contain(durableSchema)); - Assert.That(precommit.SchemaCatalog.Select(entry => entry.SchemaId), Does.Contain(alternateSchema)); + Assert.That(precommit.SchemaCatalog.Select(entry => entry.SchemaId), Does.Not.Contain(alternateSchema)); Assert.That(store.IsDirty, Is.True); Assert.That(CultDocumentMessagePackSerialization.DeserializePersistedRecord(File.ReadAllBytes(recordPath)).SchemaId, Is.EqualTo(durableSchema)); + store.FlushStageProbe = null; store.PushAll(); var compacted = CultDocumentMessagePackSerialization.DeserializeSnapshot(File.ReadAllBytes(filePath)); @@ -649,6 +652,67 @@ namespace GameCult.Caching.Tests } } + [TestCase(0)] + [TestCase(1)] + [TestCase(2)] + [TestCase(3)] + public async Task DirectoryMessagePackBackingStore_Fault_Before_Manifest_Leaves_Entire_Previous_Generation(int writtenPageCount) + { + var filePath = Path.Combine(Path.GetTempPath(), $"cultlib-tests-{Guid.NewGuid():N}.cc"); + var recordsPath = DirectoryMessagePackBackingStore.DefaultRecordDirectoryPath(filePath); + try + { + var store = new DirectoryMessagePackBackingStore(filePath, recordsPath); + var cache = new CultCache(); + cache.AddBackingStore(store); + var handles = new[] + { + new CultRecordHandle(new CultRecordKey("record:atomic:a")), + new CultRecordHandle(new CultRecordKey("record:atomic:b")), + new CultRecordHandle(new CultRecordKey("record:atomic:c")) + }; + foreach (var handle in handles) + await cache.UpsertAsync(new NamedTestEntry { Name = handle.Key.Value, Value = "old" }, handle); + store.PushAll(); + + foreach (var handle in handles) + await cache.UpsertAsync(new NamedTestEntry { Name = handle.Key.Value, Value = "new" }, handle); + store.FlushStageProbe = (stage, index) => + { + if ((writtenPageCount < handles.Length && + string.Equals(stage, DirectoryMessagePackBackingStore.BeforeRecordPageWriteStage, StringComparison.Ordinal) && + index == writtenPageCount) || + (writtenPageCount == handles.Length && + string.Equals(stage, DirectoryMessagePackBackingStore.BeforeManifestCommitStage, StringComparison.Ordinal))) + throw new IOException($"injected crash after {writtenPageCount} pages"); + }; + Assert.That(() => store.PushAll(), Throws.TypeOf()); + + using (var reopened = await CultCacheMessagePack.OpenAsync( + filePath, + new CultCacheOpenOptions { UseDirectoryStore = true })) + { + foreach (var handle in handles) + Assert.That(reopened.Get(handle.Key)?.Value, Is.EqualTo("old")); + } + + store.FlushStageProbe = null; + store.PushAll(); + using (var reopened = await CultCacheMessagePack.OpenAsync( + filePath, + new CultCacheOpenOptions { UseDirectoryStore = true })) + { + foreach (var handle in handles) + Assert.That(reopened.Get(handle.Key)?.Value, Is.EqualTo("new")); + } + } + finally + { + if (File.Exists(filePath)) File.Delete(filePath); + if (Directory.Exists(recordsPath)) Directory.Delete(recordsPath, recursive: true); + } + } + [Test] public async Task DirectoryMessagePackBackingStore_PullsOnlyExternalRecordDeltas() { @@ -743,7 +807,7 @@ namespace GameCult.Caching.Tests } [Test] - public async Task DirectoryMessagePackBackingStore_Remains_Dirty_When_Deletion_Fails() + public async Task DirectoryMessagePackBackingStore_Deletion_Commits_While_Locked_Old_Page_Remains_An_Orphan() { var filePath = Path.Combine(Path.GetTempPath(), $"cultlib-tests-{Guid.NewGuid():N}.cc"); var recordsPath = DirectoryMessagePackBackingStore.DefaultRecordDirectoryPath(filePath); @@ -760,18 +824,14 @@ namespace GameCult.Caching.Tests using (new FileStream(recordPath, FileMode.Open, FileAccess.Read, FileShare.Read)) { - Assert.That(() => store.PushAll(), Throws.Exception); + Assert.That(() => store.PushAll(), Throws.Nothing); + Assert.That(File.Exists(recordPath), Is.True); } - Assert.That(File.Exists(recordPath), Is.True); - Assert.That(store.IsDirty, Is.True); - - store.PushAll(); - - Assert.That(File.Exists(recordPath), Is.False); Assert.That(store.IsDirty, Is.False); var compacted = CultDocumentMessagePackSerialization.DeserializeSnapshot(File.ReadAllBytes(filePath)); Assert.That(compacted.SchemaCatalog, Is.Empty); + Assert.That(compacted.Records, Is.Empty); } finally { diff --git a/unity/org.gamecult.cultlib/package.json b/unity/org.gamecult.cultlib/package.json index cae02b50..b695c7b6 100644 --- a/unity/org.gamecult.cultlib/package.json +++ b/unity/org.gamecult.cultlib/package.json @@ -1,7 +1,7 @@ { "name": "org.gamecult.cultlib", "displayName": "CultLib", - "version": "1.0.50", + "version": "1.0.51", "unity": "2021.3", "description": "CultCache, CultNet, and CultMesh managed runtime assemblies for Unity.", "license": "MIT",