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
{