From 900e148d8ad6dceed4b9c14dac23d12a869cb539 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sun, 13 Sep 2026 16:50:08 +0200 Subject: [PATCH] Read only v4 directory stores; delete probes and selective hydration DirectoryMessagePackBackingStore loses HydrationFilter, FlushStageProbe, ReadStageProbe and their stage constants, ReadPersistedGeneration, PullSelected, and every legacy path: the v2 directory-indexed and v3 immutable-pages formats, v1 inline records and split pages, the index upgrade on flush, and the legacy page paths. CacheBackingStore.PullSelected and CultPersistedRecordMetadata go with them. A missing manifest is an empty store. An existing manifest whose FormatVersion is not cultcache.store.v4.directory-content-addressed-pages throws "Directory store {path} is {format}; only {v4} is readable." A clean flush writes nothing, since there is no legacy index left to upgrade. Tests deleted with the surface they injected through: the v1 missing-catalog load, pages-before-manifest, fault-before-manifest (4 cases), same-timestamp page, mutation-during-flush and reader-lease tests. The orphan-page test now checks through a fresh store pull instead of ReadPersistedGeneration. Co-Authored-By: Claude Opus 5 --- .../DirectoryMessagePackBackingStore.cs | 410 ++---------------- src/GameCult.Caching/CultCache.cs | 29 -- .../BackingStoreTests.cs | 333 +------------- 3 files changed, 34 insertions(+), 738 deletions(-) diff --git a/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs b/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs index 1fdf07c8..0e6ab7bb 100644 --- a/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs +++ b/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs @@ -5,7 +5,6 @@ using System.Diagnostics; using System.IO; using System.Linq; using System.Security.Cryptography; -using System.Text; using System.Threading; using System.Threading.Tasks; using MessagePack; @@ -17,14 +16,6 @@ namespace GameCult.Caching.MessagePack; /// public sealed class DirectoryMessagePackBackingStore : CacheBackingStore { - /// 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"; - /// 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"; private readonly FileInfo _manifestFile; private readonly DirectoryInfo _recordDirectory; @@ -33,23 +24,7 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore private readonly ConcurrentDictionary _deletedKeys = new(StringComparer.Ordinal); private readonly HashSet _hydratedKeys = new(StringComparer.Ordinal); private Dictionary _durableIndex = new(StringComparer.Ordinal); - private Dictionary _legacyInlineRecords = new(StringComparer.Ordinal); private CultSchemaCatalogEntry[] _durableCatalog = Array.Empty(); - private bool _needsIndexUpgrade; - private bool _manifestUsesImmutablePages; - private bool _manifestUsesMetadataPages; - - /// - /// Gets or sets the predicate that selects indexed record payloads for hydration. - /// A null predicate hydrates every record. - /// - public Func? HydrationFilter { get; set; } - - /// 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. @@ -70,57 +45,6 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore /// public static string DefaultRecordDirectoryPath(string manifestPath) => manifestPath + ".records"; - /// - /// Reads one complete persisted generation without deserializing documents through a - /// runtime registry. Indexed manifests are resolved only through the pages they name; - /// orphaned pages from older generations are never included. - /// - public CultPersistedStoreSnapshot ReadPersistedGeneration() - { - lock (_mutationGate) - using (AcquireCommitLease()) - { - var manifest = ReadManifest(); - var indexed = string.Equals(manifest.FormatVersion, IndexedFormatVersion, StringComparison.Ordinal) || - string.Equals(manifest.FormatVersion, LegacyMetadataPageFormatVersion, StringComparison.Ordinal) || - string.Equals(manifest.FormatVersion, LegacyIndexedFormatVersion, StringComparison.Ordinal); - var records = indexed - ? manifest.Records - .OrderBy(record => record.Key, StringComparer.Ordinal) - .Select(record => ReadPersistedRecordPage(record, manifest.FormatVersion, out _)) - .ToArray() - : ReadLegacyPersistedGenerationRecords(manifest); - return new CultPersistedStoreSnapshot - { - FormatVersion = manifest.FormatVersion, - SchemaCatalog = manifest.SchemaCatalog.ToArray(), - Records = records - }; - } - } - - private CultPersistedRecord[] ReadLegacyPersistedGenerationRecords(CultPersistedStoreSnapshot manifest) - { - var records = manifest.Records - .Where(record => !string.IsNullOrWhiteSpace(record.Key)) - .ToList(); - _recordDirectory.Refresh(); - if (_recordDirectory.Exists) - { - records.AddRange(_recordDirectory.EnumerateFiles("*.msgpack") - .OrderBy(file => file.Name, StringComparer.Ordinal) - .Select(file => CultDocumentMessagePackSerialization.DeserializePersistedRecord( - ReadAllBytesShared(file.FullName)))); - } - - return records - .Where(record => !string.IsNullOrWhiteSpace(record.Key)) - .GroupBy(record => record.Key, StringComparer.Ordinal) - .Select(group => group.Last()) - .OrderBy(record => record.Key, StringComparer.Ordinal) - .ToArray(); - } - /// public override void PullAll() { @@ -146,46 +70,20 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore var loaded = new Dictionary(StringComparer.Ordinal); using (AcquireCommitLease()) { - _recordDirectory.Refresh(); - if (!File.Exists(_manifestFile.FullName) && !_recordDirectory.Exists) - { - SetLastSchemaMigrationReports(Array.Empty()); - IsDirty = false; - return; - } - 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); + _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}"); - } + LoadRecordPages( + manifest.Records.OrderBy(record => record.Key, StringComparer.Ordinal).ToArray(), + manifest.SchemaCatalog, + loaded, + reports); + Trace($"indexed-pages loaded={loaded.Count}"); } foreach (var pair in loaded) @@ -222,54 +120,6 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore Trace("publish"); } - /// - public override void PullSelected(Func selector) - { - lock (_mutationGate) - PullSelectedCore(selector); - } - - private void PullSelectedCore(Func selector) - { - if (selector == null) throw new ArgumentNullException(nameof(selector)); - 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; - } - PublishSelected(loaded, selected.Select(record => record.Key).ToHashSet(StringComparer.Ordinal)); - SetLastSchemaMigrationReports(reports); - IsDirty = !_dirtyKeys.IsEmpty || !_deletedKeys.IsEmpty; - } - /// public override bool ContainsDurableRecord(CultRecordKey key) { @@ -355,7 +205,7 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore private void PushAllCore() { - if (!IsDirty && !_needsIndexUpgrade) + if (!IsDirty) { return; } @@ -365,14 +215,7 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore using var commitLease = AcquireCommitLease(); var currentManifest = ReadManifest(); - var currentManifestIsIndexed = string.Equals(currentManifest.FormatVersion, IndexedFormatVersion, StringComparison.Ordinal) || - string.Equals(currentManifest.FormatVersion, LegacyMetadataPageFormatVersion, StringComparison.Ordinal) || - string.Equals(currentManifest.FormatVersion, LegacyIndexedFormatVersion, StringComparison.Ordinal); - var currentManifestUsesImmutablePages = string.Equals(currentManifest.FormatVersion, IndexedFormatVersion, StringComparison.Ordinal); - var currentManifestUsesMetadataPages = string.Equals(currentManifest.FormatVersion, LegacyMetadataPageFormatVersion, StringComparison.Ordinal); - var currentIndex = currentManifestIsIndexed - ? currentManifest.Records.ToDictionary(record => record.Key, record => record, StringComparer.Ordinal) - : new Dictionary(_durableIndex, StringComparer.Ordinal); + var currentIndex = currentManifest.Records.ToDictionary(record => record.Key, record => record, StringComparer.Ordinal); foreach (var key in _deletedKeys.Keys) currentIndex.Remove(key); foreach (var key in _dirtyKeys.Keys) @@ -392,10 +235,9 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore .Where(entry => usedSchemaIds.Contains(entry.SchemaId)) .OrderBy(entry => entry.SchemaName, StringComparer.Ordinal) .ToArray(); - var keysToWrite = (_needsIndexUpgrade ? Entries.Keys : _dirtyKeys.Keys) + var keysToWrite = _dirtyKeys.Keys .OrderBy(value => value, StringComparer.Ordinal) .ToArray(); - var pageWriteIndex = 0; foreach (var key in keysToWrite) { if (!Entries.TryGetValue(key, out var stored)) @@ -411,57 +253,19 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore var pagePayload = CultDocumentMessagePackSerialization.SerializePersistedRecord(record); var indexRecord = ToContentAddressedIndexRecord(record, pagePayload); currentIndex[key] = indexRecord; - FlushStageProbe?.Invoke(BeforeRecordPageWriteStage, pageWriteIndex++); WriteFileAtomically( ContentAddressedRecordPath(indexRecord), pagePayload); } - 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 = currentManifestUsesMetadataPages - ? MetadataRecordPath(metadata) - : LegacyRecordPath(metadata.Key); - if (!File.Exists(sourcePath)) - continue; - var pagePayload = ReadAllBytesShared(sourcePath); - var indexRecord = ToContentAddressedIndexRecord(metadata, pagePayload); - currentIndex[metadata.Key] = indexRecord; - var targetPath = ContentAddressedRecordPath(indexRecord); - if (!File.Exists(targetPath)) - WriteFileAtomically(targetPath, pagePayload); - } - } - - foreach (var pair in _legacyInlineRecords.OrderBy(pair => pair.Key, StringComparer.Ordinal)) - { - if (_deletedKeys.ContainsKey(pair.Key) || _dirtyKeys.ContainsKey(pair.Key)) - continue; - var pagePayload = CultDocumentMessagePackSerialization.SerializePersistedRecord(pair.Value); - var indexRecord = ToContentAddressedIndexRecord(pair.Value, pagePayload); - currentIndex[pair.Key] = indexRecord; - WriteFileAtomically( - ContentAddressedRecordPath(indexRecord), - pagePayload); - } - - FlushStageProbe?.Invoke(BeforeManifestCommitStage, pageWriteIndex); WriteManifest(targetCatalog, currentIndex.Values .OrderBy(record => record.Key, StringComparer.Ordinal) .ToArray()); - _manifestUsesImmutablePages = true; - _manifestUsesMetadataPages = false; DeleteUnreferencedRecordPages(currentIndex.Values); _durableCatalog = targetCatalog; _durableIndex = currentIndex; - _legacyInlineRecords.Clear(); - _needsIndexUpgrade = false; _dirtyKeys.Clear(); _deletedKeys.Clear(); MarkFlushSucceeded(); @@ -478,49 +282,6 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore WriteFileAtomically(_manifestFile.FullName, CultDocumentMessagePackSerialization.SerializeSnapshot(manifest)); } - private void LoadIndexedRecords( - CultPersistedStoreSnapshot manifest, - Dictionary loaded, - List reports) - { - var selected = manifest.Records - .Where(ShouldHydrate) - .OrderBy(record => record.Key, StringComparer.Ordinal) - .ToArray(); - LoadRecordPages(selected, manifest.SchemaCatalog, loaded, reports); - } - - private CultPersistedRecord[] LoadLegacyRecords( - CultPersistedStoreSnapshot manifest, - Dictionary loaded, - List reports) - { - _legacyInlineRecords = manifest.Records - .Where(record => !string.IsNullOrWhiteSpace(record.Key)) - .GroupBy(record => record.Key, StringComparer.Ordinal) - .ToDictionary(group => group.Key, group => group.Last(), StringComparer.Ordinal); - foreach (var record in manifest.Records.Where(ShouldHydrate)) - { - var catalog = ResolveLegacyUncataloguedRecordCatalog(record, manifest.SchemaCatalog); - reports.Add(Registry.ResolvePersistedSchemaReport(record.SchemaId, catalog)); - var stored = ToStoredDocument( - record, - catalog, - (type, payload) => CultDocumentMessagePackSerialization.DeserializeUntyped(type, payload, Registry)); - loaded[stored.Key.Value] = stored; - } - - if (!_recordDirectory.Exists) - return manifest.Records; - - var pages = _recordDirectory.EnumerateFiles("*.msgpack") - .OrderBy(file => file.Name, StringComparer.Ordinal) - .Select(file => CultDocumentMessagePackSerialization.DeserializePersistedRecord(ReadAllBytesShared(file.FullName))) - .ToArray(); - LoadPersistedRecords(pages.Where(ShouldHydrate).ToArray(), manifest.SchemaCatalog, loaded, reports); - return manifest.Records.Concat(pages).ToArray(); - } - private void LoadRecordPages( CultPersistedRecord[] records, IReadOnlyCollection catalogEntries, @@ -543,11 +304,7 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore { var started = tracePages ? Stopwatch.GetTimestamp() : 0L; var metadata = records[index]; - var record = ReadPersistedRecordPage( - metadata, - _manifestUsesImmutablePages, - _manifestUsesMetadataPages, - out var pagePayload); + var record = ReadPersistedRecordPage(metadata, out var pagePayload); var catalog = ResolveLegacyUncataloguedRecordCatalog(record, catalogEntries); recordReports[index] = Registry.ResolvePersistedSchemaReport(record.SchemaId, catalog); storedRecords[index] = ToStoredDocument( @@ -596,74 +353,6 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore } } - private void LoadPersistedRecords( - CultPersistedRecord[] records, - IReadOnlyCollection catalogEntries, - Dictionary loaded, - List reports) - { - var recordReports = new CultSchemaMigrationReport[records.Length]; - var storedRecords = new CultStoredDocument[records.Length]; - Parallel.For( - 0, - records.Length, - new ParallelOptions { MaxDegreeOfParallelism = Math.Min(Environment.ProcessorCount, 8) }, - index => - { - var record = records[index]; - var catalog = ResolveLegacyUncataloguedRecordCatalog(record, catalogEntries); - recordReports[index] = Registry.ResolvePersistedSchemaReport(record.SchemaId, catalog); - storedRecords[index] = ToStoredDocument( - record, - catalog, - (type, payload) => CultDocumentMessagePackSerialization.DeserializeUntyped(type, payload, Registry)); - }); - - for (var index = 0; index < storedRecords.Length; index++) - { - reports.Add(recordReports[index]); - var stored = storedRecords[index]; - loaded[stored.Key.Value] = stored; - } - } - - private bool ShouldHydrate(CultPersistedRecord record) => - HydrationFilter?.Invoke(ToMetadata(record)) ?? true; - - private static CultPersistedRecordMetadata ToMetadata(CultPersistedRecord record) => - new(record.Key, record.SchemaId, record.StoredAt); - - private void PublishSelected( - IReadOnlyDictionary loaded, - HashSet selectedKeys) - { - foreach (var pair in loaded) - { - if (_dirtyKeys.ContainsKey(pair.Key) || _deletedKeys.ContainsKey(pair.Key)) - continue; - - if (Entries.TryGetValue(pair.Key, out var existing) && - string.Equals(existing.StoredAt, pair.Value.StoredAt, StringComparison.Ordinal) && - string.Equals(existing.Descriptor.SchemaId, pair.Value.Descriptor.SchemaId, StringComparison.Ordinal)) - continue; - - Entries[pair.Key] = pair.Value; - Loaded?.Invoke(pair.Value); - } - - foreach (var missingKey in selectedKeys.Where(key => !loaded.ContainsKey(key)).ToArray()) - { - if (_dirtyKeys.ContainsKey(missingKey) || _deletedKeys.ContainsKey(missingKey)) - continue; - if (Entries.TryRemove(missingKey, out var removed)) - Unloaded?.Invoke(removed); - _hydratedKeys.Remove(missingKey); - } - - foreach (var key in loaded.Keys) - _hydratedKeys.Add(key); - } - private static CultPersistedRecord ToIndexRecord(CultStoredDocument stored) => new() { Key = stored.Key.Value, @@ -672,14 +361,6 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore Payload = Array.Empty() }; - private static CultPersistedRecord ToIndexRecord(CultPersistedRecord record) => new() - { - Key = record.Key, - SchemaId = record.SchemaId, - StoredAt = record.StoredAt, - Payload = Array.Empty() - }; - private static CultPersistedRecord ToContentAddressedIndexRecord( CultPersistedRecord record, byte[] pagePayload) => new() @@ -690,28 +371,27 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore Payload = HashPayload(pagePayload) }; + // A missing manifest is an empty store. An existing manifest in any other format is refused. private CultPersistedStoreSnapshot ReadManifest() { - if (File.Exists(_manifestFile.FullName)) + if (!File.Exists(_manifestFile.FullName)) { - var snapshot = CultDocumentMessagePackSerialization.DeserializeSnapshot(ReadAllBytesShared(_manifestFile.FullName)); - if (snapshot.SchemaCatalog.Length > 0 || snapshot.Records.Length == 0) + return new CultPersistedStoreSnapshot { - return snapshot; - } + FormatVersion = IndexedFormatVersion, + SchemaCatalog = Array.Empty(), + Records = Array.Empty() + }; } - if (File.Exists(_manifestFile.FullName)) + var snapshot = CultDocumentMessagePackSerialization.DeserializeSnapshot(ReadAllBytesShared(_manifestFile.FullName)); + if (!string.Equals(snapshot.FormatVersion, IndexedFormatVersion, StringComparison.Ordinal)) { - return CultDocumentMessagePackSerialization.DeserializeSnapshot(ReadAllBytesShared(_manifestFile.FullName)); + throw new InvalidOperationException( + $"Directory store {_manifestFile.FullName} is {snapshot.FormatVersion}; only {IndexedFormatVersion} is readable."); } - return new CultPersistedStoreSnapshot - { - FormatVersion = "cultcache.store.v1.directory", - SchemaCatalog = Array.Empty(), - Records = Array.Empty() - }; + return snapshot; } // Explicit compatibility path for records written before catalog precommit existed. @@ -788,12 +468,6 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore : ""; } - private string LegacyRecordPath(string key) => Path.Combine(_recordDirectory.FullName, $"{HashKey(key)}.msgpack"); - - private string MetadataRecordPath(CultPersistedRecord metadata) => Path.Combine( - _recordDirectory.FullName, - $"{HashKey(metadata.Key + "\0" + metadata.SchemaId + "\0" + metadata.StoredAt)}.msgpack"); - private string ContentAddressedRecordPath(CultPersistedRecord metadata) { if (metadata.Payload == null || metadata.Payload.Length != 32) @@ -832,32 +506,13 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore } } - private CultPersistedRecord ReadPersistedRecordPage( - CultPersistedRecord metadata, - string formatVersion, - out byte[] pagePayload) => - ReadPersistedRecordPage( - metadata, - string.Equals(formatVersion, IndexedFormatVersion, StringComparison.Ordinal), - string.Equals(formatVersion, LegacyMetadataPageFormatVersion, StringComparison.Ordinal), - out pagePayload); - - private CultPersistedRecord ReadPersistedRecordPage( - CultPersistedRecord metadata, - bool usesContentAddressedPages, - bool usesMetadataPages, - out byte[] pagePayload) + private CultPersistedRecord ReadPersistedRecordPage(CultPersistedRecord metadata, out byte[] pagePayload) { - var path = usesContentAddressedPages - ? ContentAddressedRecordPath(metadata) - : usesMetadataPages - ? MetadataRecordPath(metadata) - : LegacyRecordPath(metadata.Key); + var path = ContentAddressedRecordPath(metadata); if (!File.Exists(path)) throw new InvalidDataException($"Committed record page '{path}' is missing from the selected manifest generation."); pagePayload = ReadAllBytesShared(path); - if (usesContentAddressedPages && - !HashPayload(pagePayload).SequenceEqual(metadata.Payload)) + if (!HashPayload(pagePayload).SequenceEqual(metadata.Payload)) throw new InvalidDataException($"Record page '{path}' does not match its committed content hash."); var record = CultDocumentMessagePackSerialization.DeserializePersistedRecord(pagePayload); if (!string.Equals(record.Key, metadata.Key, StringComparison.Ordinal)) @@ -865,13 +520,6 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore return record; } - private static string HashKey(string key) - { - using var sha256 = SHA256.Create(); - var hash = sha256.ComputeHash(Encoding.UTF8.GetBytes(key)); - return BitConverter.ToString(hash).Replace("-", string.Empty).ToLowerInvariant(); - } - private static void WriteFileAtomically(string path, byte[] payload) { var directory = Path.GetDirectoryName(path)!; diff --git a/src/GameCult.Caching/CultCache.cs b/src/GameCult.Caching/CultCache.cs index 7f1af80e..84b5e869 100644 --- a/src/GameCult.Caching/CultCache.cs +++ b/src/GameCult.Caching/CultCache.cs @@ -2124,27 +2124,6 @@ namespace GameCult.Caching } } - /// - /// Durable metadata available before a backing store hydrates a record payload. - /// - public sealed class CultPersistedRecordMetadata - { - /// Creates durable record metadata. - public CultPersistedRecordMetadata(string key, string schemaId, string storedAt) - { - Key = key; - SchemaId = schemaId; - StoredAt = storedAt; - } - - /// Gets the durable record key. - public string Key { get; } - /// Gets the durable schema identity. - public string SchemaId { get; } - /// Gets the record commit timestamp. - public string StoredAt { get; } - } - /// /// Base class for CultCache persistence adapters. /// @@ -2199,14 +2178,6 @@ namespace GameCult.Caching /// public abstract void PullAll(); /// - /// Pulls records selected by durable metadata. Non-indexed stores fall back to a full pull. - /// - public virtual void PullSelected(Func selector) - { - if (selector == null) throw new ArgumentNullException(nameof(selector)); - PullAll(); - } - /// /// Returns whether the backing store knows a record is durable, including records left cold by selective hydration. /// public virtual bool ContainsDurableRecord(CultRecordKey key) diff --git a/tests/GameCult.Caching.Tests/BackingStoreTests.cs b/tests/GameCult.Caching.Tests/BackingStoreTests.cs index c2eb95f4..486f644f 100644 --- a/tests/GameCult.Caching.Tests/BackingStoreTests.cs +++ b/tests/GameCult.Caching.Tests/BackingStoreTests.cs @@ -403,277 +403,6 @@ namespace GameCult.Caching.Tests } } - [Test] - public async Task DirectoryMessagePackBackingStore_Loads_Record_When_Manifest_Misses_Catalog_Entry() - { - var filePath = Path.Combine(Path.GetTempPath(), $"cultlib-tests-{Guid.NewGuid():N}.cc"); - var recordsPath = DirectoryMessagePackBackingStore.DefaultRecordDirectoryPath(filePath); - var recordPath = Path.Combine(recordsPath, "stale-schema-record.msgpack"); - - try - { - Directory.CreateDirectory(recordsPath); - - var entry = new SchemaStampedTestEntry - { - SchemaVersion = "tests.schema_stamped_entry.v1", - Name = "schema-stamped", - Value = "still readable" - }; - var record = new CultPersistedRecord - { - Key = "record:stale", - SchemaId = "sha256:stale-schema-id-from-cold-record", - StoredAt = "2026-06-25T12:00:00Z", - Payload = CultDocumentMessagePackSerialization.SerializeUntyped(entry, typeof(SchemaStampedTestEntry)) - }; - var manifest = new CultPersistedStoreSnapshot - { - FormatVersion = "cultcache.store.v1.directory", - SchemaCatalog = Array.Empty(), - Records = Array.Empty() - }; - - File.WriteAllBytes(filePath, CultDocumentMessagePackSerialization.SerializeSnapshot(manifest)); - File.WriteAllBytes(recordPath, CultDocumentMessagePackSerialization.SerializePersistedRecord(record)); - - var readStore = new DirectoryMessagePackBackingStore(filePath, recordsPath); - var rawGeneration = readStore.ReadPersistedGeneration(); - Assert.That(rawGeneration.Records.Select(value => value.Key), - Is.EqualTo(new[] { "record:stale" }), - "v1 split record pages remain part of the selected legacy generation"); - var readCache = new CultCache(); - readCache.AddBackingStore(readStore); - await readCache.PullAllBackingStoresAsync(); - - var loaded = readCache.Get(new CultRecordKey("record:stale")); - - Assert.That(loaded, Is.Not.Null); - Assert.That(loaded!.Value, Is.EqualTo("still readable")); - Assert.That(readStore.LastSchemaMigrationReports, Has.Count.EqualTo(1)); - Assert.That(readStore.LastSchemaMigrationReports[0].PersistedSchemaName, Is.EqualTo("tests.schema_stamped_entry")); - Assert.That(readStore.LastSchemaMigrationReports[0].Kind, Is.EqualTo(CultSchemaMigrationKind.CompatibleDrift)); - } - finally - { - if (File.Exists(filePath)) - { - File.Delete(filePath); - } - - if (Directory.Exists(recordsPath)) - { - Directory.Delete(recordsPath, recursive: true); - } - } - } - - [Test] - 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); - - try - { - var store = new DirectoryMessagePackBackingStore(filePath, recordsPath); - var cache = new CultCache(); - cache.AddBackingStore(store); - var key = new CultRecordKey("record:catalog-precommit"); - await cache.UpsertAsync(new NamedTestEntry { Name = "old", Value = "durable" }, new CultRecordHandle(key)); - store.PushAll(); - - var recordPath = Directory.GetFiles(recordsPath, "*.msgpack").Single(); - var durableSchema = CultDocumentMessagePackSerialization - .DeserializePersistedRecord(File.ReadAllBytes(recordPath)) - .SchemaId; - var alternateDescriptor = CultDocumentRegistry.Shared.GetRequired(); - store.Push(new CultStoredDocument( - key, - DateTimeOffset.UtcNow.ToString("O"), - alternateDescriptor, - new AlternateNamedTestEntry { Name = "new" })); - - store.FlushStageProbe = (stage, _) => - { - 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.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)); - Assert.That(compacted.SchemaCatalog.Select(entry => entry.SchemaId), Is.EqualTo(new[] { alternateSchema })); - Assert.That(store.IsDirty, Is.False); - } - finally - { - if (File.Exists(filePath)) File.Delete(filePath); - if (Directory.Exists(recordsPath)) Directory.Delete(recordsPath, recursive: true); - } - } - - [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_SameTimestampPayload_CannotOverwriteCommittedPageBeforeManifest() - { - var filePath = Path.Combine(Path.GetTempPath(), $"cultlib-tests-{Guid.NewGuid():N}.cc"); - var recordsPath = DirectoryMessagePackBackingStore.DefaultRecordDirectoryPath(filePath); - try - { - const string storedAt = "2026-08-19T00:00:00.0000000+00:00"; - var store = new DirectoryMessagePackBackingStore(filePath, recordsPath); - var cache = new CultCache(); - cache.AddBackingStore(store); - var key = new CultRecordKey("record:atomic:same-timestamp"); - var descriptor = CultDocumentRegistry.Shared.GetRequired(); - store.Push(new CultStoredDocument( - key, - storedAt, - descriptor, - new NamedTestEntry { Name = "same", Value = "old" })); - store.PushAll(); - - store.Push(new CultStoredDocument( - key, - storedAt, - descriptor, - new NamedTestEntry { Name = "same", Value = "new" })); - store.FlushStageProbe = (stage, _) => - { - if (string.Equals(stage, DirectoryMessagePackBackingStore.BeforeManifestCommitStage, StringComparison.Ordinal)) - throw new IOException("injected crash before same-timestamp manifest commit"); - }; - Assert.That(() => store.PushAll(), Throws.TypeOf()); - - using var reopened = await CultCacheMessagePack.OpenAsync( - filePath, - new CultCacheOpenOptions { UseDirectoryStore = true }); - Assert.That(reopened.Get(key)?.Value, Is.EqualTo("old")); - } - finally - { - if (File.Exists(filePath)) File.Delete(filePath); - if (Directory.Exists(recordsPath)) Directory.Delete(recordsPath, recursive: true); - } - } - - [Test] - public async Task DirectoryMessagePackBackingStore_MutationDuringFlush_RemainsDirtyAndCommitsNextGeneration() - { - 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 first = await cache.UpsertAsync(new NamedTestEntry { Name = "first", Value = "one" }); - using var flushEntered = new ManualResetEventSlim(); - using var writerAttempting = new ManualResetEventSlim(); - store.FlushStageProbe = (stage, _) => - { - if (!string.Equals(stage, DirectoryMessagePackBackingStore.BeforeManifestCommitStage, StringComparison.Ordinal)) - return; - flushEntered.Set(); - Assert.That(writerAttempting.Wait(TimeSpan.FromSeconds(5)), Is.True); - }; - - var flush = Task.Run(() => store.PushAll()); - var writer = Task.Run(async () => - { - Assert.That(flushEntered.Wait(TimeSpan.FromSeconds(5)), Is.True); - writerAttempting.Set(); - return await cache.UpsertAsync(new NamedTestEntry { Name = "second", Value = "two" }); - }); - await flush; - var second = await writer; - - Assert.That(store.IsDirty, Is.True, "the mutation blocked behind flush must remain pending"); - store.FlushStageProbe = null; - store.PushAll(); - using var reopened = await CultCacheMessagePack.OpenAsync( - filePath, - new CultCacheOpenOptions { UseDirectoryStore = true }); - Assert.That(reopened.Get(first.Key)?.Value, Is.EqualTo("one")); - Assert.That(reopened.Get(second.Key)?.Value, Is.EqualTo("two")); - } - finally - { - if (File.Exists(filePath)) File.Delete(filePath); - if (Directory.Exists(recordsPath)) Directory.Delete(recordsPath, recursive: true); - } - } - [Test] public async Task DirectoryMessagePackBackingStore_ConcurrentInstances_MergeUnderOneCommitLease() { @@ -707,60 +436,6 @@ 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() { @@ -923,9 +598,11 @@ namespace GameCult.Caching.Tests var compacted = CultDocumentMessagePackSerialization.DeserializeSnapshot(File.ReadAllBytes(filePath)); Assert.That(compacted.SchemaCatalog, Is.Empty); Assert.That(compacted.Records, Is.Empty); - var selectedGeneration = store.ReadPersistedGeneration(); - Assert.That(selectedGeneration.Records, Is.Empty, - "a raw generation reader must not resurrect an orphaned page left by another process"); + var reader = new CultCache(); + reader.AddBackingStore(new DirectoryMessagePackBackingStore(filePath, recordsPath)); + await reader.PullAllBackingStoresAsync(); + Assert.That(reader.GetAll(), Is.Empty, + "a reader must not resurrect an orphaned page left by another process"); } finally {