From a06d3f71f9271667e8d89876372a7e3a2e71114f Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sun, 13 Sep 2026 17:57:59 +0200 Subject: [PATCH] Prove conditional commit on single-file and directory stores ConditionalCommitTests: StaleExpectFailsWithNothingChanged, ExpectNullIsCreateOnce, PerEntryCommitSurvivesUnrelatedWrite, ExpectUnchangedFailsOnUnrelatedInsert, TwoWritersRacingExactlyOneWins, TryCommitReportsContended, PlainFlushNeverInterleaves, StoredAtIsStrictlyIncreasingPerKey, DirtyStoreRefusesConditionalCommit, CrossStoreConditionalBatchThrowsBeforeAnyStore, DirectoryStoreHonorsConditions (the four condition tests on a directory store). The race is in-process: two store instances on one file, two threads, the final count exactly 2 + 2N. The lock is a FileShare.None handle, which excludes a second handle in this process exactly as in another, so the cross-process TwoProcessesRacingExactlyOneWins and its console project are not added. Each writer's mismatch is scripted before the free race, so "each writer reports at least one mismatch" does not depend on scheduling. PlainFlushNeverInterleaves counts an open that loses to a replace in flight as no read; every read that gets bytes must deserialize. Co-Authored-By: Claude Opus 5 --- .../ConditionalCommitTests.cs | 348 ++++++++++++++++++ 1 file changed, 348 insertions(+) create mode 100644 tests/GameCult.Caching.Tests/ConditionalCommitTests.cs diff --git a/tests/GameCult.Caching.Tests/ConditionalCommitTests.cs b/tests/GameCult.Caching.Tests/ConditionalCommitTests.cs new file mode 100644 index 00000000..49e6dc5b --- /dev/null +++ b/tests/GameCult.Caching.Tests/ConditionalCommitTests.cs @@ -0,0 +1,348 @@ +#nullable enable +using System; +using System.IO; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using GameCult.Caching; +using GameCult.Caching.MessagePack; +using MessagePack; +using NUnit.Framework; +using R3; + +namespace GameCult.Caching.Tests +{ + public class ConditionalCommitTests + { + private static readonly CultRecordKey Key = new("counter"); + private string _directory = string.Empty; + + [SetUp] + public void CreateDirectory() + { + _directory = Path.Combine(Path.GetTempPath(), $"cultlib-conditional-{Guid.NewGuid():N}"); + Directory.CreateDirectory(_directory); + } + + [TearDown] + public void DeleteDirectory() + { + if (Directory.Exists(_directory)) + Directory.Delete(_directory, recursive: true); + } + + [Test] + public void StaleExpectFailsWithNothingChanged() => StaleExpect(SingleFile(PathOf("store.cc"))); + + [Test] + public void ExpectNullIsCreateOnce() => CreateOnce(SingleFile(PathOf("store.cc"))); + + [Test] + public void PerEntryCommitSurvivesUnrelatedWrite() => UnrelatedWrite(SingleFile(PathOf("store.cc"))); + + [Test] + public void ExpectUnchangedFailsOnUnrelatedInsert() => UnrelatedInsert(SingleFile(PathOf("store.cc"))); + + [Test] + public void DirectoryStoreHonorsConditions() + { + StaleExpect(DirectoryStore(PathOf("stale.cc"))); + CreateOnce(DirectoryStore(PathOf("create.cc"))); + UnrelatedWrite(DirectoryStore(PathOf("unrelated.cc"))); + UnrelatedInsert(DirectoryStore(PathOf("insert.cc"))); + } + + // The lock is a FileShare.None handle, which excludes another handle in this process exactly as it does one in + // another process, so two store instances on two threads prove the property without a process boundary. + [Test] + public void TwoWritersRacingExactlyOneWins() + { + const int n = 100; + var store = SingleFile(PathOf("race.cc")); + Create(store, 0); + using var a = store.Open(); + using var b = store.Open(); + + Assert.That(Increment(a), Is.True); + Assert.That(Increment(b), Is.False, "b observed before a committed and still won"); + b.PullAllBackingStoresAsync(); + Assert.That(Increment(b), Is.True); + Assert.That(Increment(a), Is.False, "a committed over b's newer record from its stale view"); + + var racers = new[] { a, b }.Select(cache => Task.Run(() => + { + var wins = 0; + for (var attempt = 0; wins < n && attempt < 50 * n; attempt++) + { + cache.PullAllBackingStoresAsync(); + if (Increment(cache)) + wins++; + } + + return wins; + })).ToArray(); + Task.WaitAll(racers, TimeSpan.FromMinutes(2)); + + Assert.That(racers.Select(racer => racer.Result), Is.All.EqualTo(n)); + using var reader = store.Open(); + Assert.That(reader.Get(Key)!.Value, Is.EqualTo(2 + 2 * n)); + } + + [Test] + public void TryCommitReportsContended() + { + var path = PathOf("contended.cc"); + var store = SingleFile(path); + Create(store, 0); + using var cache = store.Open(); + var before = File.ReadAllBytes(path); + Task waiting; + + using (new FileStream(path + ".lock", FileMode.OpenOrCreate, FileAccess.ReadWrite, FileShare.None)) + { + Assert.That(cache.TryCommit(batch => batch.Upsert(new Counter { Name = "x" }, new CultRecordHandle(new CultRecordKey("x")))), + Is.EqualTo(CultCommitOutcome.Contended)); + Assert.That(File.ReadAllBytes(path), Is.EqualTo(before)); + Assert.That(cache.Get(new CultRecordKey("x")), Is.Null); + + waiting = Task.Run(() => cache.Commit(batch => batch.Upsert(new Counter { Name = "y" }, new CultRecordHandle(new CultRecordKey("y"))))); + Assert.That(waiting.Wait(200), Is.False, "Commit did not wait for the held lock"); + } + + Assert.That(waiting.Wait(TimeSpan.FromSeconds(10)) && waiting.Result, Is.True); + Assert.That(cache.Get(new CultRecordKey("y")), Is.Not.Null); + } + + [Test] + public void PlainFlushNeverInterleaves() + { + var path = PathOf("interleave.cc"); + var store = SingleFile(path); + Create(store, 0); + using var flusher = store.Open(); + using var committer = store.Open(); + var running = true; + var reads = 0; + + var flushing = Task.Run(() => + { + for (var i = 0; i < 50; i++) + { + flusher.UpsertAsync(new Counter { Name = "flushed", Value = i }, new CultRecordHandle(new CultRecordKey("flushed"))); + flusher.FlushAllBackingStores(); + } + }); + var committing = Task.Run(() => + { + for (var i = 0; i < 50; i++) + { + var key = new CultRecordKey($"committed:{i}"); + committer.Commit(batch => + { + batch.Expect(key, null); + batch.Upsert(new Counter { Name = key.Value, Value = i }, new CultRecordHandle(key)); + }); + } + }); + var reading = Task.Run(() => + { + while (Volatile.Read(ref running)) + { + byte[] bytes; + try + { + using var stream = new FileStream(path, FileMode.Open, FileAccess.Read, FileShare.ReadWrite | FileShare.Delete); + using var buffer = new MemoryStream(); + stream.CopyTo(buffer); + bytes = buffer.ToArray(); + } + catch (IOException) + { + // An open that loses to a replace in flight (moved away, or held by the replace) is not a read. + continue; + } + + CultDocumentMessagePackSerialization.DeserializeSnapshot(bytes); + reads++; + } + }); + + Task.WaitAll(new[] { flushing, committing }, TimeSpan.FromMinutes(2)); + Volatile.Write(ref running, false); + Assert.That(() => reading.Wait(TimeSpan.FromSeconds(10)), Throws.Nothing); + Assert.That(reads, Is.GreaterThan(0)); + } + + [Test] + public void StoredAtIsStrictlyIncreasingPerKey() + { + using var cache = SingleFile(PathOf("stamps.cc")).Open(); + var stamps = new string[1000]; + for (var i = 0; i < stamps.Length; i++) + { + cache.UpsertAsync(new Counter { Name = "stamped", Value = i }, new CultRecordHandle(Key)); + stamps[i] = cache.AllStoredDocuments.Single().StoredAt; + } + + Assert.That(stamps.Distinct().Count(), Is.EqualTo(stamps.Length)); + var instants = stamps.Select(stamp => DateTimeOffset.Parse(stamp, null, System.Globalization.DateTimeStyles.RoundtripKind)).ToArray(); + Assert.That(instants.Zip(instants.Skip(1), (earlier, later) => later > earlier), Is.All.True); + } + + [Test] + public void DirtyStoreRefusesConditionalCommit() + { + using var cache = SingleFile(PathOf("dirty.cc")).Open(); + cache.UpsertAsync(new Counter { Name = "staged" }); + + Assert.That(() => cache.Commit(batch => + { + batch.Expect(Key, null); + batch.Upsert(new Counter { Name = "counter" }, new CultRecordHandle(Key)); + }), Throws.InvalidOperationException.With.Message.Contains("flush")); + Assert.That(cache.Get(Key), Is.Null); + } + + [Test] + public void CrossStoreConditionalBatchThrowsBeforeAnyStore() + { + using var cache = new CultCache(Registry); + cache.AddBackingStore(new SingleFileMessagePackBackingStore(PathOf("counters.cc")), typeof(Counter)); + cache.AddBackingStore(new SingleFileMessagePackBackingStore(PathOf("tallies.cc")), typeof(Tally)); + + Assert.That(() => cache.Commit(batch => + { + batch.Expect(Key, null); + batch.Upsert(new Counter { Name = "counter" }, new CultRecordHandle(Key)); + batch.Upsert(new Tally { Name = "tally" }, new CultRecordHandle(new CultRecordKey("tally"))); + }), Throws.InvalidOperationException.With.Message.Contains("one home store")); + Assert.That(System.IO.Directory.GetFiles(_directory), Is.Empty, "a store file or lock file was touched"); + } + + private static void StaleExpect(StoreFactory store) + { + Create(store, 0); + using var a = store.Open(); + using var b = store.Open(); + Assert.That(Increment(a), Is.True); + var before = File.ReadAllBytes(store.Path); + var observedByB = b.Get(Key)!; + var changes = 0; + using var subscription = b.Watch().Subscribe(_ => changes++); + + Assert.That(Commit(b, observedByB, 100), Is.False, $"{store.Path}: b's stale Expect committed"); + Assert.That(File.ReadAllBytes(store.Path), Is.EqualTo(before)); + Assert.That(b.Get(Key), Is.SameAs(observedByB)); + Assert.That(b.IsDirty, Is.False); + Assert.That(changes, Is.Zero); + } + + private static void CreateOnce(StoreFactory store) + { + using var a = store.Open(); + using var b = store.Open(); + Assert.That(CreateCommit(a, 1), Is.True); + Assert.That(CreateCommit(b, 2), Is.False, $"{store.Path}: a second create-once commit won"); + using var reader = store.Open(); + Assert.That(reader.Get(Key)!.Value, Is.EqualTo(1)); + } + + private static void UnrelatedWrite(StoreFactory store) + { + Create(store, 0); + using var a = store.Open(); + using var b = store.Open(); + Assert.That(a.Commit(batch => batch.Upsert(new Counter { Name = "b" }, new CultRecordHandle(new CultRecordKey("b")))), Is.True); + + Assert.That(Increment(b), Is.True, $"{store.Path}: an unrelated write failed a per-entry condition"); + using var reader = store.Open(); + Assert.That(reader.Get(Key)!.Value, Is.EqualTo(1)); + Assert.That(reader.Get(new CultRecordKey("b")), Is.Not.Null, $"{store.Path}: the commit dropped a record it never loaded"); + } + + private static void UnrelatedInsert(StoreFactory store) + { + Create(store, 0); + using var a = store.Open(); + using var b = store.Open(); + Assert.That(a.Commit(batch => batch.Upsert(new Counter { Name = "c" }, new CultRecordHandle(new CultRecordKey("c")))), Is.True); + + Assert.That(b.Commit(batch => + { + batch.ExpectUnchanged(); + batch.Upsert(new Counter { Name = "counter", Value = 5 }, new CultRecordHandle(Key)); + }), Is.False, $"{store.Path}: ExpectUnchanged ignored an insert"); + } + + private static void Create(StoreFactory store, int value) + { + using var cache = store.Open(); + Assert.That(CreateCommit(cache, value), Is.True); + } + + private static bool CreateCommit(CultCache cache, int value) => cache.Commit(batch => + { + batch.Expect(Key, null); + batch.Upsert(new Counter { Name = "counter", Value = value }, new CultRecordHandle(Key)); + }); + + private static bool Increment(CultCache cache) + { + var observed = cache.Get(Key)!; + return Commit(cache, observed, observed.Value + 1); + } + + private static bool Commit(CultCache cache, Counter observed, int value) => cache.Commit(batch => + { + batch.Expect(Key, observed); + batch.Upsert(new Counter { Name = "counter", Value = value }, new CultRecordHandle(Key)); + }); + + private string PathOf(string fileName) => System.IO.Path.Combine(_directory, fileName); + + private static readonly CultDocumentRegistry Registry = CultDocumentRegistry.ForTypes(new[] { typeof(Counter), typeof(Tally) }); + + private static StoreFactory SingleFile(string path) => new(path, () => new SingleFileMessagePackBackingStore(path)); + + private static StoreFactory DirectoryStore(string path) => new(path, () => new DirectoryMessagePackBackingStore(path)); + + private sealed class StoreFactory + { + private readonly Func _create; + + public StoreFactory(string path, Func create) + { + Path = path; + _create = create; + } + + public string Path { get; } + + public CultCache Open() + { + var cache = new CultCache(Registry); + cache.AddBackingStore(_create()); + return cache; + } + } + + [CultDocument("tests.conditional_counter", "tests.conditional_counter.v1")] + internal sealed class Counter + { + [Key(0)] + [CultName] + public string Name = string.Empty; + + [Key(1)] + public int Value; + } + + [CultDocument("tests.conditional_tally", "tests.conditional_tally.v1")] + internal sealed class Tally + { + [Key(0)] + [CultName] + public string Name = string.Empty; + } + } +}