Caching: a plain flush and a directory write refuse a store another process stamped newer since attach

Single-file PushAll reads the format stamp under the lock and refuses cultcache.store.v<N> with N > 2;
the directory store throws NotSupportedException for N > 4 from the manifest parse every write already runs.
An unreadable stamp keeps today's behaviour.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-10-22 16:43:11 +02:00
commit d44188637d
4 changed files with 159 additions and 0 deletions

View file

@ -548,6 +548,12 @@ 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);

View file

@ -449,6 +449,14 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
}
var snapshot = CultDocumentMessagePackSerialization.DeserializeSnapshot(manifestBytes);
if (CultPersistedStoreSnapshot.FormatNumber(snapshot.FormatVersion) is int number && number > 4)
{
// Every write path parses the manifest under its commit lease before it writes a page or the manifest, so a
// store another process has since stamped newer is refused here, untouched.
throw new NotSupportedException(
$"Directory store {_manifestFile.FullName} is {snapshot.FormatVersion}, a newer format than this runtime reads ({IndexedFormatVersion}).");
}
if (!string.Equals(snapshot.FormatVersion, IndexedFormatVersion, StringComparison.Ordinal))
{
throw new InvalidOperationException(

View file

@ -66,6 +66,24 @@ namespace GameCult.Caching
public const string FormatV2 = "cultcache.store.v2";
public string FormatVersion { get; set; } = FormatV1;
/// <summary>
/// The N of a header shaped exactly <c>cultcache.store.v&lt;N&gt;</c> or <c>cultcache.store.v&lt;N&gt;.&lt;suffix&gt;</c>;
/// null for any other header. A header whose N exceeds what a runtime reads is a newer format, which that runtime
/// refuses and must never rewrite.
/// </summary>
internal static int? FormatNumber(string? header)
{
const string prefix = "cultcache.store.v";
if (header == null || !header.StartsWith(prefix, StringComparison.Ordinal))
return null;
var end = prefix.Length;
while (end < header.Length && header[end] >= '0' && header[end] <= '9')
end++;
if (end == prefix.Length || end - prefix.Length > 6 || (end < header.Length && header[end] != '.'))
return null;
return int.Parse(header.Substring(prefix.Length, end - prefix.Length), System.Globalization.CultureInfo.InvariantCulture);
}
public CultSchemaCatalogEntry[] SchemaCatalog { get; set; } = Array.Empty<CultSchemaCatalogEntry>();
public CultPersistedRecord[] Records { get; set; } = Array.Empty<CultPersistedRecord>();
}
@ -3354,6 +3372,14 @@ namespace GameCult.Caching
protected FileInfo FileInfo { get; }
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);
@ -3429,6 +3455,7 @@ namespace GameCult.Caching
{
using (AcquireLock(wait: true))
{
RefuseToRewriteANewerFormat();
WriteSnapshot(
Entries.Values.Select(entry => ToPersistedRecord(entry, SerializePayload)),
Entries.Values.Select(entry => entry.Descriptor.ToCatalogEntry()));
@ -3520,6 +3547,30 @@ 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()
{
FileInfo.Refresh();
if (!FileInfo.Exists)
return;
string? header;
try
{
header = ReadFormatVersion(ReadAllBytesShared(FileInfo.FullName));
}
catch (Exception)
{
return;
}
if (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()
{
FileInfo.Refresh();

View file

@ -156,6 +156,100 @@ namespace GameCult.Caching.Tests
}
}
// Two processes: this one attached a v1 store, then a later build rewrote it as v3. A plain flush, a staged write
// and a staged delete must each refuse to downgrade it, leaving the file as the other process wrote it.
[Test]
public async Task SingleFileFlushRefusesAStoreAnotherProcessStampedNewerSinceAttach()
{
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();
var snapshot = CultDocumentMessagePackSerialization.DeserializeSnapshot(File.ReadAllBytes(file));
snapshot.FormatVersion = "cultcache.store.v3";
File.WriteAllBytes(file, CultDocumentMessagePackSerialization.SerializeSnapshot(snapshot));
var before = File.ReadAllBytes(file);
await cache.AddAsync(new PreCut2FixtureNote { Text = "staged after the other process wrote v3" });
Assert.Throws<NotSupportedException>(() => cache.FlushAllBackingStores());
Assert.That(File.ReadAllBytes(file), Is.EqualTo(before));
cache.Remove(new CultRecordKey("item:anvil"));
Assert.Throws<NotSupportedException>(() => cache.BackingStores.Single().PushAll());
Assert.That(File.ReadAllBytes(file), Is.EqualTo(before));
}
finally
{
if (Directory.Exists(root)) Directory.Delete(root, recursive: true);
}
}
// An unreadable stamp is not a newer format: a flush still heals a damaged store.
[Test]
public async Task SingleFileFlushStillHealsAStoreWhoseStampIsUnreadable()
{
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();
File.WriteAllBytes(file, new byte[] { 0xc1, 0xc1 });
await cache.AddAsync(new PreCut2FixtureNote { Text = "heals" });
cache.FlushAllBackingStores();
Assert.That(CultDocumentMessagePackSerialization.DeserializeSnapshot(File.ReadAllBytes(file)).FormatVersion,
Is.EqualTo("cultcache.store.v1"));
}
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)
{
var root = CopyFixture("directory-v4");
try
{
var manifest = Path.Combine(root, "store.cc");
using var cache = new CultCache();
cache.AddBackingStore(new DirectoryMessagePackBackingStore(manifest));
await cache.PullAllBackingStoresAsync();
var snapshot = CultDocumentMessagePackSerialization.DeserializeSnapshot(File.ReadAllBytes(manifest));
snapshot.FormatVersion = newer;
File.WriteAllBytes(manifest, CultDocumentMessagePackSerialization.SerializeSnapshot(snapshot));
Func<System.Collections.Generic.Dictionary<string, byte[]>> files = () => Directory
.GetFiles(root, "*", SearchOption.AllDirectories)
.Where(path => !path.EndsWith(".lock", StringComparison.Ordinal))
.ToDictionary(path => path, File.ReadAllBytes);
var before = files();
await cache.AddAsync(new PreCut2FixtureNote { Text = "staged after the other process wrote the newer format" });
Assert.Throws<NotSupportedException>(() => cache.FlushAllBackingStores());
Assert.That(files().Keys, Is.EquivalentTo(before.Keys));
foreach (var pair in before) Assert.That(File.ReadAllBytes(pair.Key), Is.EqualTo(pair.Value), pair.Key);
cache.Remove(new CultRecordKey("item:anvil"));
Assert.Throws<NotSupportedException>(() => cache.FlushAllBackingStores());
Assert.That(files().Keys, Is.EquivalentTo(before.Keys));
foreach (var pair in before) Assert.That(File.ReadAllBytes(pair.Key), Is.EqualTo(pair.Value), pair.Key);
}
finally
{
if (Directory.Exists(root)) Directory.Delete(root, recursive: true);
}
}
[Test]
public void SingleFileRefusesExtraRecordSlotNamingTheRecord()
{