diff --git a/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs b/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs index 0c980c61..4f1a2521 100644 --- a/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs +++ b/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs @@ -21,6 +21,8 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore 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"; + /// Read-probe stage emitted while the selected manifest generation is leased. + public const string AfterManifestReadStage = "after-manifest-read"; private const string LegacyIndexedFormatVersion = "cultcache.store.v2.directory-indexed"; private const string LegacyMetadataPageFormatVersion = "cultcache.store.v3.directory-immutable-pages"; private const string IndexedFormatVersion = "cultcache.store.v4.directory-content-addressed-pages"; @@ -46,6 +48,9 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore /// Optional deterministic fault/progress hook for durability tests and host diagnostics. public Action? FlushStageProbe { get; set; } + /// Optional deterministic read hook for generation-consistency tests and host diagnostics. + public Action? ReadStageProbe { get; set; } + /// /// Creates a paged MessagePack backing store. /// @@ -86,46 +91,50 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore startupPhase.Restart(); } - _recordDirectory.Refresh(); - if (!File.Exists(_manifestFile.FullName) && !_recordDirectory.Exists) - { - SetLastSchemaMigrationReports(Array.Empty()); - IsDirty = false; - return; - } - - var manifest = ReadManifest(); - Trace($"manifest records={manifest.Records.Length}"); - _durableCatalog = manifest.SchemaCatalog; var reports = new List(); var loaded = new Dictionary(StringComparer.Ordinal); - var indexed = string.Equals(manifest.FormatVersion, IndexedFormatVersion, StringComparison.Ordinal) || - string.Equals(manifest.FormatVersion, LegacyMetadataPageFormatVersion, StringComparison.Ordinal) || - string.Equals(manifest.FormatVersion, LegacyIndexedFormatVersion, StringComparison.Ordinal); - if (indexed) + using (AcquireCommitLease()) { - _manifestUsesImmutablePages = string.Equals(manifest.FormatVersion, IndexedFormatVersion, StringComparison.Ordinal); - _manifestUsesMetadataPages = string.Equals(manifest.FormatVersion, LegacyMetadataPageFormatVersion, StringComparison.Ordinal); - _needsIndexUpgrade = !_manifestUsesImmutablePages; - _durableIndex = manifest.Records - .Where(record => !string.IsNullOrWhiteSpace(record.Key)) - .GroupBy(record => record.Key, StringComparer.Ordinal) - .ToDictionary(group => group.Key, group => group.Last(), StringComparer.Ordinal); + _recordDirectory.Refresh(); + if (!File.Exists(_manifestFile.FullName) && !_recordDirectory.Exists) + { + SetLastSchemaMigrationReports(Array.Empty()); + IsDirty = false; + return; + } - LoadIndexedRecords(manifest, loaded, reports); - Trace($"indexed-pages loaded={loaded.Count}"); - } - else - { - // V1 manifests have no page index. Hydrate once to discover the durable keys; - // the next flush replaces this compatibility path with the indexed manifest. - _needsIndexUpgrade = true; - var legacyRecords = LoadLegacyRecords(manifest, loaded, reports); - _durableIndex = legacyRecords - .Where(record => !string.IsNullOrWhiteSpace(record.Key)) - .GroupBy(record => record.Key, StringComparer.Ordinal) - .ToDictionary(group => group.Key, group => ToIndexRecord(group.Last()), StringComparer.Ordinal); - Trace($"legacy-pages loaded={loaded.Count}"); + var manifest = ReadManifest(); + ReadStageProbe?.Invoke(AfterManifestReadStage); + Trace($"manifest records={manifest.Records.Length}"); + _durableCatalog = manifest.SchemaCatalog; + var indexed = string.Equals(manifest.FormatVersion, IndexedFormatVersion, StringComparison.Ordinal) || + string.Equals(manifest.FormatVersion, LegacyMetadataPageFormatVersion, StringComparison.Ordinal) || + string.Equals(manifest.FormatVersion, LegacyIndexedFormatVersion, StringComparison.Ordinal); + if (indexed) + { + _manifestUsesImmutablePages = string.Equals(manifest.FormatVersion, IndexedFormatVersion, StringComparison.Ordinal); + _manifestUsesMetadataPages = string.Equals(manifest.FormatVersion, LegacyMetadataPageFormatVersion, StringComparison.Ordinal); + _needsIndexUpgrade = !_manifestUsesImmutablePages; + _durableIndex = manifest.Records + .Where(record => !string.IsNullOrWhiteSpace(record.Key)) + .GroupBy(record => record.Key, StringComparer.Ordinal) + .ToDictionary(group => group.Key, group => group.Last(), StringComparer.Ordinal); + + LoadIndexedRecords(manifest, loaded, reports); + Trace($"indexed-pages loaded={loaded.Count}"); + } + else + { + // V1 manifests have no page index. Hydrate once to discover the durable keys; + // the next flush replaces this compatibility path with the indexed manifest. + _needsIndexUpgrade = true; + var legacyRecords = LoadLegacyRecords(manifest, loaded, reports); + _durableIndex = legacyRecords + .Where(record => !string.IsNullOrWhiteSpace(record.Key)) + .GroupBy(record => record.Key, StringComparer.Ordinal) + .ToDictionary(group => group.Key, group => ToIndexRecord(group.Last()), StringComparer.Ordinal); + Trace($"legacy-pages loaded={loaded.Count}"); + } } foreach (var pair in loaded) @@ -175,30 +184,39 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore private void PullSelectedCore(Func selector) { if (selector == null) throw new ArgumentNullException(nameof(selector)); - var manifest = ReadManifest(); - if (!string.Equals(manifest.FormatVersion, IndexedFormatVersion, StringComparison.Ordinal) && - !string.Equals(manifest.FormatVersion, LegacyMetadataPageFormatVersion, StringComparison.Ordinal) && - !string.Equals(manifest.FormatVersion, LegacyIndexedFormatVersion, StringComparison.Ordinal)) + var fallbackToFullPull = false; + CultPersistedRecord[] selected = Array.Empty(); + var loaded = new Dictionary(StringComparer.Ordinal); + var reports = new List(); + using (AcquireCommitLease()) + { + var manifest = ReadManifest(); + ReadStageProbe?.Invoke(AfterManifestReadStage); + fallbackToFullPull = !string.Equals(manifest.FormatVersion, IndexedFormatVersion, StringComparison.Ordinal) && + !string.Equals(manifest.FormatVersion, LegacyMetadataPageFormatVersion, StringComparison.Ordinal) && + !string.Equals(manifest.FormatVersion, LegacyIndexedFormatVersion, StringComparison.Ordinal); + if (!fallbackToFullPull) + { + _manifestUsesImmutablePages = string.Equals(manifest.FormatVersion, IndexedFormatVersion, StringComparison.Ordinal); + _manifestUsesMetadataPages = string.Equals(manifest.FormatVersion, LegacyMetadataPageFormatVersion, StringComparison.Ordinal); + _needsIndexUpgrade = !_manifestUsesImmutablePages; + _durableCatalog = manifest.SchemaCatalog; + _durableIndex = manifest.Records + .Where(record => !string.IsNullOrWhiteSpace(record.Key)) + .GroupBy(record => record.Key, StringComparer.Ordinal) + .ToDictionary(group => group.Key, group => group.Last(), StringComparer.Ordinal); + selected = manifest.Records + .Where(record => selector(ToMetadata(record))) + .OrderBy(record => record.Key, StringComparer.Ordinal) + .ToArray(); + LoadRecordPages(selected, manifest.SchemaCatalog, loaded, reports); + } + } + if (fallbackToFullPull) { PullAllCore(); return; } - - _manifestUsesImmutablePages = string.Equals(manifest.FormatVersion, IndexedFormatVersion, StringComparison.Ordinal); - _manifestUsesMetadataPages = string.Equals(manifest.FormatVersion, LegacyMetadataPageFormatVersion, StringComparison.Ordinal); - _needsIndexUpgrade = !_manifestUsesImmutablePages; - _durableCatalog = manifest.SchemaCatalog; - _durableIndex = manifest.Records - .Where(record => !string.IsNullOrWhiteSpace(record.Key)) - .GroupBy(record => record.Key, StringComparer.Ordinal) - .ToDictionary(group => group.Key, group => group.Last(), StringComparer.Ordinal); - var selected = manifest.Records - .Where(record => selector(ToMetadata(record))) - .OrderBy(record => record.Key, StringComparer.Ordinal) - .ToArray(); - var loaded = new Dictionary(StringComparer.Ordinal); - var reports = new List(); - LoadRecordPages(selected, manifest.SchemaCatalog, loaded, reports); PublishSelected(loaded, selected.Select(record => record.Key).ToHashSet(StringComparer.Ordinal)); SetLastSchemaMigrationReports(reports); IsDirty = !_dirtyKeys.IsEmpty || !_deletedKeys.IsEmpty; @@ -438,7 +456,7 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore ? MetadataRecordPath(metadata) : LegacyRecordPath(metadata.Key); if (!File.Exists(path)) - return; + throw new InvalidDataException($"Committed record page '{path}' is missing from the selected manifest generation."); var pagePayload = ReadAllBytesShared(path); if (_manifestUsesImmutablePages && !HashPayload(pagePayload).SequenceEqual(metadata.Payload)) @@ -788,6 +806,7 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore private FileStream AcquireCommitLease() { + Directory.CreateDirectory(_recordDirectory.FullName); var lockPath = Path.Combine(_recordDirectory.FullName, ".commit.lock"); var started = Stopwatch.StartNew(); while (true) diff --git a/tests/GameCult.Caching.Tests/BackingStoreTests.cs b/tests/GameCult.Caching.Tests/BackingStoreTests.cs index e43fa7b6..ee833f17 100644 --- a/tests/GameCult.Caching.Tests/BackingStoreTests.cs +++ b/tests/GameCult.Caching.Tests/BackingStoreTests.cs @@ -839,6 +839,103 @@ namespace GameCult.Caching.Tests } } + [Test] + public async Task DirectoryMessagePackBackingStore_ReaderLeasesManifestUntilEveryReferencedPageIsHydrated() + { + var filePath = Path.Combine(Path.GetTempPath(), $"cultlib-tests-{Guid.NewGuid():N}.cc"); + var recordsPath = DirectoryMessagePackBackingStore.DefaultRecordDirectoryPath(filePath); + try + { + var key = new CultRecordKey("record:reader-generation"); + var writerStore = new DirectoryMessagePackBackingStore(filePath, recordsPath); + var writerCache = new CultCache(); + writerCache.AddBackingStore(writerStore); + await writerCache.UpsertAsync( + new NamedTestEntry { Name = "generation", Value = "old" }, + new CultRecordHandle(key)); + writerStore.PushAll(); + + var readerStore = new DirectoryMessagePackBackingStore(filePath, recordsPath); + var readerCache = new CultCache(); + readerCache.AddBackingStore(readerStore); + using var manifestRead = new ManualResetEventSlim(); + using var releaseReader = new ManualResetEventSlim(); + readerStore.ReadStageProbe = stage => + { + if (!string.Equals(stage, DirectoryMessagePackBackingStore.AfterManifestReadStage, StringComparison.Ordinal)) + return; + manifestRead.Set(); + Assert.That(releaseReader.Wait(TimeSpan.FromSeconds(5)), Is.True); + }; + + var read = Task.Run(() => readerStore.PullAll()); + Assert.That(manifestRead.Wait(TimeSpan.FromSeconds(5)), Is.True); + await writerCache.UpsertAsync( + new NamedTestEntry { Name = "generation", Value = "new" }, + new CultRecordHandle(key)); + var write = Task.Run(() => writerStore.PushAll()); + Assert.That(write.Wait(TimeSpan.FromMilliseconds(100)), Is.False, + "writer must wait while a reader hydrates its selected manifest generation"); + + releaseReader.Set(); + await Task.WhenAll(read, write); + Assert.That(readerCache.Get(key)?.Value, Is.EqualTo("old")); + + using var reopened = await CultCacheMessagePack.OpenAsync( + filePath, + new CultCacheOpenOptions { UseDirectoryStore = true }); + Assert.That(reopened.Get(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_ReleasesGenerationLeaseBeforePublishingObservers() + { + var filePath = Path.Combine(Path.GetTempPath(), $"cultlib-tests-{Guid.NewGuid():N}.cc"); + var recordsPath = DirectoryMessagePackBackingStore.DefaultRecordDirectoryPath(filePath); + try + { + var writerStore = new DirectoryMessagePackBackingStore(filePath, recordsPath); + var writerCache = new CultCache(); + writerCache.AddBackingStore(writerStore); + await writerCache.UpsertAsync(new NamedTestEntry { Name = "source", Value = "one" }); + writerStore.PushAll(); + + var readerStore = new DirectoryMessagePackBackingStore(filePath, recordsPath); + var readerCache = new CultCache(); + readerCache.AddBackingStore(readerStore); + var observerRan = false; + using var subscription = readerStore.EntryAdded.Subscribe(_ => + { + if (observerRan) + return; + observerRan = true; + readerCache.UpsertAsync(new NamedTestEntry { Name = "observer", Value = "two" }) + .GetAwaiter() + .GetResult(); + readerStore.PushAll(); + }); + + Assert.That(() => readerStore.PullAll(), Throws.Nothing); + Assert.That(observerRan, Is.True); + using var reopened = await CultCacheMessagePack.OpenAsync( + filePath, + new CultCacheOpenOptions { UseDirectoryStore = true }); + Assert.That(reopened.GetAll().Select(value => value.Name), + Is.EquivalentTo(new[] { "source", "observer" })); + } + finally + { + if (File.Exists(filePath)) File.Delete(filePath); + if (Directory.Exists(recordsPath)) Directory.Delete(recordsPath, recursive: true); + } + } + [Test] public async Task DirectoryMessagePackBackingStore_PullsOnlyExternalRecordDeltas() { diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.dll b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.dll index b279cd09..1deb8004 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.dll and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.dll differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.pdb b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.pdb index b2446e7f..edd92539 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.pdb and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.MessagePack.pdb differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.dll b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.dll index 92231494..931e42dc 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.dll and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.dll differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.pdb b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.pdb index 888adde0..60ce5583 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.pdb and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Caching.pdb differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.dll b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.dll index be0edb36..bc67fc77 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.dll and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.dll differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.pdb b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.pdb index 84bb0c96..f702015c 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.pdb and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Logging.pdb differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.Quic.Native.dll b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.Quic.Native.dll index 47771673..96f9481b 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.Quic.Native.dll and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.Quic.Native.dll differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.Quic.Native.pdb b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.Quic.Native.pdb index 31760e9e..fc8890e2 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.Quic.Native.pdb and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.Quic.Native.pdb differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.dll b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.dll index 6d06333d..65f566e9 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.dll and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.dll differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.pdb b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.pdb index 2b0311d9..ca39b949 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.pdb and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Mesh.pdb differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.WebSockets.dll b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.WebSockets.dll index 3b414164..7cfbbf13 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.WebSockets.dll and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.WebSockets.dll differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.dll b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.dll index a6163a0a..40d4ec74 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.dll and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.dll differ diff --git a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.pdb b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.pdb index 5e79480f..e1aa9f03 100644 Binary files a/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.pdb and b/unity/org.gamecult.cultlib/Runtime/Plugins/GameCult.Networking.pdb differ diff --git a/unity/org.gamecult.cultlib/package.json b/unity/org.gamecult.cultlib/package.json index ad99d1ba..b441d31b 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.52", + "version": "1.0.53", "unity": "2021.3", "description": "CultCache, CultNet, and CultMesh managed runtime assemblies for Unity.", "license": "MIT",