diff --git a/src/GameCult.Caching.MessagePack/CultCacheMessagePack.cs b/src/GameCult.Caching.MessagePack/CultCacheMessagePack.cs index 7e32c799..ad2d73a3 100644 --- a/src/GameCult.Caching.MessagePack/CultCacheMessagePack.cs +++ b/src/GameCult.Caching.MessagePack/CultCacheMessagePack.cs @@ -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. /// public Action? ConfigureStore { get; set; } + + /// + /// Gets or sets whether the cache should use a paged directory store instead of one whole-file snapshot. + /// + public bool UseDirectoryStore { get; set; } + + /// + /// Gets or sets the directory used for paged records. When omitted, the store uses the file path plus ".records". + /// + public string? DirectoryStorePath { get; set; } + + /// + /// Gets or sets an optional callback used to customize the directory backing store before opening. + /// + public Action? ConfigureDirectoryStore { get; set; } } /// @@ -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; } diff --git a/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs b/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs new file mode 100644 index 00000000..991037da --- /dev/null +++ b/src/GameCult.Caching.MessagePack/DirectoryMessagePackBackingStore.cs @@ -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; + +/// +/// MessagePack backing store that keeps a small schema manifest hot and stores each record as its own cold file. +/// +public sealed class DirectoryMessagePackBackingStore : CacheBackingStore +{ + private readonly FileInfo _manifestFile; + private readonly DirectoryInfo _recordDirectory; + private readonly ConcurrentDictionary _dirtyKeys = new(StringComparer.Ordinal); + private readonly ConcurrentDictionary _deletedKeys = new(StringComparer.Ordinal); + + /// + /// Creates a paged MessagePack backing store. + /// + 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)); + } + + /// + /// Gets the default record directory path for a manifest path. + /// + public static string DefaultRecordDirectoryPath(string manifestPath) => manifestPath + ".records"; + + /// + public override void PullAll() + { + if (!_manifestFile.Exists && !_recordDirectory.Exists) + { + SetLastSchemaMigrationReports(Array.Empty()); + IsDirty = false; + return; + } + + var manifest = ReadManifest(); + var reports = new List(); + 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; + } + + /// + public override void Push(CultStoredDocument entry) + { + Entries[entry.Key.Value] = entry; + _dirtyKeys[entry.Key.Value] = true; + _deletedKeys.TryRemove(entry.Key.Value, out _); + IsDirty = true; + } + + /// + 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; + } + + /// + 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() + }; + 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(), + Records = Array.Empty() + }; + } + + 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); + } + } + } +} diff --git a/tests/GameCult.Caching.Tests/BackingStoreTests.cs b/tests/GameCult.Caching.Tests/BackingStoreTests.cs index 471d40d3..0e65cc36 100644 --- a/tests/GameCult.Caching.Tests/BackingStoreTests.cs +++ b/tests/GameCult.Caching.Tests/BackingStoreTests.cs @@ -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(first.Key)!.Value, Is.EqualTo(new string('a', 4096))); + Assert.That(read.Get(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() {