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

Precommit directory store schema catalog

Browse source
This commit is contained in:
Meta Vens 2026-07-13 17:18:07 +01:00
commit fe55bff2e0
2 changed files with 139 additions and 24 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -42,7 +42,7 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
/// <inheritdoc />
public override void PullAll()
{
if (!_manifestFile.Exists && !_recordDirectory.Exists)
if (!File.Exists(_manifestFile.FullName) && !_recordDirectory.Exists)
{
SetLastSchemaMigrationReports(Array.Empty<CultSchemaMigrationReport>());
IsDirty = false;
@ -53,7 +53,7 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
var reports = new List<CultSchemaMigrationReport>();
foreach (var record in manifest.Records)
{
var catalog = ResolveCatalogForRecord(record, manifest.SchemaCatalog);
var catalog = ResolveLegacyUncataloguedRecordCatalog(record, manifest.SchemaCatalog);
reports.Add(Registry.ResolvePersistedSchemaReport(record.SchemaId, catalog));
var stored = ToStoredDocument(record, catalog, CultDocumentMessagePackSerialization.DeserializeUntyped);
Entries[stored.Key.Value] = stored;
@ -66,7 +66,7 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
foreach (var recordFile in _recordDirectory.EnumerateFiles("*.msgpack").OrderBy(file => file.Name, StringComparer.Ordinal))
{
var record = CultDocumentMessagePackSerialization.DeserializePersistedRecord(File.ReadAllBytes(recordFile.FullName));
var catalog = ResolveCatalogForRecord(record, manifest.SchemaCatalog);
var catalog = ResolveLegacyUncataloguedRecordCatalog(record, manifest.SchemaCatalog);
reports.Add(Registry.ResolvePersistedSchemaReport(record.SchemaId, catalog));
var stored = ToStoredDocument(record, catalog, CultDocumentMessagePackSerialization.DeserializeUntyped);
Entries[stored.Key.Value] = stored;
@ -108,14 +108,16 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
Directory.CreateDirectory(_manifestFile.DirectoryName!);
Directory.CreateDirectory(_recordDirectory.FullName);
foreach (var key in _deletedKeys.Keys)
{
var path = RecordPath(key);
if (File.Exists(path))
{
File.Delete(path);
}
}
var targetCatalog = BuildTargetCatalog();
var durableCatalog = ReadManifest().SchemaCatalog;
var precommitCatalog = durableCatalog
.Concat(targetCatalog)
.GroupBy(entry => entry.SchemaId, StringComparer.Ordinal)
.Select(group => group.Last())
.OrderBy(entry => entry.SchemaName, StringComparer.Ordinal)
.ToArray();
WriteManifest(precommitCatalog);
foreach (var key in _dirtyKeys.Keys.OrderBy(value => value, StringComparer.Ordinal))
{
@ -129,27 +131,50 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
WriteFileAtomically(RecordPath(key), CultDocumentMessagePackSerialization.SerializePersistedRecord(record));
}
var manifest = new CultPersistedStoreSnapshot
foreach (var key in _deletedKeys.Keys.OrderBy(value => value, StringComparer.Ordinal))
{
FormatVersion = "cultcache.store.v1.directory",
SchemaCatalog = Entries.Values
.Select(entry => entry.Descriptor.ToCatalogEntry())
.GroupBy(entry => entry.SchemaId, StringComparer.Ordinal)
.Select(group => group.First())
.OrderBy(entry => entry.SchemaName, StringComparer.Ordinal)
.ToArray(),
Records = Array.Empty<CultPersistedRecord>()
};
WriteFileAtomically(_manifestFile.FullName, CultDocumentMessagePackSerialization.SerializeSnapshot(manifest));
var path = RecordPath(key);
if (File.Exists(path))
{
File.Delete(path);
}
}
if (!CatalogsEqual(precommitCatalog, targetCatalog))
{
WriteManifest(targetCatalog);
}
_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)
{
var manifest = new CultPersistedStoreSnapshot
{
FormatVersion = "cultcache.store.v1.directory",
SchemaCatalog = catalog,
Records = Array.Empty<CultPersistedRecord>()
};
WriteFileAtomically(_manifestFile.FullName, CultDocumentMessagePackSerialization.SerializeSnapshot(manifest));
}
private CultPersistedStoreSnapshot ReadManifest()
{
if (_manifestFile.Exists)
if (File.Exists(_manifestFile.FullName))
{
var snapshot = CultDocumentMessagePackSerialization.DeserializeSnapshot(File.ReadAllBytes(_manifestFile.FullName));
if (snapshot.SchemaCatalog.Length > 0 || snapshot.Records.Length == 0)
@ -158,7 +183,7 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
}
}
if (_manifestFile.Exists)
if (File.Exists(_manifestFile.FullName))
{
return CultDocumentMessagePackSerialization.DeserializeSnapshot(File.ReadAllBytes(_manifestFile.FullName));
}
@ -171,7 +196,8 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
};
}
private IReadOnlyCollection<CultSchemaCatalogEntry> ResolveCatalogForRecord(
// Explicit compatibility path for records written before catalog precommit existed.
private IReadOnlyCollection<CultSchemaCatalogEntry> ResolveLegacyUncataloguedRecordCatalog(
CultPersistedRecord record,
IReadOnlyCollection<CultSchemaCatalogEntry> manifestCatalog)
{

89
tests/GameCult.Caching.Tests/BackingStoreTests.cs
View file

@ -282,6 +282,95 @@ namespace GameCult.Caching.Tests
}
}
[Test]
public async Task DirectoryMessagePackBackingStore_Precommits_Catalog_And_Remains_Dirty_When_Record_Write_Fails()
{
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<NamedTestEntry>(key));
store.PushAll();
var recordPath = Directory.GetFiles(recordsPath, "*.msgpack").Single();
var durableSchema = CultDocumentMessagePackSerialization
.DeserializePersistedRecord(File.ReadAllBytes(recordPath))
.SchemaId;
var alternateDescriptor = CultDocumentRegistry.Shared.GetRequired<AlternateNamedTestEntry>();
store.Push(new CultStoredDocument(
key,
DateTimeOffset.UtcNow.ToString("O"),
alternateDescriptor,
new AlternateNamedTestEntry { Name = "new" }));
using (new FileStream(recordPath, FileMode.Open, FileAccess.Read, FileShare.Read))
{
Assert.That(() => store.PushAll(), Throws.Exception);
}
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.Contain(alternateSchema));
Assert.That(store.IsDirty, Is.True);
Assert.That(CultDocumentMessagePackSerialization.DeserializePersistedRecord(File.ReadAllBytes(recordPath)).SchemaId, Is.EqualTo(durableSchema));
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);
}
}
[Test]
public async Task DirectoryMessagePackBackingStore_Remains_Dirty_When_Deletion_Fails()
{
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 handle = await cache.AddAsync(new NamedTestEntry { Name = "delete", Value = "durable" });
store.PushAll();
var recordPath = Directory.GetFiles(recordsPath, "*.msgpack").Single();
Assert.That(cache.Remove(handle.Key), Is.True);
using (new FileStream(recordPath, FileMode.Open, FileAccess.Read, FileShare.Read))
{
Assert.That(() => store.PushAll(), Throws.Exception);
}
Assert.That(File.Exists(recordPath), Is.True);
Assert.That(store.IsDirty, Is.True);
store.PushAll();
Assert.That(File.Exists(recordPath), Is.False);
Assert.That(store.IsDirty, Is.False);
var compacted = CultDocumentMessagePackSerialization.DeserializeSnapshot(File.ReadAllBytes(filePath));
Assert.That(compacted.SchemaCatalog, Is.Empty);
}
finally
{
if (File.Exists(filePath)) File.Delete(filePath);
if (Directory.Exists(recordsPath)) Directory.Delete(recordsPath, recursive: true);
}
}
[Test]
public async Task CultCache_RuntimeType_Upsert_Snapshot_And_Remove_Work_For_Editor_Tooling()
{