GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions 1

Read only v4 directory stores; delete probes and selective hydration

Browse source 
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 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-13 16:50:08 +02:00
commit 900e148d8a
3 changed files with 35 additions and 739 deletions
Download patch file Download diff file Expand all files Collapse all files

410
src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs
View file

@ -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;
/// </summary>
public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
{
/// <summary>Fault-probe stage emitted before each immutable record page write.</summary>
public const string BeforeRecordPageWriteStage = "before-record-page-write";
/// <summary>Fault-probe stage emitted after all pages exist and before manifest finality.</summary>
public const string BeforeManifestCommitStage = "before-manifest-commit";
/// <summary>Read-probe stage emitted while the selected manifest generation is leased.</summary>
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<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;
private bool _manifestUsesImmutablePages;
private bool _manifestUsesMetadataPages;
/// <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>Optional deterministic fault/progress hook for durability tests and host diagnostics.</summary>
public Action<string, int>? FlushStageProbe { get; set; }
/// <summary>Optional deterministic read hook for generation-consistency tests and host diagnostics.</summary>
public Action<string>? ReadStageProbe { get; set; }
/// <summary>
/// Creates a paged MessagePack backing store.
@ -70,57 +45,6 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
/// </summary>
public static string DefaultRecordDirectoryPath(string manifestPath) => manifestPath + ".records";
/// <summary>
/// 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.
/// </summary>
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();
}
/// <inheritdoc />
public override void PullAll()
{
@ -146,46 +70,20 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
var loaded = new Dictionary<string, CultStoredDocument>(StringComparer.Ordinal);
using (AcquireCommitLease())
{
_recordDirectory.Refresh();
if (!File.Exists(_manifestFile.FullName) && !_recordDirectory.Exists)
{
SetLastSchemaMigrationReports(Array.Empty<CultSchemaMigrationReport>());
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");
}
/// <inheritdoc />
public override void PullSelected(Func<CultPersistedRecordMetadata, bool> selector)
{
lock (_mutationGate)
PullSelectedCore(selector);
}
private void PullSelectedCore(Func<CultPersistedRecordMetadata, bool> selector)
{
if (selector == null) throw new ArgumentNullException(nameof(selector));
var fallbackToFullPull = false;
CultPersistedRecord[] selected = Array.Empty<CultPersistedRecord>();
var loaded = new Dictionary<string, CultStoredDocument>(StringComparer.Ordinal);
var reports = new List<CultSchemaMigrationReport>();
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;
}
/// <inheritdoc />
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<string, CultPersistedRecord>(_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<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,
@ -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<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;
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<byte>()
};
private static CultPersistedRecord ToIndexRecord(CultPersistedRecord record) => new()
{
Key = record.Key,
SchemaId = record.SchemaId,
StoredAt = record.StoredAt,
Payload = Array.Empty<byte>()
};
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<CultSchemaCatalogEntry>(),
Records = Array.Empty<CultPersistedRecord>()
};
}
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<CultSchemaCatalogEntry>(),
Records = Array.Empty<CultPersistedRecord>()
};
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)!;