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

Page CultCache MessagePack records

Browse source
This commit is contained in:
Meta Vens 2026-05-31 17:32:42 +01:00
commit 086522029c
3 changed files with 309 additions and 5 deletions
Download patch file Download diff file Expand all files Collapse all files

38
src/GameCult.Caching.MessagePack/CultCacheMessagePack.cs
View file

@ -1,4 +1,5 @@
using System;
using System.IO;
using System.Threading.Tasks;
namespace GameCult.Caching.MessagePack
@ -37,6 +38,21 @@ namespace GameCult.Caching.MessagePack
/// Gets or sets an optional callback used to customize the backing store before opening.
/// </summary>
public Action<SingleFileMessagePackBackingStore>? ConfigureStore { get; set; }
/// <summary>
/// Gets or sets whether the cache should use a paged directory store instead of one whole-file snapshot.
/// </summary>
public bool UseDirectoryStore { get; set; }
/// <summary>
/// Gets or sets the directory used for paged records. When omitted, the store uses the file path plus ".records".
/// </summary>
public string? DirectoryStorePath { get; set; }
/// <summary>
/// Gets or sets an optional callback used to customize the directory backing store before opening.
/// </summary>
public Action<DirectoryMessagePackBackingStore>? ConfigureDirectoryStore { get; set; }
}
/// <summary>
@ -62,13 +78,25 @@ namespace GameCult.Caching.MessagePack
};
options.ConfigureCache?.Invoke(cache);
var store = new SingleFileMessagePackBackingStore(filePath)
if (options.UseDirectoryStore || Directory.Exists(options.DirectoryStorePath ?? DirectoryMessagePackBackingStore.DefaultRecordDirectoryPath(filePath)))
{
FlushOnDispose = options.StoreFlushOnDispose
};
options.ConfigureStore?.Invoke(store);
var directoryStore = new DirectoryMessagePackBackingStore(filePath, options.DirectoryStorePath)
{
FlushOnDispose = options.StoreFlushOnDispose
};
options.ConfigureDirectoryStore?.Invoke(directoryStore);
cache.AddBackingStore(directoryStore);
}
else
{
var store = new SingleFileMessagePackBackingStore(filePath)
{
FlushOnDispose = options.StoreFlushOnDispose
};
options.ConfigureStore?.Invoke(store);
cache.AddBackingStore(store);
}
cache.AddBackingStore(store);
return cache;
}

216
src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs Normal file
View file

