C#: a write through a live link at the store path is refused; read-under-lock pinned (store-link-cs) #151

Merged
libby merged 5 commits from store-replace/link-cs into main 2026-11-13 00:36:21 +00:00
8 changed files with 674 additions and 32 deletions

View file

@ -710,17 +710,23 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
// Only a writer creates the records folder and the lock, so opening the store creates nothing. A reader opens an
// existing lock and returns null when there is none; that does not mean no writer committed (a store may be copied
// without its lock, and a first writer may create it mid-load), so an unleased load checks that it settled.
// A writer's lease is also where a link at the manifest path is refused (a commit would replace the link with a file): before
// the records folder or the lock is made, and again once the lease is held, because a link can be put there while it waited.
private FileStream? AcquireCommitLease(bool wait, bool create)
{
if (create)
{
CultStoreLinks.Refuse(_manifestFile.FullName);
Directory.CreateDirectory(_recordDirectory.FullName);
}
var lockPath = Path.Combine(_recordDirectory.FullName, ".commit.lock");
var started = Stopwatch.StartNew();
while (true)
{
FileStream held;
try
{
return new FileStream(
held = new FileStream(
lockPath,
create ? FileMode.OpenOrCreate : FileMode.Open,
FileAccess.ReadWrite,
@ -743,7 +749,23 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
catch (IOException) when (started.Elapsed < TimeSpan.FromSeconds(30))
{
Thread.Sleep(10);
continue;
}
if (create)
{
try
{
CultStoreLinks.Refuse(_manifestFile.FullName);
}
catch
{
held.Dispose();
throw;
}
}
return held;
}
}

View file

@ -5132,7 +5132,7 @@ namespace GameCult.Caching
var attributes = FileInfo.Attributes;
if ((int)attributes == -1)
return null;
if ((attributes & FileAttributes.Directory) != 0 && (attributes & FileAttributes.ReparsePoint) == 0)
if ((attributes & FileAttributes.Directory) != 0 && Directory.Exists(FileInfo.FullName))
throw new IOException("The store path is a directory, not a store file.");
try
{
@ -5154,25 +5154,27 @@ namespace GameCult.Caching
}
// Something is at the path even when the path cannot be opened: a link keeps the reparse-point attribute whether or not its
// target is there. A path with nothing at it reports no attributes at all (-1).
private bool PathIsALink()
{
FileInfo.Refresh();
var attributes = FileInfo.Attributes;
return (int)attributes != -1 && (attributes & FileAttributes.ReparsePoint) != 0;
}
// target is there.
private bool PathIsALink() => CultStoreLinks.IsLink(FileInfo.FullName);
// A write never goes through a link at the store path: replacing it would silently detach the store from its link.
private void RefuseLinkedStore() => CultStoreLinks.Refuse(FileInfo.FullName);
// The lock is a sidecar opened exclusively, which excludes other handles in this process and in others alike.
// Every write takes the lock first, so the link is refused here twice: before the lock file or anything else is made, and
// again once the lock is held, because a link can be put at the path while this writer waited on another's lock.
private FileStream? AcquireLock(bool wait)
{
RefuseLinkedStore();
Directory.CreateDirectory(FileInfo.DirectoryName!);
var lockPath = FileInfo.FullName + ".lock";
var started = Stopwatch.StartNew();
while (true)
{
FileStream held;
try
{
return new FileStream(lockPath, FileMode.OpenOrCreate, FileAccess.ReadWrite, FileShare.None, bufferSize: 1);
held = new FileStream(lockPath, FileMode.OpenOrCreate, FileAccess.ReadWrite, FileShare.None, bufferSize: 1);
}
catch (IOException) when (!wait)
{
@ -5181,7 +5183,20 @@ namespace GameCult.Caching
catch (IOException) when (started.Elapsed < TimeSpan.FromSeconds(30))
{
Thread.Sleep(10);
continue;
}
try
{
RefuseLinkedStore();
}
catch
{
held.Dispose();
throw;
}
return held;
}
}

View file

@ -0,0 +1,24 @@
using System.IO;
namespace GameCult.Caching
{
/// <summary>
/// A write was attempted at a store path that is a link (a symbolic link to a file or a directory, a Windows junction or any
/// other reparse-point link). The write is refused before the lock, the staging file or the replace, and again once the lock
/// is held, so a link that exists when a writer takes the lock leaves the link and its target untouched. A link created by
/// something that ignores the lock after that check, while the commit builds and writes the new image, is still replaced;
/// that window cannot be closed without native no-follow opens. Configure the link's target as the store path to write to
/// it; reads through a live link are unchanged.
/// </summary>
public sealed class CultStoreLinkedException : IOException
{
public CultStoreLinkedException(string path)
: base($"CultCache store {path} is a link, and a write at it would replace the link; configure the link's target as the store path.")
{
Path = path;
}
/// <summary>The store path that is a link.</summary>
public string Path { get; }
}
}

View file

@ -0,0 +1,22 @@
using System.IO;
namespace GameCult.Caching
{
// The one decision on whether a store-owned file is a link, shared by every store kind that replaces files.
internal static class CultStoreLinks
{
// A path with nothing at it reports no attributes at all (-1). Any reparse point counts as a link, whether or not its
// target is there.
internal static bool IsLink(string path)
{
var attributes = new FileInfo(path).Attributes;
return (int)attributes != -1 && (attributes & FileAttributes.ReparsePoint) != 0;
}
internal static void Refuse(string path)
{
if (IsLink(path))
throw new CultStoreLinkedException(path);
}
}
}

View file

@ -0,0 +1,330 @@
#nullable enable
using System;
using System.IO;
using System.Linq;
using System.Threading.Tasks;
using GameCult.Caching.MessagePack;
using NUnit.Framework;
using static GameCult.Caching.Tests.ElementIdTests;
using static GameCult.Caching.Tests.StoreReadabilityTests;
namespace GameCult.Caching.Tests
{
// A write never replaces a link at a store-owned path. The refusal holds for every way a write reaches a store (waiting or
// not, as a judge of another store's commit), for a link put there while the write waited on the store's lock, and for the
// directory store's manifest.
public class StoreLinkRefusalTests
{
private static readonly CultDocumentRegistry Registry = CultDocumentRegistry.ForTypes(new[] { typeof(IdDeck), typeof(VectorItem) });
private string _directory = "";
[SetUp]
public void SetUp()
{
_directory = Path.Combine(Path.GetTempPath(), "cultlib-link-refusal-" + Guid.NewGuid().ToString("N"));
Directory.CreateDirectory(_directory);
}
[TearDown]
public void TearDown()
{
if (!Directory.Exists(_directory))
return;
foreach (var entry in Directory.EnumerateFileSystemEntries(_directory).Where(IsALink))
File.Delete(entry);
Directory.Delete(_directory, true);
}
private string PathOf(string name) => Path.Combine(_directory, name);
private static bool IsALink(string path)
{
var attributes = new FileInfo(path).Attributes;
return (int)attributes != -1 && (attributes & FileAttributes.ReparsePoint) != 0;
}
private static void Link(string path, string target)
{
try
{
File.CreateSymbolicLink(path, target);
}
catch (Exception ex) when (ex is IOException or UnauthorizedAccessException)
{
Assert.Ignore("A file symlink needs a privilege this account lacks: " + ex.Message);
}
}
private static void SeedSingle(string path)
{
using var seed = CultCacheMessagePack.Create(path, new CultCacheOpenOptions { Registry = Registry });
seed.Commit(batch => batch.Upsert(typeof(IdDeck), new IdDeck { Name = "seed" }, new CultRecordKey("seed")));
}
private static void SeedItem(string path)
{
using var seed = new CultCache(Registry);
seed.AddBackingStore(new SingleFileMessagePackBackingStore(path), typeof(VectorItem));
seed.Commit(batch => batch.Upsert(typeof(VectorItem), new VectorItem(), new CultRecordKey("item")));
}
// Runs `write` while another holder has `lockPath`; `swap` runs while it waits, then the lock is released.
private static Exception? WriteWhileWaiting(string lockPath, Action write, Action swap)
{
using var held = new FileStream(lockPath, FileMode.OpenOrCreate, FileAccess.ReadWrite, FileShare.None);
var task = Task.Run(write);
Assert.That(task.Wait(TimeSpan.FromMilliseconds(600)), Is.False, "the lock is held, so the write must wait");
swap();
held.Dispose();
try
{
Assert.That(task.Wait(TimeSpan.FromSeconds(10)), Is.True);
return null;
}
catch (AggregateException ex)
{
return ex.GetBaseException();
}
}
private static void ReplaceWithLink(string path, string target)
{
File.Delete(path);
Link(path, target);
}
// ---- a link put at the store path while the write waited on the lock ----
[Test]
public void ACommitRefusesALinkPutAtTheStoreWhileItWaitedOnTheLock()
{
var path = PathOf("store.cc");
var target = PathOf("shared.cc");
SeedSingle(path);
SeedSingle(target);
var before = File.ReadAllBytes(target);
using var cache = CultCacheMessagePack.Create(path, new CultCacheOpenOptions { Registry = Registry, FlushOnDispose = false, StoreFlushOnDispose = false });
var error = WriteWhileWaiting(path + ".lock",
() => cache.Commit(batch => batch.Upsert(typeof(IdDeck), new IdDeck { Name = "late" }, new CultRecordKey("late"))),
() => ReplaceWithLink(path, target));
Assert.That(error, Is.InstanceOf<CultStoreLinkedException>(), error?.ToString());
Assert.That(((CultStoreLinkedException)error!).Path, Is.EqualTo(path));
Assert.That(IsALink(path), "the link was replaced");
Assert.That(File.ReadAllBytes(target), Is.EqualTo(before));
}
[Test]
public void AFlushRefusesALinkPutAtTheStoreWhileItWaitedOnTheLock()
{
var path = PathOf("store.cc");
var target = PathOf("shared.cc");
SeedSingle(path);
SeedSingle(target);
var before = File.ReadAllBytes(target);
using var cache = CultCacheMessagePack.Create(path, new CultCacheOpenOptions { Registry = Registry, FlushOnDispose = false, StoreFlushOnDispose = false });
cache.UpsertAsync(typeof(IdDeck), new IdDeck { Name = "late" }, new CultRecordKey("late")).GetAwaiter().GetResult();
var error = WriteWhileWaiting(path + ".lock", () => cache.FlushAllBackingStores(), () => ReplaceWithLink(path, target));
Assert.That(error, Is.InstanceOf<CultStoreLinkedException>(), error?.ToString());
Assert.That(IsALink(path), "the link was replaced");
Assert.That(File.ReadAllBytes(target), Is.EqualTo(before));
}
// ---- the non-waiting and judging write paths ----
[Test]
public void ATryCommitAtALiveLinkIsRefusedNotContendedOrReplaced()
{
var path = PathOf("store.cc");
var link = PathOf("link.cc");
SeedSingle(path);
Link(link, path);
var before = File.ReadAllBytes(path);
using var cache = CultCacheMessagePack.Create(link, new CultCacheOpenOptions { Registry = Registry, FlushOnDispose = false, StoreFlushOnDispose = false });
var refusal = Assert.Throws<CultStoreLinkedException>(() =>
cache.TryCommit(batch => batch.Upsert(typeof(IdDeck), new IdDeck { Name = "t" }, new CultRecordKey("t"))))!;
Assert.That(refusal.Path, Is.EqualTo(link));
Assert.That(refusal.Message, Does.Contain(link));
Assert.That(IsALink(link), "the link was replaced");
Assert.That(File.ReadAllBytes(path), Is.EqualTo(before));
Assert.That(File.Exists(link + ".lock"), Is.False, "the refused write took the lock");
}
[Test]
public void ATryCommitAtALiveLinkWhoseLockIsHeldIsStillRefused()
{
var path = PathOf("store.cc");
var link = PathOf("link.cc");
SeedSingle(path);
Link(link, path);
using var cache = CultCacheMessagePack.Create(link, new CultCacheOpenOptions { Registry = Registry, FlushOnDispose = false, StoreFlushOnDispose = false });
using var held = new FileStream(link + ".lock", FileMode.OpenOrCreate, FileAccess.ReadWrite, FileShare.None);
Assert.Throws<CultStoreLinkedException>(() =>
cache.TryCommit(batch => batch.Upsert(typeof(IdDeck), new IdDeck { Name = "t" }, new CultRecordKey("t"))));
Assert.That(IsALink(link), "the link was replaced");
}
// A batch that upserts into one store and removes a key the cache does not hold has the other store judge that removal
// under its own lock (JudgeChecks).
private CultCache OpenTwoStores(string homePath, string otherPath)
{
var cache = new CultCache(Registry);
cache.AddBackingStore(new SingleFileMessagePackBackingStore(homePath), typeof(IdDeck));
cache.AddBackingStore(new SingleFileMessagePackBackingStore(otherPath), typeof(VectorItem));
return cache;
}
[Test]
public void AnotherStoreAtALiveLinkRefusesTheChecksItIsAskedToJudge()
{
var other = PathOf("other.cc");
var link = PathOf("other-link.cc");
SeedItem(other);
Link(link, other);
var before = File.ReadAllBytes(other);
using var cache = OpenTwoStores(PathOf("home.cc"), link);
var refusal = Assert.Throws<CultStoreLinkedException>(() => cache.Commit(batch =>
{
batch.Upsert(typeof(IdDeck), new IdDeck { Name = "home" }, new CultRecordKey("home"));
batch.Remove(new CultRecordKey("ghost"));
}))!;
Assert.That(refusal.Path, Is.EqualTo(link));
Assert.That(IsALink(link), "the link was replaced");
Assert.That(File.ReadAllBytes(other), Is.EqualTo(before));
Assert.That(File.Exists(PathOf("home.cc")), Is.False, "the home store was written though a judge refused");
}
[Test]
public void AJudgeReadsTheFileItHoldsUnderTheLockNotTheFileBeforeItWaited()
{
var other = PathOf("other.cc");
SeedItem(other);
using var cache = OpenTwoStores(PathOf("home.cc"), other);
var error = WriteWhileWaiting(other + ".lock",
() => cache.Commit(batch =>
{
batch.Upsert(typeof(IdDeck), new IdDeck { Name = "home" }, new CultRecordKey("home"));
batch.Remove(new CultRecordKey("ghost"));
}),
() => File.WriteAllBytes(other, Array.Empty<byte>()));
Assert.That(error, Is.InstanceOf<CultStoreUnreadableException>(), error?.ToString());
}
// ---- the directory store's manifest ----
private DirectoryMessagePackBackingStore SeedDirectory(string manifest)
{
using (var seed = new CultCache(Registry))
{
seed.AddBackingStore(new DirectoryMessagePackBackingStore(manifest), typeof(IdDeck));
seed.Commit(batch => batch.Upsert(typeof(IdDeck), new IdDeck { Name = "seed" }, new CultRecordKey("seed")));
}
return new DirectoryMessagePackBackingStore(manifest);
}
private CultCache OpenDirectory(DirectoryMessagePackBackingStore store)
{
var cache = new CultCache(Registry);
cache.AddBackingStore(store, typeof(IdDeck));
return cache;
}
[Test]
public void ADirectoryStoreCommitAtALiveManifestLinkIsRefusedAndNothingChanges()
{
var manifest = PathOf("real.cc");
var link = PathOf("link.cc");
SeedDirectory(manifest);
Link(link, manifest);
Directory.Move(manifest + ".records", link + ".records");
var before = File.ReadAllBytes(manifest);
var pages = Directory.GetFileSystemEntries(link + ".records").OrderBy(entry => entry).ToArray();
using var cache = OpenDirectory(new DirectoryMessagePackBackingStore(link));
var refusal = Assert.Throws<CultStoreLinkedException>(() =>
cache.Commit(batch => batch.Upsert(typeof(IdDeck), new IdDeck { Name = "b" }, new CultRecordKey("b"))))!;
Assert.That(refusal.Path, Is.EqualTo(link));
Assert.That(IsALink(link), "the link was replaced");
Assert.That(File.ReadAllBytes(manifest), Is.EqualTo(before));
Assert.That(Directory.GetFileSystemEntries(link + ".records").OrderBy(entry => entry), Is.EqualTo(pages), "the refused write left files beside the manifest");
}
// The refusal comes before the records folder or the lease file is made: a store that was never written leaves nothing behind.
[Test]
public void ADirectoryStoreCommitAtAManifestLinkPutThereAfterOpenCreatesNothing()
{
var link = PathOf("link.cc");
using var cache = OpenDirectory(new DirectoryMessagePackBackingStore(link));
Link(link, PathOf("unmounted.cc"));
Assert.Throws<CultStoreLinkedException>(() =>
cache.Commit(batch => batch.Upsert(typeof(IdDeck), new IdDeck { Name = "b" }, new CultRecordKey("b"))));
Assert.That(IsALink(link), "the link was replaced");
Assert.That(Directory.Exists(link + ".records"), Is.False, "the refused write made the records folder or its lease");
}
[Test]
public void ADirectoryStoreFlushAtALiveManifestLinkIsRefused()
{
var manifest = PathOf("real.cc");
var link = PathOf("link.cc");
SeedDirectory(manifest);
Link(link, manifest);
Directory.Move(manifest + ".records", link + ".records");
var before = File.ReadAllBytes(manifest);
using var cache = OpenDirectory(new DirectoryMessagePackBackingStore(link));
cache.UpsertAsync(typeof(IdDeck), new IdDeck { Name = "f" }, new CultRecordKey("f")).GetAwaiter().GetResult();
Assert.Throws<CultStoreLinkedException>(() => cache.FlushAllBackingStores());
Assert.That(IsALink(link), "the link was replaced");
Assert.That(File.ReadAllBytes(manifest), Is.EqualTo(before));
}
[Test]
public void ADirectoryStoreCommitRefusesAManifestLinkPutThereWhileItWaitedOnTheLease()
{
var manifest = PathOf("store.cc");
var target = PathOf("shared.cc");
SeedDirectory(manifest);
SeedDirectory(target);
var before = File.ReadAllBytes(target);
using var cache = OpenDirectory(new DirectoryMessagePackBackingStore(manifest));
var error = WriteWhileWaiting(Path.Combine(manifest + ".records", ".commit.lock"),
() => cache.Commit(batch => batch.Upsert(typeof(IdDeck), new IdDeck { Name = "late" }, new CultRecordKey("late"))),
() => ReplaceWithLink(manifest, target));
Assert.That(error, Is.InstanceOf<CultStoreLinkedException>(), error?.ToString());
Assert.That(IsALink(manifest), "the link was replaced");
Assert.That(File.ReadAllBytes(target), Is.EqualTo(before));
}
[Test]
public void ADirectoryStoreReadsThroughALiveManifestLink()
{
var manifest = PathOf("real.cc");
var link = PathOf("link.cc");
SeedDirectory(manifest);
Link(link, manifest);
Directory.Move(manifest + ".records", link + ".records");
using var cache = OpenDirectory(new DirectoryMessagePackBackingStore(link));
Assert.That(cache.Get<IdDeck>(new CultRecordKey("seed"))?.Name, Is.EqualTo("seed"));
}
}
}

View file

@ -266,7 +266,7 @@ namespace GameCult.Caching.Tests
using var cache = CultCacheMessagePack.Create(path, new CultCacheOpenOptions { Registry = Registry });
CreateDanglingDirectoryLink(path, Path.Combine(_directory, "unmounted"));
Assert.That(() => cache.Commit(batch => batch.Upsert(typeof(IdDeck), new IdDeck { Name = "e" }, new CultRecordKey("e"))), Throws.TypeOf<IOException>());
Assert.That(() => cache.Commit(batch => batch.Upsert(typeof(IdDeck), new IdDeck { Name = "e" }, new CultRecordKey("e"))), Throws.TypeOf<CultStoreLinkedException>());
Assert.That(IsALink(path), "the link was replaced");
Assert.That(Directory.Exists(Path.Combine(_directory, "unmounted")), Is.False, "a write went through the link");
}
@ -292,11 +292,66 @@ namespace GameCult.Caching.Tests
using var cache = CultCacheMessagePack.Create(path, new CultCacheOpenOptions { Registry = Registry });
File.Delete(target);
Assert.That(() => cache.Commit(batch => batch.Upsert(typeof(IdDeck), new IdDeck { Name = "e" }, new CultRecordKey("e"))), Throws.TypeOf<IOException>());
Assert.That(() => cache.Commit(batch => batch.Upsert(typeof(IdDeck), new IdDeck { Name = "e" }, new CultRecordKey("e"))), Throws.TypeOf<CultStoreLinkedException>());
Assert.That(new FileInfo(path).LinkTarget, Is.Not.Null, "the link was replaced");
Assert.That(File.Exists(target), Is.False, "a write went through the link");
}
// A write at a store path that is a link is refused with a typed error before the lock or the replace: the link, its target
// and the directory beside them are as they were.
[Test]
public void ACacheCommitAtALiveLinkIsRefusedAndNothingChanges()
{
var target = Path.Combine(_directory, "target.cc");
var link = Path.Combine(_directory, "link.cc");
using (var seed = CultCacheMessagePack.Create(target, new CultCacheOpenOptions { Registry = Registry }))
seed.Commit(batch => batch.Upsert(typeof(IdDeck), new IdDeck { Name = "a" }, new CultRecordKey("a")));
CreateFileLinkOrIgnore(link, target);
var before = File.ReadAllBytes(target);
using var cache = CultCacheMessagePack.Create(link, new CultCacheOpenOptions { Registry = Registry, FlushOnDispose = false, StoreFlushOnDispose = false });
var refusal = Assert.Throws<CultStoreLinkedException>(() => cache.Commit(batch => batch.Upsert(typeof(IdDeck), new IdDeck { Name = "b" }, new CultRecordKey("b"))))!;
Assert.That(refusal.Message, Does.Contain(link));
Assert.That(IsALink(link), "the link was replaced");
Assert.That(File.ReadAllBytes(target), Is.EqualTo(before));
Assert.That(File.Exists(link + ".lock"), Is.False, "the refused write took the lock");
Assert.That(Directory.GetFileSystemEntries(_directory).Where(entry => entry.EndsWith(".tmp")), Is.Empty);
}
[Test]
public void AFlushAtALiveLinkIsRefusedAndNothingChanges()
{
var target = Path.Combine(_directory, "target.cc");
var link = Path.Combine(_directory, "link.cc");
using (var seed = CultCacheMessagePack.Create(target, new CultCacheOpenOptions { Registry = Registry }))
seed.Commit(batch => batch.Upsert(typeof(IdDeck), new IdDeck { Name = "a" }, new CultRecordKey("a")));
CreateFileLinkOrIgnore(link, target);
var before = File.ReadAllBytes(target);
using var cache = CultCacheMessagePack.Create(link, new CultCacheOpenOptions { Registry = Registry, FlushOnDispose = false, StoreFlushOnDispose = false });
cache.UpsertAsync(typeof(IdDeck), new IdDeck { Name = "b" }, new CultRecordKey("b")).GetAwaiter().GetResult();
Assert.Throws<CultStoreLinkedException>(() => cache.FlushAllBackingStores());
Assert.That(IsALink(link), "the link was replaced");
Assert.That(File.ReadAllBytes(target), Is.EqualTo(before));
Assert.That(File.Exists(link + ".lock"), Is.False);
}
[Test]
public void AnOpenAtALiveLinkReadsTheTargetsRecords()
{
var target = Path.Combine(_directory, "target.cc");
var link = Path.Combine(_directory, "link.cc");
using (var seed = CultCacheMessagePack.Create(target, new CultCacheOpenOptions { Registry = Registry }))
seed.Commit(batch => batch.Upsert(typeof(IdDeck), new IdDeck { Name = "a" }, new CultRecordKey("a")));
CreateFileLinkOrIgnore(link, target);
using var cache = CultCacheMessagePack.Create(link, new CultCacheOpenOptions { Registry = Registry, FlushOnDispose = false, StoreFlushOnDispose = false });
Assert.That(cache.Get<IdDeck>(new CultRecordKey("a"))?.Name, Is.EqualTo("a"));
}
// A directory at the path is something at the path: it is an I/O error on open and on commit, never an empty store, and
// it is left as it was.
[Test]

View file

@ -0,0 +1,132 @@
#nullable enable
using System;
using System.IO;
using System.Threading.Tasks;
using GameCult.Caching.MessagePack;
using NUnit.Framework;
using static GameCult.Caching.Tests.ElementIdTests;
namespace GameCult.Caching.Tests
{
// A write decides from the file it holds the lock over, not from the file as it was before it waited: whatever another
// writer left while this one waited for <path>.lock is what the write applies onto, or refuses.
public class StoreWriteReadsUnderTheLockTests
{
private static readonly CultDocumentRegistry Registry = CultDocumentRegistry.ForTypes(new[] { typeof(IdDeck) });
private static readonly CultSchemaCatalogEntry One = new() { SchemaId = "tests.under-lock.one", SchemaName = "tests.under-lock.one" };
private string _directory = "";
private string _path = "";
[SetUp]
public void SetUp()
{
_directory = Path.Combine(Path.GetTempPath(), "cultlib-under-lock-" + Guid.NewGuid().ToString("N"));
Directory.CreateDirectory(_directory);
_path = Path.Combine(_directory, "store.cc");
}
[TearDown]
public void TearDown()
{
if (Directory.Exists(_directory))
Directory.Delete(_directory, true);
}
private void Seed()
{
using var cache = CultCacheMessagePack.Create(_path, new CultCacheOpenOptions { Registry = Registry });
cache.Commit(batch => batch.Upsert(typeof(IdDeck), new IdDeck { Name = "seed" }, new CultRecordKey("seed")));
}
// The write starts and waits on the held lock; while it waits, the file is replaced by what `swap` makes of it.
private Exception? WriteWhileTheFileChangesUnderIt(Action write, Action swap)
{
using var held = new FileStream(_path + ".lock", FileMode.OpenOrCreate, FileAccess.ReadWrite, FileShare.None);
var task = Task.Run(write);
Assert.That(task.Wait(TimeSpan.FromMilliseconds(600)), Is.False, "the lock is held, so the write must wait");
swap();
held.Dispose();
try
{
Assert.That(task.Wait(TimeSpan.FromSeconds(10)), Is.True);
return null;
}
catch (AggregateException ex)
{
return ex.GetBaseException();
}
}
private static void ZeroBytes(string path) => File.WriteAllBytes(path, Array.Empty<byte>());
private static CultPersistedStoreSnapshot Snapshot(string path) =>
CultDocumentMessagePackSerialization.DeserializeSnapshot(File.ReadAllBytes(path));
private void MarkV3()
{
var snapshot = Snapshot(_path);
snapshot.FormatVersion = CultPersistedStoreSnapshot.FormatV3;
File.WriteAllBytes(_path, CultDocumentMessagePackSerialization.SerializeSnapshot(snapshot));
}
private void CommitWrite(Action swap, out Exception? error)
{
using var cache = CultCacheMessagePack.Create(_path, new CultCacheOpenOptions { Registry = Registry, FlushOnDispose = false, StoreFlushOnDispose = false });
error = WriteWhileTheFileChangesUnderIt(
() => cache.Commit(batch => batch.Upsert(typeof(IdDeck), new IdDeck { Name = "committed" }, new CultRecordKey("committed"))), swap);
}
private void FlushWriteWhile(Action swap, out Exception? error)
{
using var cache = CultCacheMessagePack.Create(_path, new CultCacheOpenOptions { Registry = Registry, FlushOnDispose = false, StoreFlushOnDispose = false });
cache.UpsertAsync(typeof(IdDeck), new IdDeck { Name = "flushed" }, new CultRecordKey("flushed")).GetAwaiter().GetResult();
error = WriteWhileTheFileChangesUnderIt(() => cache.FlushAllBackingStores(), swap);
}
private void CachelessWriteWhile(Action swap, out Exception? error) =>
error = WriteWhileTheFileChangesUnderIt(
() => new SingleFileMessagePackBackingStore(_path).WriteRecord(
new CultPersistedRecord { Key = "written", SchemaId = One.SchemaId, StoredAt = "2026-10-22T00:00:00.0000000+00:00", Payload = new byte[] { 1 } }, One),
swap);
private delegate void WriteWhile(Action swap, out Exception? error);
[Test]
public void AFlushDecidesFromTheFileHeldUnderTheLock() => RefusesWhatItFindsUnderTheLock(FlushWriteWhile);
[Test]
public void ACommitDecidesFromTheFileHeldUnderTheLock() => RefusesWhatItFindsUnderTheLock(CommitWrite);
[Test]
public void ACachelessWriteDecidesFromTheFileHeldUnderTheLock() => RefusesWhatItFindsUnderTheLock(CachelessWriteWhile);
[Test]
public void AFlushKeepsTheIdHeaderOfTheFileHeldUnderTheLock() => KeepsTheV3HeaderItFindsUnderTheLock(FlushWriteWhile);
[Test]
public void ACommitKeepsTheIdHeaderOfTheFileHeldUnderTheLock() => KeepsTheV3HeaderItFindsUnderTheLock(CommitWrite);
[Test]
public void ACachelessWriteKeepsTheIdHeaderOfTheFileHeldUnderTheLock() => KeepsTheV3HeaderItFindsUnderTheLock(CachelessWriteWhile);
private void RefusesWhatItFindsUnderTheLock(WriteWhile write)
{
Seed();
write(() => ZeroBytes(_path), out var error);
Assert.That(error, Is.InstanceOf<CultStoreUnreadableException>());
Assert.That(new FileInfo(_path).Length, Is.Zero, "the refused file was written over");
}
private void KeepsTheV3HeaderItFindsUnderTheLock(WriteWhile write)
{
Seed();
write(MarkV3, out var error);
Assert.That(error, Is.Null, error?.ToString());
Assert.That(Snapshot(_path).FormatVersion, Is.EqualTo(CultPersistedStoreSnapshot.FormatV3), "the write read the file as it was before the lock");
}
}
}

View file

@ -434,32 +434,74 @@ public sealed class CultMeshSingleFileStoreTests
TempFiles().Should().BeEmpty();
}
// The store's replace decides what a write leaves where a live link was, and a Mesh write leaves what a cache commit leaves:
// one replace, whatever the platform does with a link.
[Test]
public void AMeshWriteOverALiveLinkLeavesWhatACacheCommitLeaves()
// A write at a store path that is a link is refused before the lock, the staging file and the replace: the link and its
// target stay as they were, and nothing is left beside them.
private void AssertRefusedAndUntouched(string link, string target, byte[] targetBefore, Action write)
{
string Outcome(string name, Action<string> write)
var refusal = Assert.Throws<CultStoreLinkedException>(() => write())!;
refusal.Message.Should().Contain(link);
IsALink(link).Should().BeTrue("the link was replaced");
File.ReadAllBytes(target).Should().Equal(targetBefore);
File.Exists(link + ".lock").Should().BeFalse("the refused write took the lock");
TempFiles().Should().BeEmpty();
}
[Test]
public void AMeshWriteAtALiveLinkIsRefusedAndNothingChanges()
{
var target = Path.Combine(_root, "target.cc");
var link = Path.Combine(_root, "link.cc");
CultMesh.WriteSingleFileDocument(target, Key, Publication("first"));
CreateLiveLinkOrIgnore(link, target);
var before = File.ReadAllBytes(target);
AssertRefusedAndUntouched(link, target, before, () => CultMesh.WriteSingleFileDocument(link, Key, Publication("second")));
AssertRefusedAndUntouched(link, target, before, () => CultMesh.WriteSingleFileDocumentPayload(link, Key, Raw, null, EmptyArray));
}
[Test]
public void AReadThroughALiveLinkStillWorks()
{
var target = Path.Combine(_root, "target.cc");
var link = Path.Combine(_root, "link.cc");
CultMesh.WriteSingleFileDocument(target, Key, Publication("first"));
CreateLiveLinkOrIgnore(link, target);
CultMesh.ReadSingleFileDocument<CultMeshBodyPublicationDocument>(link, Key).ProducerId.Should().Be("first");
RecordsSeenByACache(link).Should().Be(1);
}
// A link to a directory that is there is a directory, not an unreachable store; a write at it is a link refusal.
[Test]
public void ALiveDirectoryLinkIsADirectoryErrorOnReadAndALinkRefusalOnWrite()
{
var directory = Path.Combine(_root, "real-directory");
var link = Path.Combine(_root, "directory-link.cc");
Directory.CreateDirectory(directory);
CreateDirectoryLink(link, directory);
Assert.Catch<IOException>(() => CultMesh.ReadSingleFileDocumentPayload(link, Key, "raw:schema"))!.Message.Should().Contain("directory");
Assert.Catch<IOException>(() => RecordsSeenByACache(link))!.Message.Should().Contain("directory");
Assert.Throws<CultStoreLinkedException>(() => CultMesh.WriteSingleFileDocument(link, Key, Publication()));
Directory.GetFileSystemEntries(directory).Should().BeEmpty();
IsALink(link).Should().BeTrue();
}
private static void CreateDirectoryLink(string path, string target)
{
if (!OperatingSystem.IsWindows())
{
var target = Path.Combine(_root, name + "-target.cc");
var link = Path.Combine(_root, name + "-link.cc");
CultMesh.WriteSingleFileDocument(target, Key, Publication("first"));
CreateLiveLinkOrIgnore(link, target);
var before = File.ReadAllBytes(target);
write(link);
CultMesh.ReadSingleFileDocument<CultMeshBodyPublicationDocument>(link, Key).ProducerId.Should().Be("second", name);
return $"linkKept={IsALink(link)} targetChanged={!File.ReadAllBytes(target).SequenceEqual(before)}";
Directory.CreateSymbolicLink(path, target);
return;
}
var viaMesh = Outcome("mesh", link => CultMesh.WriteSingleFileDocument(link, Key, Publication("second")));
var viaCache = Outcome("cache", link =>
var mklink = Process.Start(new ProcessStartInfo("cmd.exe", $"/c mklink /J \"{path}\" \"{target}\"")
{
using var cache = CultCacheMessagePack.Create(link, ReadOnlyOpen);
cache.Commit(batch => batch.Upsert(typeof(CultMeshBodyPublicationDocument), Publication("second"), Key));
});
viaMesh.Should().Be(viaCache);
CreateNoWindow = true,
RedirectStandardOutput = true,
RedirectStandardError = true,
})!;
mklink.WaitForExit();
Assert.That(mklink.ExitCode, Is.Zero, "mklink /J failed");
}
}