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

Caching: SingleFileBackingStore.PushAll owns the flush guard; it reads the disk as attach and commit do, ReadFormatVersion is deleted

Browse source 
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-10-22 17:00:10 +02:00
commit 0bf11ddc94
3 changed files with 147 additions and 29 deletions
Download patch file Download diff file Expand all files Collapse all files

6
src/GameCult.Caching.MessagePack/CultDocumentMessagePackSerialization.cs
View file

@ -548,12 +548,6 @@ public class SingleFileMessagePackBackingStore : SingleFileBackingStore
return snapshot;
}
protected override string? ReadFormatVersion(byte[] data)
{
var reader = new MessagePackReader(data);
return reader.ReadArrayHeader() > 0 ? reader.ReadString() : null;
}
protected override byte[] SerializePayload(object document)
{
return CultDocumentMessagePackSerialization.SerializeUntyped(document, document.GetType(), Registry);

42
src/GameCult.Caching/CultCache.cs
View file

@ -3373,13 +3373,6 @@ namespace GameCult.Caching
protected abstract byte[] SerializeSnapshot(CultPersistedStoreSnapshot snapshot);
protected abstract CultPersistedStoreSnapshot DeserializeSnapshot(byte[] data);
/// <summary>
/// Reads only the format header of a stored file, or null when this store cannot tell. A plain flush rewrites the
/// whole file from this cache's view, so it first reads this header and refuses a file stamped with a newer
/// format than the store reads; a store that returns null here is not guarded.
/// </summary>
protected virtual string? ReadFormatVersion(byte[] data) => null;
protected abstract byte[] SerializePayload(object document);
protected abstract object DeserializePayload(Type documentType, byte[] payload);
@ -3455,7 +3448,7 @@ namespace GameCult.Caching
{
using (AcquireLock(wait: true))
{
RefuseToRewriteANewerFormat();
RefuseToRewriteAStoreItCannotRead();
WriteSnapshot(
Entries.Values.Select(entry => ToPersistedRecord(entry, SerializePayload)),
Entries.Values.Select(entry => entry.Descriptor.ToCatalogEntry()));
@ -3547,28 +3540,31 @@ namespace GameCult.Caching
Judging!(arriving, departing);
}
// A plain flush replaces the file without merging it, so under the lock it first reads the stamp: a file another
// process has since written in a newer format is refused and left as it is. An unreadable stamp (missing, empty,
// corrupt) is not newer, and the flush proceeds as it always has, so it can still heal a damaged store.
private void RefuseToRewriteANewerFormat()
// A plain flush replaces the file without merging it, so under the lock it first reads the disk the way attach and
// commit do: a store this runtime does not read (a newer format, or a header the reader refuses) is refused and left
// as it is, and so is a read that fails on I/O. A file that merely will not parse (empty, corrupt) is not a newer
// store, and the flush proceeds as it always has, so it can still heal a damaged store.
private void RefuseToRewriteAStoreItCannotRead()
{
FileInfo.Refresh();
if (!FileInfo.Exists)
return;
string? header;
try
{
header = ReadFormatVersion(ReadAllBytesShared(FileInfo.FullName));
ReadSnapshot();
}
catch (NotSupportedException)
{
throw;
}
catch (IOException)
{
throw;
}
catch (UnauthorizedAccessException)
{
throw;
}
catch (Exception)
{
return;
}
if (CultPersistedStoreSnapshot.FormatNumber(header) is int number && number > 2)
throw new NotSupportedException(
$"Store {FileInfo.FullName} is {header}, a newer format than this runtime reads " +
$"({CultPersistedStoreSnapshot.FormatV1} and {CultPersistedStoreSnapshot.FormatV2}); a flush would downgrade it, so none was written.");
}
private CultPersistedStoreSnapshot? ReadSnapshot()

128
tests/GameCult.Caching.Tests/PreCut2StoreFormatTests.cs
View file

@ -213,6 +213,134 @@ namespace GameCult.Caching.Tests
}
}
// A store subclass that overrides nothing beyond its four abstract members is guarded by the base, not by an opt-in.
private sealed class PlainCustomStore : SingleFileBackingStore
{
public PlainCustomStore(string path) : base(path) { }
protected override byte[] SerializeSnapshot(CultPersistedStoreSnapshot snapshot) =>
CultDocumentMessagePackSerialization.SerializeSnapshot(snapshot);
protected override CultPersistedStoreSnapshot DeserializeSnapshot(byte[] data)
{
var snapshot = CultDocumentMessagePackSerialization.DeserializeSnapshot(data);
CultDocumentMessagePackSerialization.RequireSingleFileFormat(snapshot);
return snapshot;
}
protected override byte[] SerializePayload(object document) =>
CultDocumentMessagePackSerialization.SerializeUntyped(document, document.GetType(), Registry);
protected override object DeserializePayload(Type documentType, byte[] payload) =>
CultDocumentMessagePackSerialization.DeserializeUntyped(documentType, payload, Registry);
}
private sealed class FlakyReadStore : SingleFileMessagePackBackingStore
{
public Exception? FailReadWith;
public FlakyReadStore(string path) : base(path) { }
protected override CultPersistedStoreSnapshot DeserializeSnapshot(byte[] data) =>
FailReadWith != null ? throw FailReadWith : base.DeserializeSnapshot(data);
}
private static void StampHeader(string file, string header)
{
var snapshot = CultDocumentMessagePackSerialization.DeserializeSnapshot(File.ReadAllBytes(file));
snapshot.FormatVersion = header;
File.WriteAllBytes(file, CultDocumentMessagePackSerialization.SerializeSnapshot(snapshot));
}
[Test]
public async Task CustomSingleFileStoreThatOverridesNothingIsRefusedAtFlushOfANewerStore()
{
var root = CopyFixture("single-file-v1");
try
{
var file = Path.Combine(root, "store.msgpack");
using var cache = new CultCache();
cache.AddBackingStore(new PlainCustomStore(file));
await cache.PullAllBackingStoresAsync();
StampHeader(file, "cultcache.store.v3");
var before = File.ReadAllBytes(file);
await cache.AddAsync(new PreCut2FixtureNote { Text = "staged" });
Assert.Throws<NotSupportedException>(() => cache.FlushAllBackingStores());
Assert.That(File.ReadAllBytes(file), Is.EqualTo(before));
}
finally
{
if (Directory.Exists(root)) Directory.Delete(root, recursive: true);
}
}
// Attach, commit and flush share one rule for which formats are readable: a header the reader refuses at attach is
// refused at flush, whatever its spelling.
[TestCase("cultcache.store.v3-ids")]
[TestCase("cultcache.store.v1000000")]
[TestCase("cultcache.v3")]
[TestCase("cultcache.store.v3")]
[TestCase("cultcache.store.v2")]
[TestCase("cultcache.store.v1")]
public async Task FlushRefusesExactlyTheHeadersAttachRefuses(string header)
{
var root = CopyFixture("single-file-v1");
try
{
var file = Path.Combine(root, "store.msgpack");
using var cache = new CultCache();
cache.AddBackingStore(new SingleFileMessagePackBackingStore(file));
await cache.PullAllBackingStoresAsync();
StampHeader(file, header);
var before = File.ReadAllBytes(file);
bool attachRefuses;
using (var fresh = new CultCache())
{
try { fresh.AddBackingStore(new SingleFileMessagePackBackingStore(file)); attachRefuses = false; }
catch (NotSupportedException) { attachRefuses = true; }
}
await cache.AddAsync(new PreCut2FixtureNote { Text = "staged" });
var flushRefuses = false;
try { cache.FlushAllBackingStores(); }
catch (NotSupportedException) { flushRefuses = true; }
Assert.That(flushRefuses, Is.EqualTo(attachRefuses), header);
if (attachRefuses) Assert.That(File.ReadAllBytes(file), Is.EqualTo(before), header);
}
finally
{
if (Directory.Exists(root)) Directory.Delete(root, recursive: true);
}
}
// A read that fails on I/O is not an unreadable stamp to heal over: the flush refuses and the file is untouched.
[Test]
public async Task SingleFileFlushRefusesWhenReadingTheDiskFailsOnIO()
{
var root = CopyFixture("single-file-v1");
try
{
var file = Path.Combine(root, "store.msgpack");
var store = new FlakyReadStore(file);
using var cache = new CultCache();
cache.AddBackingStore(store);
await cache.PullAllBackingStoresAsync();
var before = File.ReadAllBytes(file);
await cache.AddAsync(new PreCut2FixtureNote { Text = "staged" });
store.FailReadWith = new IOException("transient");
Assert.Throws<IOException>(() => cache.FlushAllBackingStores());
Assert.That(File.ReadAllBytes(file), Is.EqualTo(before));
}
finally
{
if (Directory.Exists(root)) Directory.Delete(root, recursive: true);
}
}
[TestCase("cultcache.store.v5.directory-content-addressed-pages")]
[TestCase("cultcache.store.v9")]
public async Task DirectoryWritesRefuseAStoreAnotherProcessStampedNewerSinceAttach(string newer)