Make CultCache hydration subscription-shaped
This commit is contained in:
Meta Vens
parent
54e2e69ff7
commit
085e177742
8 changed files with 554 additions and 72 deletions
|
|
@ -16,10 +16,22 @@ namespace GameCult.Caching.MessagePack;
|
|||
/// </summary>
|
||||
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<string, bool> _dirtyKeys = new(StringComparer.Ordinal);
|
||||
private readonly ConcurrentDictionary<string, bool> _deletedKeys = new(StringComparer.Ordinal);
|
||||
private readonly HashSet<string> _hydratedKeys = new(StringComparer.Ordinal);
|
||||
private Dictionary<string, CultPersistedRecord> _durableIndex = new(StringComparer.Ordinal);
|
||||
private Dictionary<string, CultPersistedRecord> _legacyInlineRecords = new(StringComparer.Ordinal);
|
||||
private CultSchemaCatalogEntry[] _durableCatalog = Array.Empty<CultSchemaCatalogEntry>();
|
||||
private bool _needsIndexUpgrade;
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the predicate that selects indexed record payloads for hydration.
|
||||
/// A null predicate hydrates every record.
|
||||
/// </summary>
|
||||
public Func<CultPersistedRecordMetadata, bool>? HydrationFilter { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// 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<CultSchemaMigrationReport>();
|
||||
var loaded = new Dictionary<string, CultStoredDocument>(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;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public override void PullSelected(Func<CultPersistedRecordMetadata, bool> 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<string, CultStoredDocument>(StringComparer.Ordinal);
|
||||
var reports = new List<CultSchemaMigrationReport>();
|
||||
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<string, CultPersistedRecord>(_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<CultPersistedRecord>()
|
||||
Records = index
|
||||
};
|
||||
WriteFileAtomically(_manifestFile.FullName, CultDocumentMessagePackSerialization.SerializeSnapshot(manifest));
|
||||
}
|
||||
|
||||
private void LoadIndexedRecords(
|
||||
CultPersistedStoreSnapshot manifest,
|
||||
Dictionary<string, CultStoredDocument> loaded,
|
||||
List<CultSchemaMigrationReport> 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<string, CultStoredDocument> loaded,
|
||||
List<CultSchemaMigrationReport> 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<CultSchemaCatalogEntry> catalogEntries,
|
||||
Dictionary<string, CultStoredDocument> loaded,
|
||||
List<CultSchemaMigrationReport> 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<CultSchemaCatalogEntry> catalogEntries,
|
||||
Dictionary<string, CultStoredDocument> loaded,
|
||||
List<CultSchemaMigrationReport> 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<string, CultStoredDocument> loaded,
|
||||
HashSet<string> 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<byte>()
|
||||
};
|
||||
|
||||
private static CultPersistedRecord ToIndexRecord(CultPersistedRecord record) => new()
|
||||
{
|
||||
Key = record.Key,
|
||||
SchemaId = record.SchemaId,
|
||||
StoredAt = record.StoredAt,
|
||||
Payload = Array.Empty<byte>()
|
||||
};
|
||||
|
||||
private CultPersistedStoreSnapshot ReadManifest()
|
||||
{
|
||||
if (File.Exists(_manifestFile.FullName))
|
||||
|
|
|
|||
Loading…
Reference in a new issue