diff --git a/src/GameCult.Caching.MessagePack/CultCacheMessagePack.cs b/src/GameCult.Caching.MessagePack/CultCacheMessagePack.cs
index ad2d73a3..3a57d960 100644
--- a/src/GameCult.Caching.MessagePack/CultCacheMessagePack.cs
+++ b/src/GameCult.Caching.MessagePack/CultCacheMessagePack.cs
@@ -49,6 +49,12 @@ namespace GameCult.Caching.MessagePack
///
public string? DirectoryStorePath { get; set; }
+ ///
+ /// Gets or sets the records to hydrate when opening a paged directory store.
+ /// The directory manifest is always read; rejected record payloads are never opened.
+ ///
+ public Func? DirectoryStoreHydrationFilter { get; set; }
+
///
/// Gets or sets an optional callback used to customize the directory backing store before opening.
///
@@ -82,7 +88,8 @@ namespace GameCult.Caching.MessagePack
{
var directoryStore = new DirectoryMessagePackBackingStore(filePath, options.DirectoryStorePath)
{
- FlushOnDispose = options.StoreFlushOnDispose
+ FlushOnDispose = options.StoreFlushOnDispose,
+ HydrationFilter = options.DirectoryStoreHydrationFilter
};
options.ConfigureDirectoryStore?.Invoke(directoryStore);
cache.AddBackingStore(directoryStore);
diff --git a/src/GameCult.Caching.MessagePack/CultDocumentMessagePackSerialization.cs b/src/GameCult.Caching.MessagePack/CultDocumentMessagePackSerialization.cs
index 21278fa8..00ed2993 100644
--- a/src/GameCult.Caching.MessagePack/CultDocumentMessagePackSerialization.cs
+++ b/src/GameCult.Caching.MessagePack/CultDocumentMessagePackSerialization.cs
@@ -75,9 +75,18 @@ public static class CultDocumentMessagePackSerialization
///
public static byte[] SerializeUntyped(object value, Type type)
{
+ return SerializeUntyped(value, type, CultDocumentRegistry.Shared);
+ }
+
+ ///
+ /// Serializes a value through an explicitly owned document registry.
+ ///
+ public static byte[] SerializeUntyped(object value, Type type, CultDocumentRegistry registry)
+ {
+ if (registry == null) throw new ArgumentNullException(nameof(registry));
if (value != null)
{
- var descriptor = CultDocumentRegistry.Shared.GetRequired(type);
+ var descriptor = registry.GetRequired(type);
if (descriptor.GeneratedPayloadSerializer != null)
{
return descriptor.GeneratedPayloadSerializer(value);
@@ -92,7 +101,16 @@ public static class CultDocumentMessagePackSerialization
///
public static object DeserializeUntyped(Type type, byte[] payload)
{
- var descriptor = CultDocumentRegistry.Shared.GetRequired(type);
+ return DeserializeUntyped(type, payload, CultDocumentRegistry.Shared);
+ }
+
+ ///
+ /// Deserializes a value through an explicitly owned document registry.
+ ///
+ public static object DeserializeUntyped(Type type, byte[] payload, CultDocumentRegistry registry)
+ {
+ if (registry == null) throw new ArgumentNullException(nameof(registry));
+ var descriptor = registry.GetRequired(type);
if (descriptor.GeneratedPayloadDeserializer != null)
{
return descriptor.GeneratedPayloadDeserializer(payload);
@@ -444,7 +462,7 @@ public class SingleFileMessagePackBackingStore : SingleFileBackingStore
///
protected override byte[] SerializePayload(object document)
{
- return CultDocumentMessagePackSerialization.SerializeUntyped(document, document.GetType());
+ return CultDocumentMessagePackSerialization.SerializeUntyped(document, document.GetType(), Registry);
}
///
@@ -452,6 +470,6 @@ public class SingleFileMessagePackBackingStore : SingleFileBackingStore
///
protected override object DeserializePayload(Type documentType, byte[] payload)
{
- return CultDocumentMessagePackSerialization.DeserializeUntyped(documentType, payload);
+ return CultDocumentMessagePackSerialization.DeserializeUntyped(documentType, payload, Registry);
}
}
diff --git a/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs b/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs
index 58bbcae6..8deafeac 100644
--- a/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs
+++ b/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs
@@ -16,10 +16,22 @@ namespace GameCult.Caching.MessagePack;
///
public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
{
+ private const string IndexedFormatVersion = "cultcache.store.v2.directory-indexed";
private readonly FileInfo _manifestFile;
private readonly DirectoryInfo _recordDirectory;
private readonly ConcurrentDictionary _dirtyKeys = new(StringComparer.Ordinal);
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;
+
+ ///
+ /// Gets or sets the predicate that selects indexed record payloads for hydration.
+ /// A null predicate hydrates every record.
+ ///
+ public Func? HydrationFilter { get; set; }
///
/// Creates a paged MessagePack backing store.
@@ -52,46 +64,30 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
}
var manifest = ReadManifest();
+ _durableCatalog = manifest.SchemaCatalog;
var reports = new List();
var loaded = new Dictionary(StringComparer.Ordinal);
- foreach (var record in manifest.Records)
+ var indexed = string.Equals(manifest.FormatVersion, IndexedFormatVersion, StringComparison.Ordinal);
+ if (indexed)
{
- var catalog = ResolveLegacyUncataloguedRecordCatalog(record, manifest.SchemaCatalog);
- reports.Add(Registry.ResolvePersistedSchemaReport(record.SchemaId, catalog));
- var stored = ToStoredDocument(record, catalog, CultDocumentMessagePackSerialization.DeserializeUntyped);
- loaded[stored.Key.Value] = stored;
+ _needsIndexUpgrade = false;
+ _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);
}
-
- if (_recordDirectory.Exists)
+ else
{
- var recordFiles = _recordDirectory
- .EnumerateFiles("*.msgpack")
- .OrderBy(file => file.Name, StringComparer.Ordinal)
- .ToArray();
- var recordReports = new CultSchemaMigrationReport[recordFiles.Length];
- var storedRecords = new CultStoredDocument[recordFiles.Length];
- Parallel.For(
- 0,
- recordFiles.Length,
- new ParallelOptions { MaxDegreeOfParallelism = Math.Min(Environment.ProcessorCount, 8) },
- index =>
- {
- var record = CultDocumentMessagePackSerialization.DeserializePersistedRecord(
- ReadAllBytesShared(recordFiles[index].FullName));
- var catalog = ResolveLegacyUncataloguedRecordCatalog(record, manifest.SchemaCatalog);
- recordReports[index] = Registry.ResolvePersistedSchemaReport(record.SchemaId, catalog);
- storedRecords[index] = ToStoredDocument(
- record,
- catalog,
- CultDocumentMessagePackSerialization.DeserializeUntyped);
- });
-
- for (var index = 0; index < storedRecords.Length; index++)
- {
- reports.Add(recordReports[index]);
- var stored = storedRecords[index];
- loaded[stored.Key.Value] = stored;
- }
+ // 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);
}
foreach (var pair in loaded)
@@ -113,15 +109,47 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
EntryUpdated.OnNext(pair.Value);
}
- foreach (var removedKey in Entries.Keys.Where(key => !loaded.ContainsKey(key)).ToArray())
+ foreach (var removedKey in _hydratedKeys.Where(key => !loaded.ContainsKey(key)).ToArray())
{
if (_dirtyKeys.ContainsKey(removedKey) || _deletedKeys.ContainsKey(removedKey))
continue;
if (Entries.TryRemove(removedKey, out var removed))
EntryDeleted.OnNext(removed);
+ _hydratedKeys.Remove(removedKey);
}
+ foreach (var key in loaded.Keys)
+ _hydratedKeys.Add(key);
+
+ SetLastSchemaMigrationReports(reports);
+ IsDirty = !_dirtyKeys.IsEmpty || !_deletedKeys.IsEmpty;
+ }
+
+ ///
+ public override void PullSelected(Func selector)
+ {
+ if (selector == null) throw new ArgumentNullException(nameof(selector));
+ var manifest = ReadManifest();
+ if (!string.Equals(manifest.FormatVersion, IndexedFormatVersion, StringComparison.Ordinal))
+ {
+ PullAll();
+ return;
+ }
+
+ _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;
}
@@ -150,18 +178,46 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
Directory.CreateDirectory(_manifestFile.DirectoryName!);
Directory.CreateDirectory(_recordDirectory.FullName);
- var targetCatalog = BuildTargetCatalog();
- var durableCatalog = ReadManifest().SchemaCatalog;
- var precommitCatalog = durableCatalog
+ var currentManifest = ReadManifest();
+ var currentIndex = string.Equals(currentManifest.FormatVersion, IndexedFormatVersion, StringComparison.Ordinal)
+ ? currentManifest.Records.ToDictionary(record => record.Key, record => record, StringComparer.Ordinal)
+ : new Dictionary(_durableIndex, StringComparer.Ordinal);
+ var precommitIndex = currentIndex.Values
+ .Select(ToIndexRecord)
+ .OrderBy(record => record.Key, StringComparer.Ordinal)
+ .ToArray();
+ foreach (var key in _deletedKeys.Keys)
+ currentIndex.Remove(key);
+ foreach (var key in _dirtyKeys.Keys)
+ {
+ if (Entries.TryGetValue(key, out var stored))
+ currentIndex[key] = ToIndexRecord(stored);
+ }
+
+ var catalogCandidates = currentManifest.SchemaCatalog
+ .Concat(_durableCatalog)
+ .Concat(Entries.Values.Select(entry => entry.Descriptor.ToCatalogEntry()))
+ .GroupBy(entry => entry.SchemaId, StringComparer.Ordinal)
+ .Select(group => group.Last())
+ .ToArray();
+ var usedSchemaIds = currentIndex.Values.Select(record => record.SchemaId).ToHashSet(StringComparer.Ordinal);
+ var targetCatalog = catalogCandidates
+ .Where(entry => usedSchemaIds.Contains(entry.SchemaId))
+ .OrderBy(entry => entry.SchemaName, StringComparer.Ordinal)
+ .ToArray();
+ var precommitCatalog = currentManifest.SchemaCatalog
.Concat(targetCatalog)
.GroupBy(entry => entry.SchemaId, StringComparer.Ordinal)
.Select(group => group.Last())
.OrderBy(entry => entry.SchemaName, StringComparer.Ordinal)
.ToArray();
- WriteManifest(precommitCatalog);
+ WriteManifest(precommitCatalog, precommitIndex);
- foreach (var key in _dirtyKeys.Keys.OrderBy(value => value, StringComparer.Ordinal))
+ var keysToWrite = (_needsIndexUpgrade ? Entries.Keys : _dirtyKeys.Keys)
+ .OrderBy(value => value, StringComparer.Ordinal)
+ .ToArray();
+ foreach (var key in keysToWrite)
{
if (!Entries.TryGetValue(key, out var stored))
{
@@ -169,10 +225,22 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
}
var record = ToPersistedRecord(stored, document =>
- CultDocumentMessagePackSerialization.SerializeUntyped(document, stored.Descriptor.DocumentType));
+ CultDocumentMessagePackSerialization.SerializeUntyped(
+ document,
+ stored.Descriptor.DocumentType,
+ Registry));
WriteFileAtomically(RecordPath(key), CultDocumentMessagePackSerialization.SerializePersistedRecord(record));
}
+ foreach (var pair in _legacyInlineRecords.OrderBy(pair => pair.Key, StringComparer.Ordinal))
+ {
+ if (_deletedKeys.ContainsKey(pair.Key) || _dirtyKeys.ContainsKey(pair.Key))
+ continue;
+ WriteFileAtomically(
+ RecordPath(pair.Key),
+ CultDocumentMessagePackSerialization.SerializePersistedRecord(pair.Value));
+ }
+
foreach (var key in _deletedKeys.Keys.OrderBy(value => value, StringComparer.Ordinal))
{
var path = RecordPath(key);
@@ -182,38 +250,199 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
}
}
- if (!CatalogsEqual(precommitCatalog, targetCatalog))
- {
- WriteManifest(targetCatalog);
- }
+ WriteManifest(targetCatalog, currentIndex.Values
+ .OrderBy(record => record.Key, StringComparer.Ordinal)
+ .ToArray());
+ _durableCatalog = targetCatalog;
+ _durableIndex = currentIndex;
+ _legacyInlineRecords.Clear();
+ _needsIndexUpgrade = false;
_dirtyKeys.Clear();
_deletedKeys.Clear();
MarkFlushSucceeded();
}
- private CultSchemaCatalogEntry[] BuildTargetCatalog() => Entries.Values
- .Select(entry => entry.Descriptor.ToCatalogEntry())
- .GroupBy(entry => entry.SchemaId, StringComparer.Ordinal)
- .Select(group => group.First())
- .OrderBy(entry => entry.SchemaName, StringComparer.Ordinal)
- .ToArray();
-
- private static bool CatalogsEqual(CultSchemaCatalogEntry[] left, CultSchemaCatalogEntry[] right) =>
- CultDocumentMessagePackSerialization.SerializeSchemaCatalog(left)
- .SequenceEqual(CultDocumentMessagePackSerialization.SerializeSchemaCatalog(right));
-
- private void WriteManifest(CultSchemaCatalogEntry[] catalog)
+ private void WriteManifest(CultSchemaCatalogEntry[] catalog, CultPersistedRecord[] index)
{
var manifest = new CultPersistedStoreSnapshot
{
- FormatVersion = "cultcache.store.v1.directory",
+ FormatVersion = IndexedFormatVersion,
SchemaCatalog = catalog,
- Records = Array.Empty()
+ Records = index
};
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,
+ 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 metadata = records[index];
+ var path = RecordPath(metadata.Key);
+ if (!File.Exists(path))
+ return;
+ var record = CultDocumentMessagePackSerialization.DeserializePersistedRecord(ReadAllBytesShared(path));
+ if (!string.Equals(record.Key, metadata.Key, StringComparison.Ordinal))
+ throw new InvalidDataException($"Record page '{path}' contains key '{record.Key}', expected '{metadata.Key}'.");
+ 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++)
+ {
+ var stored = storedRecords[index];
+ if (stored == null)
+ continue;
+ reports.Add(recordReports[index]);
+ loaded[stored.Key.Value] = stored;
+ }
+ }
+
+ 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;
+ if (existing == null)
+ EntryAdded.OnNext(pair.Value);
+ else
+ EntryUpdated.OnNext(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))
+ EntryDeleted.OnNext(removed);
+ _hydratedKeys.Remove(missingKey);
+ }
+
+ foreach (var key in loaded.Keys)
+ _hydratedKeys.Add(key);
+ }
+
+ private static CultPersistedRecord ToIndexRecord(CultStoredDocument stored) => new()
+ {
+ Key = stored.Key.Value,
+ SchemaId = stored.Descriptor.SchemaId,
+ StoredAt = stored.StoredAt,
+ Payload = Array.Empty()
+ };
+
+ private static CultPersistedRecord ToIndexRecord(CultPersistedRecord record) => new()
+ {
+ Key = record.Key,
+ SchemaId = record.SchemaId,
+ StoredAt = record.StoredAt,
+ Payload = Array.Empty()
+ };
+
private CultPersistedStoreSnapshot ReadManifest()
{
if (File.Exists(_manifestFile.FullName))
diff --git a/src/GameCult.Caching/CultCache.cs b/src/GameCult.Caching/CultCache.cs
index b06391be..a580604b 100644
--- a/src/GameCult.Caching/CultCache.cs
+++ b/src/GameCult.Caching/CultCache.cs
@@ -420,6 +420,28 @@ namespace GameCult.Caching
Refresh();
}
+ private CultDocumentRegistry(bool discoverLoadedDocuments)
+ {
+ if (discoverLoadedDocuments)
+ Refresh();
+ }
+
+ ///
+ /// Creates a registry containing exactly the requested document types without scanning loaded assemblies.
+ ///
+ public static CultDocumentRegistry ForTypes(IEnumerable documentTypes)
+ {
+ if (documentTypes == null) throw new ArgumentNullException(nameof(documentTypes));
+ var registry = new CultDocumentRegistry(discoverLoadedDocuments: false);
+ foreach (var type in documentTypes.Distinct())
+ {
+ registry.GetRequired(type ?? throw new ArgumentException(
+ "Document type collections cannot contain null entries.",
+ nameof(documentTypes)));
+ }
+ return registry;
+ }
+
///
/// Gets all known document descriptors.
///
@@ -1392,6 +1414,22 @@ namespace GameCult.Caching
await Task.CompletedTask;
}
+ ///
+ /// Pulls persisted records selected by durable metadata from every attached backing store.
+ /// Stores without indexed selection support preserve correctness by performing a full pull.
+ ///
+ public async Task PullBackingStoreRecordsAsync(Func selector)
+ {
+ if (selector == null) throw new ArgumentNullException(nameof(selector));
+ foreach (var store in _backingStores)
+ {
+ store.PullSelected(selector);
+ }
+
+ _hasUnflushedMutations = _backingStores.Any(store => store.IsDirty);
+ await Task.CompletedTask;
+ }
+
///
/// Flushes all attached backing stores.
///
@@ -1892,6 +1930,27 @@ 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.
///
@@ -1970,6 +2029,14 @@ 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();
+ }
+ ///
/// Pushes one stored document into the backing store.
///
public abstract void Push(CultStoredDocument entry);
diff --git a/src/GameCult.Mesh/CultMeshDocumentRegistries.cs b/src/GameCult.Mesh/CultMeshDocumentRegistries.cs
index a210671a..352b2641 100644
--- a/src/GameCult.Mesh/CultMeshDocumentRegistries.cs
+++ b/src/GameCult.Mesh/CultMeshDocumentRegistries.cs
@@ -26,13 +26,7 @@ namespace GameCult.Mesh
throw new ArgumentNullException(nameof(documentTypes));
}
- var registry = new CultDocumentRegistry();
- foreach (var documentType in DistinctDocumentTypes(documentTypes))
- {
- registry.GetRequired(documentType);
- }
-
- return registry;
+ return CultDocumentRegistry.ForTypes(DistinctDocumentTypes(documentTypes));
}
///
diff --git a/src/GameCult.Mesh/docs/reliability-migration-plan.md b/src/GameCult.Mesh/docs/reliability-migration-plan.md
index 72e60ca9..658d5e0a 100644
--- a/src/GameCult.Mesh/docs/reliability-migration-plan.md
+++ b/src/GameCult.Mesh/docs/reliability-migration-plan.md
@@ -47,6 +47,34 @@ usable while endpoints fail or move.
## Target Authority Map
+### Local persistence hydration
+
+**Owner:** the attached CultCache backing store owns durable record discovery;
+the consuming service owns its working-set selector.
+
+**Inputs:** an indexed manifest containing record key, schema identity, and
+commit time; an explicit metadata predicate supplied at open or at a later
+subscription boundary.
+
+**Outputs:** only selected typed documents enter the live cache. Record payload
+pages outside the working set remain durable and unopened. Indexed stores may
+hydrate an additional selection without replaying or evicting the existing hot
+set; non-indexed stores retain the correct full-pull fallback.
+
+**Derived state:** hydrated keys, dirty keys, and migration reports are local
+backing-store state. The manifest index remains the durable catalog even when a
+consumer has only a narrow working set in memory.
+
+**Forbidden writers:** service boot code must not scan page directories or
+deserialize broad snapshots to discover keys. A filtered flush must not delete
+unsubscribed records, and implicit global assembly discovery must not expand an
+explicit service registry.
+
+**Cut line:** directory manifests now carry metadata-only record indexes;
+explicit registries contain exactly the requested document types; selective
+pulls use the same indexed authority instead of reopening the store through a
+second compatibility path.
+
### Discovery control plane
**Owner:** `CultMeshDiscoveryService`.
diff --git a/tests/GameCult.Caching.Tests/BackingStoreTests.cs b/tests/GameCult.Caching.Tests/BackingStoreTests.cs
index 9e9a7606..e8fdd884 100644
--- a/tests/GameCult.Caching.Tests/BackingStoreTests.cs
+++ b/tests/GameCult.Caching.Tests/BackingStoreTests.cs
@@ -321,6 +321,133 @@ namespace GameCult.Caching.Tests
}
}
+ [Test]
+ public async Task DirectoryMessagePackBackingStore_IndexedFilter_DoesNotOpenOrDeleteColdPayloads()
+ {
+ var filePath = Path.Combine(Path.GetTempPath(), $"cultlib-tests-{Guid.NewGuid():N}.cc");
+ var recordsPath = DirectoryMessagePackBackingStore.DefaultRecordDirectoryPath(filePath);
+
+ try
+ {
+ CultRecordHandle hot;
+ CultRecordHandle cold;
+ using (var seed = await CultCacheMessagePack.OpenAsync(
+ filePath,
+ new CultCacheOpenOptions { UseDirectoryStore = true }))
+ {
+ hot = await seed.UpsertAsync(new NamedTestEntry { Name = "hot", Value = "hydrate-me" });
+ cold = await seed.UpsertAsync(new NamedTestEntry
+ {
+ Name = "cold",
+ Value = new string('c', 1024 * 1024)
+ });
+ await seed.FlushAsync();
+ }
+
+ var manifest = CultDocumentMessagePackSerialization.DeserializeSnapshot(File.ReadAllBytes(filePath));
+ Assert.That(manifest.FormatVersion, Is.EqualTo("cultcache.store.v2.directory-indexed"));
+ Assert.That(manifest.Records, Has.Length.EqualTo(2));
+ Assert.That(manifest.Records.All(record => record.Payload.Length == 0), Is.True);
+
+ var coldPath = Directory.GetFiles(recordsPath, "*.msgpack")
+ .Single(path => string.Equals(
+ CultDocumentMessagePackSerialization.DeserializePersistedRecord(File.ReadAllBytes(path)).Key,
+ cold.Key.Value,
+ StringComparison.Ordinal));
+
+ using (new FileStream(coldPath, FileMode.Open, FileAccess.Read, FileShare.None))
+ using (var selected = await CultCacheMessagePack.OpenAsync(
+ filePath,
+ new CultCacheOpenOptions
+ {
+ UseDirectoryStore = true,
+ DirectoryStoreHydrationFilter = metadata =>
+ string.Equals(metadata.Key, hot.Key.Value, StringComparison.Ordinal)
+ }))
+ {
+ Assert.That(selected.Get(hot.Key)?.Value, Is.EqualTo("hydrate-me"));
+ Assert.That(selected.Get(cold.Key), Is.Null);
+ await selected.UpsertAsync(new NamedTestEntry { Name = "new", Value = "persist-with-cold-page-locked" });
+ await selected.FlushAsync();
+ }
+
+ using var reopened = await CultCacheMessagePack.OpenAsync(
+ filePath,
+ new CultCacheOpenOptions { UseDirectoryStore = true });
+ Assert.That(reopened.Get(cold.Key)?.Value.Length, Is.EqualTo(1024 * 1024));
+ Assert.That(reopened.GetAll().Count(), Is.EqualTo(3));
+
+ var finalManifest = CultDocumentMessagePackSerialization.DeserializeSnapshot(File.ReadAllBytes(filePath));
+ Assert.That(finalManifest.Records, Has.Length.EqualTo(3));
+ Assert.That(finalManifest.Records.Any(record => record.Key == cold.Key.Value), Is.True);
+ Assert.That(finalManifest.Records.All(record => record.Payload.Length == 0), Is.True);
+ }
+ finally
+ {
+ if (File.Exists(filePath))
+ File.Delete(filePath);
+ if (Directory.Exists(recordsPath))
+ Directory.Delete(recordsPath, recursive: true);
+ }
+ }
+
+ [Test]
+ public async Task DirectoryMessagePackBackingStore_PullSelected_HydratesMatchingColdPagesOnly()
+ {
+ var filePath = Path.Combine(Path.GetTempPath(), $"cultlib-tests-{Guid.NewGuid():N}.cc");
+ var recordsPath = DirectoryMessagePackBackingStore.DefaultRecordDirectoryPath(filePath);
+
+ try
+ {
+ CultRecordHandle hot;
+ CultRecordHandle requested;
+ CultRecordHandle unrelated;
+ using (var seed = await CultCacheMessagePack.OpenAsync(
+ filePath,
+ new CultCacheOpenOptions { UseDirectoryStore = true }))
+ {
+ hot = await seed.UpsertAsync(new NamedTestEntry { Name = "hot", Value = "already-loaded" });
+ requested = await seed.UpsertAsync(new NamedTestEntry { Name = "requested", Value = "load-later" });
+ unrelated = await seed.UpsertAsync(new NamedTestEntry
+ {
+ Name = "unrelated",
+ Value = new string('u', 1024 * 1024)
+ });
+ await seed.FlushAsync();
+ }
+
+ string PageFor(CultRecordHandle handle) =>
+ Directory.GetFiles(recordsPath, "*.msgpack").Single(path => string.Equals(
+ CultDocumentMessagePackSerialization.DeserializePersistedRecord(File.ReadAllBytes(path)).Key,
+ handle.Key.Value,
+ StringComparison.Ordinal));
+
+ using var selected = await CultCacheMessagePack.OpenAsync(
+ filePath,
+ new CultCacheOpenOptions
+ {
+ UseDirectoryStore = true,
+ DirectoryStoreHydrationFilter = metadata => metadata.Key == hot.Key.Value
+ });
+ Assert.That(selected.Get(requested.Key), Is.Null);
+
+ using (new FileStream(PageFor(unrelated), FileMode.Open, FileAccess.Read, FileShare.None))
+ {
+ await selected.PullBackingStoreRecordsAsync(metadata => metadata.Key == requested.Key.Value);
+ Assert.That(selected.Get(requested.Key)?.Value, Is.EqualTo("load-later"));
+ Assert.That(selected.Get(hot.Key)?.Value, Is.EqualTo("already-loaded"));
+ Assert.That(selected.Get(unrelated.Key), Is.Null);
+ }
+ }
+ finally
+ {
+ if (File.Exists(filePath))
+ File.Delete(filePath);
+ if (Directory.Exists(recordsPath))
+ Directory.Delete(recordsPath, recursive: true);
+ }
+ }
+
[Test]
public async Task DirectoryMessagePackBackingStore_Loads_Record_When_Manifest_Misses_Catalog_Entry()
{
diff --git a/tests/GameCult.Caching.Tests/CultDocumentRegistryTests.cs b/tests/GameCult.Caching.Tests/CultDocumentRegistryTests.cs
index bb80ce98..1a1e83a3 100644
--- a/tests/GameCult.Caching.Tests/CultDocumentRegistryTests.cs
+++ b/tests/GameCult.Caching.Tests/CultDocumentRegistryTests.cs
@@ -10,6 +10,18 @@ namespace GameCult.Caching.Tests;
[TestFixture]
public sealed class CultDocumentRegistryTests
{
+ [Test]
+ public void ForTypesContainsOnlyExplicitDocuments()
+ {
+ var requested = CreateDocumentType("explicit", "tests.registry.explicit", "v1");
+ var unrelated = CreateDocumentType("unrelated", "tests.registry.unrelated", "v1");
+
+ var registry = CultDocumentRegistry.ForTypes(new[] { requested, requested });
+
+ Assert.That(registry.AllDescriptors.Select(value => value.DocumentType), Is.EqualTo(new[] { requested }));
+ Assert.That(registry.AllDescriptors.Any(value => value.DocumentType == unrelated), Is.False);
+ }
+
[Test]
public void IdenticalTypeRegistrationIsIdempotent()
{