@ -0,0 +1,216 @@
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Security.Cryptography;
using System.Text;
namespace GameCult.Caching.MessagePack;
/// <summary>
/// MessagePack backing store that keeps a small schema manifest hot and stores each record as its own cold file.
/// </summary>
public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
{
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);
/// <summary>
/// Creates a paged MessagePack backing store.
/// </summary>
public DirectoryMessagePackBackingStore(string manifestPath, string? recordDirectoryPath = null)
{
if (string.IsNullOrWhiteSpace(manifestPath))
{
throw new ArgumentException("Manifest path must be non-empty.", nameof(manifestPath));
}
_manifestFile = new FileInfo(manifestPath);
_recordDirectory = new DirectoryInfo(recordDirectoryPath ?? DefaultRecordDirectoryPath(manifestPath));
}
/// <summary>
/// Gets the default record directory path for a manifest path.
/// </summary>
public static string DefaultRecordDirectoryPath(string manifestPath) => manifestPath + ".records";
/// <inheritdoc />
public override void PullAll()
{
if (!_manifestFile.Exists && !_recordDirectory.Exists)
{
SetLastSchemaMigrationReports(Array.Empty<CultSchemaMigrationReport>());
IsDirty = false;
return;
}
var manifest = ReadManifest();
var reports = new List<CultSchemaMigrationReport>();
foreach (var record in manifest.Records)
{
reports.Add(Registry.ResolvePersistedSchemaReport(record.SchemaId, manifest.SchemaCatalog));
var stored = ToStoredDocument(record, manifest.SchemaCatalog, CultDocumentMessagePackSerialization.DeserializeUntyped);
Entries[stored.Key.Value] = stored;
_dirtyKeys[stored.Key.Value] = true;
EntryAdded.OnNext(stored);
}
if (_recordDirectory.Exists)
{
foreach (var recordFile in _recordDirectory.EnumerateFiles("*.msgpack").OrderBy(file => file.Name, StringComparer.Ordinal))
{
var record = CultDocumentMessagePackSerialization.DeserializePersistedRecord(File.ReadAllBytes(recordFile.FullName));
reports.Add(Registry.ResolvePersistedSchemaReport(record.SchemaId, manifest.SchemaCatalog));
var stored = ToStoredDocument(record, manifest.SchemaCatalog, CultDocumentMessagePackSerialization.DeserializeUntyped);
Entries[stored.Key.Value] = stored;
EntryAdded.OnNext(stored);
}
}
SetLastSchemaMigrationReports(reports);
if (manifest.Records.Length == 0)
{
_dirtyKeys.Clear();
}
_deletedKeys.Clear();
IsDirty = _dirtyKeys.Count > 0;
}
/// <inheritdoc />
public override void Push(CultStoredDocument entry)
{
Entries[entry.Key.Value] = entry;
_dirtyKeys[entry.Key.Value] = true;
_deletedKeys.TryRemove(entry.Key.Value, out _);
IsDirty = true;
}
/// <inheritdoc />
public override void Delete(CultStoredDocument entry)
{
Entries.TryRemove(entry.Key.Value, out _);
_dirtyKeys.TryRemove(entry.Key.Value, out _);
_deletedKeys[entry.Key.Value] = true;
IsDirty = true;
}
/// <inheritdoc />
public override void PushAll(bool soft = false)
{
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);
}
}
foreach (var key in _dirtyKeys.Keys.OrderBy(value => value, StringComparer.Ordinal))
{
if (!Entries.TryGetValue(key, out var stored))
{
continue;
}
var record = ToPersistedRecord(stored, document =>
CultDocumentMessagePackSerialization.SerializeUntyped(document, stored.Descriptor.DocumentType));
WriteFileAtomically(RecordPath(key), CultDocumentMessagePackSerialization.SerializePersistedRecord(record));
}
var manifest = new CultPersistedStoreSnapshot
{
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));
_dirtyKeys.Clear();
_deletedKeys.Clear();
MarkFlushSucceeded();
}
private CultPersistedStoreSnapshot ReadManifest()
{
if (_manifestFile.Exists)
{
var snapshot = CultDocumentMessagePackSerialization.DeserializeSnapshot(File.ReadAllBytes(_manifestFile.FullName));
if (snapshot.SchemaCatalog.Length > 0 || snapshot.Records.Length == 0)
{
return snapshot;
}
}
if (_manifestFile.Exists)
{
return CultDocumentMessagePackSerialization.DeserializeSnapshot(File.ReadAllBytes(_manifestFile.FullName));
}
return new CultPersistedStoreSnapshot
{
FormatVersion = "cultcache.store.v1.directory",
SchemaCatalog = Array.Empty<CultSchemaCatalogEntry>(),
Records = Array.Empty<CultPersistedRecord>()
};
}
private string RecordPath(string key) => Path.Combine(_recordDirectory.FullName, $"{HashKey(key)}.msgpack");
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)!;
Directory.CreateDirectory(directory);
var tempPath = Path.Combine(directory, $".{Path.GetFileName(path)}.{Guid.NewGuid():N}.tmp");
try
{
using (var stream = new FileStream(
tempPath,
FileMode.CreateNew,
FileAccess.Write,
FileShare.None,
bufferSize: 4096,
FileOptions.WriteThrough))
{
stream.Write(payload, 0, payload.Length);
stream.Flush(true);
}
if (File.Exists(path))
{
File.Replace(tempPath, path, null);
}
else
{
File.Move(tempPath, path);
}
}
finally
{
if (File.Exists(tempPath))
{
File.Delete(tempPath);
}
}
}
}

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

@ -122,6 +122,66 @@ namespace GameCult.Caching.Tests
}
}
[Test]
public async Task DirectoryMessagePackBackingStore_Writes_Record_Pages_Without_Rewriting_Cold_Records()
{
var filePath = Path.Combine(Path.GetTempPath(), $"cultlib-tests-{Guid.NewGuid():N}.cc");
var recordsPath = DirectoryMessagePackBackingStore.DefaultRecordDirectoryPath(filePath);
try
{
var cache = await CultCacheMessagePack.OpenAsync(
filePath,
new CultCacheOpenOptions { UseDirectoryStore = true });
var first = await cache.UpsertAsync(new NamedTestEntry
{
Name = "cold",
Value = new string('a', 4096)
});
await cache.FlushAsync();
cache.Dispose();
var firstRecord = Directory.GetFiles(recordsPath, "*.msgpack").Single();
var firstWrite = File.GetLastWriteTimeUtc(firstRecord);
await Task.Delay(1100);
var reopened = await CultCacheMessagePack.OpenAsync(
filePath,
new CultCacheOpenOptions { UseDirectoryStore = true });
var second = await reopened.UpsertAsync(new NamedTestEntry
{
Name = "hot",
Value = "new"
});
await reopened.FlushAsync();
reopened.Dispose();
var recordFiles = Directory.GetFiles(recordsPath, "*.msgpack");
Assert.That(recordFiles, Has.Length.EqualTo(2));
Assert.That(File.GetLastWriteTimeUtc(firstRecord), Is.EqualTo(firstWrite));
Assert.That(new FileInfo(filePath).Length, Is.LessThan(new FileInfo(firstRecord).Length));
var read = await CultCacheMessagePack.OpenAsync(
filePath,
new CultCacheOpenOptions { UseDirectoryStore = true });
Assert.That(read.Get<NamedTestEntry>(first.Key)!.Value, Is.EqualTo(new string('a', 4096)));
Assert.That(read.Get<NamedTestEntry>(second.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 CultCache_RuntimeType_Upsert_Snapshot_And_Remove_Work_For_Editor_Tooling()
{