Caching: Update reads version and copy from one durable store read (update-copy-from-stored-bytes) #170
7 changed files with 586 additions and 51 deletions
|
|
@ -72,8 +72,8 @@ public static class FirstCall
|
|||
|
||||
## C# update with retry
|
||||
|
||||
Change a record you may not be the only one changing. `Update` reads the record, hands your function a private
|
||||
copy, and commits the result only if the record is still the one it read. If another writer got there first, it
|
||||
Change a record you may not be the only one changing. `Update` reads the record as its store holds it, hands your function a fresh
|
||||
copy of it, and commits the result only if the record is still the one it read. If another writer got there first, it
|
||||
reads again and runs your function again, up to five times. The source is
|
||||
[`tests/GameCult.Caching.Tests/UpdateCallTests.cs`](../../tests/GameCult.Caching.Tests/UpdateCallTests.cs); the
|
||||
snippet reuses `Item` from the first call.
|
||||
|
|
|
|||
|
|
@ -64,48 +64,7 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
|
|||
var reports = new List<CultSchemaMigrationReport>();
|
||||
var loaded = new Dictionary<string, CultStoredDocument>(StringComparer.Ordinal);
|
||||
var foreign = new List<CultForeignRecord>();
|
||||
for (var attempt = 1; ; attempt++)
|
||||
{
|
||||
loaded.Clear();
|
||||
reports.Clear();
|
||||
foreign.Clear();
|
||||
// Only pages named by the manifest read under the lease are loaded; orphaned pages are never loaded. With no
|
||||
// lock to lease (none created yet, or a store copied without it) a writer may commit mid-load, so an unleased
|
||||
// load stands only if its manifest is still current; if the manifest moved it reloads, under the lease the
|
||||
// writer created. A page that fails under an unmoved manifest is corruption, not a race.
|
||||
using var lease = AcquireCommitLease(wait: true, create: false);
|
||||
var manifestBytes = ReadManifestBytes();
|
||||
var (manifest, _) = ParseManifest(manifestBytes);
|
||||
Trace($"manifest records={manifest.Records.Length}");
|
||||
try
|
||||
{
|
||||
if (lease == null)
|
||||
UnleasedManifestRead?.Invoke();
|
||||
LoadRecordPages(
|
||||
manifest.Records.OrderBy(record => record.Key, StringComparer.Ordinal).ToArray(),
|
||||
manifest.SchemaCatalog,
|
||||
loaded,
|
||||
reports,
|
||||
foreign);
|
||||
if (lease != null || !ManifestMoved(manifestBytes))
|
||||
{
|
||||
Trace($"indexed-pages loaded={loaded.Count}");
|
||||
break;
|
||||
}
|
||||
}
|
||||
catch (Exception exception) when (lease == null && IsTornGeneration(exception))
|
||||
{
|
||||
// The re-read stays out of the filter: an exception thrown inside a filter is swallowed as false.
|
||||
if (!ManifestMoved(manifestBytes))
|
||||
throw;
|
||||
if (attempt == UnleasedLoadAttempts)
|
||||
throw Unsettled(exception);
|
||||
continue;
|
||||
}
|
||||
|
||||
if (attempt == UnleasedLoadAttempts)
|
||||
throw Unsettled(null);
|
||||
}
|
||||
LoadManifestPages(null, loaded, reports, foreign, Trace);
|
||||
|
||||
bool Staged(string key) => _dirtyKeys.ContainsKey(key) || _deletedKeys.ContainsKey(key);
|
||||
var arrived = loaded
|
||||
|
|
@ -136,6 +95,70 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
|
|||
Trace("publish");
|
||||
}
|
||||
|
||||
// The one load of manifest pages, for the whole store (key null) or one record: under the lease, with the moved-manifest retry.
|
||||
// It reads only; adopting what it loaded is the caller's.
|
||||
private void LoadManifestPages(
|
||||
string? key,
|
||||
Dictionary<string, CultStoredDocument> loaded,
|
||||
List<CultSchemaMigrationReport> reports,
|
||||
List<CultForeignRecord> foreign,
|
||||
Action<string> trace)
|
||||
{
|
||||
for (var attempt = 1; ; attempt++)
|
||||
{
|
||||
loaded.Clear();
|
||||
reports.Clear();
|
||||
foreign.Clear();
|
||||
// Only pages named by the manifest read under the lease are loaded; orphaned pages are never loaded. With no
|
||||
// lock to lease (none created yet, or a store copied without it) a writer may commit mid-load, so an unleased
|
||||
// load stands only if its manifest is still current; if the manifest moved it reloads, under the lease the
|
||||
// writer created. A page that fails under an unmoved manifest is corruption, not a race.
|
||||
using var lease = AcquireCommitLease(wait: true, create: false);
|
||||
var manifestBytes = ReadManifestBytes();
|
||||
var (manifest, _) = ParseManifest(manifestBytes);
|
||||
trace($"manifest records={manifest.Records.Length}");
|
||||
try
|
||||
{
|
||||
if (lease == null)
|
||||
UnleasedManifestRead?.Invoke();
|
||||
LoadRecordPages(
|
||||
manifest.Records.Where(record => key == null || string.Equals(record.Key, key, StringComparison.Ordinal))
|
||||
.OrderBy(record => record.Key, StringComparer.Ordinal).ToArray(),
|
||||
manifest.SchemaCatalog,
|
||||
loaded,
|
||||
reports,
|
||||
foreign);
|
||||
if (lease != null || !ManifestMoved(manifestBytes))
|
||||
{
|
||||
trace($"indexed-pages loaded={loaded.Count}");
|
||||
break;
|
||||
}
|
||||
}
|
||||
catch (Exception exception) when (lease == null && IsTornGeneration(exception))
|
||||
{
|
||||
// The re-read stays out of the filter: an exception thrown inside a filter is swallowed as false.
|
||||
if (!ManifestMoved(manifestBytes))
|
||||
throw;
|
||||
if (attempt == UnleasedLoadAttempts)
|
||||
throw Unsettled(exception);
|
||||
continue;
|
||||
}
|
||||
|
||||
if (attempt == UnleasedLoadAttempts)
|
||||
throw Unsettled(null);
|
||||
}
|
||||
}
|
||||
|
||||
// The record at a key as its page holds it now, decoded fresh and stamped with VersionOf its manifest payload, the version
|
||||
// CommitBatch compares. Null when none is there or it is foreign. Touches no entry, hydration, report or foreign list of the
|
||||
// store, and raises nothing.
|
||||
internal override CultStoredDocument? ReadDurable(string key) => Held(() =>
|
||||
{
|
||||
var loaded = new Dictionary<string, CultStoredDocument>(StringComparer.Ordinal);
|
||||
LoadManifestPages(key, loaded, new List<CultSchemaMigrationReport>(), new List<CultForeignRecord>(), _ => { });
|
||||
return loaded.TryGetValue(key, out var stored) ? stored : null;
|
||||
});
|
||||
|
||||
// Variants wait for the directory store's own cut; until then it refuses one rather than persist a resolved view as a plain page.
|
||||
private static void RefuseVariant(CultStoredDocument entry)
|
||||
{
|
||||
|
|
|
|||
|
|
@ -2682,34 +2682,66 @@ namespace GameCult.Caching
|
|||
}
|
||||
|
||||
/// <summary>
|
||||
/// Reads the record at <paramref name="key"/>, applies <paramref name="apply"/> to a private copy of it, and commits the result only
|
||||
/// if the record is still the one that was read; when another writer got there first it re-reads and runs <paramref name="apply"/>
|
||||
/// again, up to <paramref name="maxAttempts"/> times. The write is durable when Update returns.
|
||||
/// Reads the record at <paramref name="key"/> as its home store holds it, applies <paramref name="apply"/> to a fresh copy of it, and
|
||||
/// commits the result only if the record is still the one that was read; when another writer got there first it re-reads and runs
|
||||
/// <paramref name="apply"/> again, up to <paramref name="maxAttempts"/> times. The write is durable when Update returns.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <paramref name="apply"/> may run up to <paramref name="maxAttempts"/> times. It gets a private copy (null when the key holds
|
||||
/// nothing), which it may change and return. It must not write to the cache or have other side effects. Return null to write
|
||||
/// nothing. A store with staged blind writes refuses Update until it is flushed, as it refuses any conditional commit. The cache
|
||||
/// needs a codec to copy with. Running out of attempts returns <see cref="CultUpdateOutcome.LostRace"/>; every other failure throws.
|
||||
/// <paramref name="apply"/> may run up to <paramref name="maxAttempts"/> times. It gets the record as the store that holds it has it now, read from every home
|
||||
/// of T and its routed subtypes and decoded afresh for this attempt (null when the key holds nothing), which it may change and return. A record held by two stores, or an apply that changes the record's home store, throws. A change made in place to an
|
||||
/// instance the cache holds is neither seen nor written. It must not write to the cache or have other side effects. Return null to write
|
||||
/// nothing. A store with staged blind writes refuses Update until it is flushed, as it refuses any conditional commit. A cache with no
|
||||
/// store attached copies its held instance and needs a codec; a store kind outside the library cannot serve Update (<see cref="NotSupportedException"/>). Running out of attempts returns <see cref="CultUpdateOutcome.LostRace"/>; every other failure throws.
|
||||
/// </remarks>
|
||||
public CultUpdateResult<T> Update<T>(CultRecordKey key, Func<T?, T?> apply, int maxAttempts = 5) where T : class
|
||||
{
|
||||
if (apply == null) throw new ArgumentNullException(nameof(apply));
|
||||
if (maxAttempts < 1) throw new ArgumentOutOfRangeException(nameof(maxAttempts), maxAttempts, "At least one attempt is required.");
|
||||
var codec = _codec ?? throw new InvalidOperationException("This cache has no codec; create it with CultCacheMessagePack.Create.");
|
||||
for (var attempt = 1; ; attempt++)
|
||||
{
|
||||
var (version, copy) = Held(() =>
|
||||
var (version, copy, source) = Held(() =>
|
||||
{
|
||||
if (!_entries.TryGetValue(key.Value, out var held))
|
||||
return ((string?)null, (T?)null);
|
||||
if (held.Variant != null || held.Document is not T)
|
||||
if (_stores.Count == 0)
|
||||
{
|
||||
// No store: the held entry is the record.
|
||||
_entries.TryGetValue(key.Value, out var held);
|
||||
var codec = _codec ?? throw new InvalidOperationException("This cache has no codec; create it with CultCacheMessagePack.Create.");
|
||||
if (held == null)
|
||||
return ((string?)null, (T?)null, (CacheBackingStore?)null);
|
||||
if (held.Variant != null || held.Document is not T)
|
||||
throw new InvalidOperationException($"Record {key.Value} is a variant or not a {typeof(T).Name}; Update cannot copy it.");
|
||||
return (held.StoredVersion, (T?)codec.Clone(held.Document, held.Descriptor.DocumentType), null);
|
||||
}
|
||||
|
||||
// The store with the key is among the homes of T and its routed subtypes; the cache's own view of the key may be stale, so it is not asked.
|
||||
var homes = CandidateHomes(typeof(T));
|
||||
if (homes.Length == 0)
|
||||
throw new InvalidOperationException("A conditional batch names its home store through a record it upserts or removes.");
|
||||
CacheBackingStore? holder = null;
|
||||
CultStoredDocument? durable = null;
|
||||
foreach (var candidate in homes)
|
||||
{
|
||||
if (candidate.ReadDurable(key.Value) is not { } found)
|
||||
continue;
|
||||
if (holder != null)
|
||||
throw new InvalidOperationException($"Record {key.Value} is held by two stores, {holder} and {candidate}; Update cannot choose.");
|
||||
holder = candidate;
|
||||
durable = found;
|
||||
}
|
||||
|
||||
if (durable == null)
|
||||
return ((string?)null, (T?)null, (CacheBackingStore?)null);
|
||||
if (durable.Variant != null || durable.Document is not T typed)
|
||||
throw new InvalidOperationException($"Record {key.Value} is a variant or not a {typeof(T).Name}; Update cannot copy it.");
|
||||
return (held.StoredVersion, (T?)codec.Clone(held.Document, held.Descriptor.DocumentType));
|
||||
CultElementIds.Plan(durable.Descriptor.DocumentType, durable.Document, key.Value, deterministic: true).Apply();
|
||||
return (durable.StoredVersion, typed, holder);
|
||||
});
|
||||
var next = apply(copy);
|
||||
if (next == null)
|
||||
return new CultUpdateResult<T>(CultUpdateOutcome.Declined, Get<T>(key));
|
||||
if (source != null && Held(() => Home(next.GetType())) is var landing && landing != source)
|
||||
throw new InvalidOperationException(
|
||||
$"Update of {key.Value} would move the record from {source} to {landing?.ToString() ?? "memory"}; remove it and write it at its new home.");
|
||||
CultCommitOutcome outcome;
|
||||
try
|
||||
{
|
||||
|
|
@ -2728,9 +2760,11 @@ namespace GameCult.Caching
|
|||
return new CultUpdateResult<T>(CultUpdateOutcome.Updated, next);
|
||||
if (outcome != CultCommitOutcome.Mismatch)
|
||||
throw new InvalidOperationException($"Update of {key.Value} was refused: {outcome}.");
|
||||
Held(() => { Home(next.GetType())?.PullAll(); return true; });
|
||||
if (attempt >= maxAttempts)
|
||||
{
|
||||
Held(() => { Home(next.GetType())?.PullAll(); return true; });
|
||||
return new CultUpdateResult<T>(CultUpdateOutcome.LostRace, Get<T>(key));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -3751,6 +3785,18 @@ namespace GameCult.Caching
|
|||
return changes;
|
||||
}
|
||||
|
||||
// Every store a record of this type, or of a registered subtype of it, can live in.
|
||||
private CacheBackingStore[] CandidateHomes(Type type)
|
||||
{
|
||||
var homes = new HashSet<CacheBackingStore>();
|
||||
if (Home(type) is { } own)
|
||||
homes.Add(own);
|
||||
foreach (var descriptor in _registry.AllDescriptors)
|
||||
if (type.IsAssignableFrom(descriptor.DocumentType) && Home(descriptor.DocumentType) is { } routed)
|
||||
homes.Add(routed);
|
||||
return _stores.Select(entry => entry.Store).Where(homes.Contains).ToArray();
|
||||
}
|
||||
|
||||
// The most derived routed home type assignable from the document type, else the untyped store.
|
||||
private CacheBackingStore? Home(Type documentType)
|
||||
{
|
||||
|
|
@ -4027,6 +4073,15 @@ namespace GameCult.Caching
|
|||
internal void UseRegistry(CultDocumentRegistry? registry) => _registry = registry;
|
||||
|
||||
public abstract void PullAll();
|
||||
|
||||
/// <summary>
|
||||
/// The record at <paramref name="key"/> as this store holds it now, decoded as a load decodes it by this call (a new instance nobody
|
||||
/// else holds) and stamped with the version this store's commit compares. Null when the store holds nothing there or holds it foreign.
|
||||
/// It changes none of the store's entries, staged writes or reports and raises nothing. Internal because only this library can stamp a version.
|
||||
/// </summary>
|
||||
internal virtual CultStoredDocument? ReadDurable(string key) =>
|
||||
throw new NotSupportedException($"{this} cannot read one record, which Update needs.");
|
||||
|
||||
public abstract void Push(CultStoredDocument entry);
|
||||
public abstract void Delete(CultStoredDocument entry);
|
||||
public void PushAll() => PushAll(soft: false);
|
||||
|
|
@ -5173,6 +5228,9 @@ namespace GameCult.Caching
|
|||
return ToStoredDocument(record, durable.Version(key), durable.Snapshot.SchemaCatalog, DeserializeMarked, out _);
|
||||
}
|
||||
|
||||
// Update's read: the durable record at the key, decoded afresh, stamped with the version CommitBatch compares.
|
||||
internal override CultStoredDocument? ReadDurable(string key) => Held(() => DurableAt(ReadSnapshot(), key));
|
||||
|
||||
// The file now holds every record written, under the id and at the storedAt the write gave it, and the version of those bytes:
|
||||
// the store's entries and the cache's say so. A copied record is on disk as the file had it, so nothing learns anything new about it: in particular the
|
||||
// ids it minted at load are still in memory only.
|
||||
|
|
|
|||
|
|
@ -0,0 +1,131 @@
|
|||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.IO;
|
||||
using GameCult.Caching.MessagePack;
|
||||
using MessagePack;
|
||||
using NUnit.Framework;
|
||||
|
||||
namespace GameCult.Caching.OutsideStoreTests
|
||||
{
|
||||
// variants:follow_up:update-copy-from-stored-bytes: Update takes its version and its copy from one durable read of the home
|
||||
// store, never from the held instance. Consumer side: public API only, so a change made in place to an instance the cache
|
||||
// hands out can never be committed as if it had been read.
|
||||
public class HeldInstanceUpdateTests
|
||||
{
|
||||
[CultDocument("tests.held_counter", "tests.held_counter.v1")]
|
||||
[MessagePackObject]
|
||||
public sealed class HeldCounter
|
||||
{
|
||||
[Key(0)] [CultName] public string Name { get; set; } = string.Empty;
|
||||
[Key(1)] public int Value { get; set; }
|
||||
}
|
||||
|
||||
private static readonly CultDocumentRegistry Registry = CultDocumentRegistry.ForTypes(new[] { typeof(HeldCounter) });
|
||||
private static readonly CultRecordKey Key = new("counter");
|
||||
|
||||
private string _directory = string.Empty;
|
||||
|
||||
[SetUp]
|
||||
public void CreateDirectory()
|
||||
{
|
||||
_directory = Path.Combine(Path.GetTempPath(), $"cultlib-held-update-{Guid.NewGuid():N}");
|
||||
Directory.CreateDirectory(_directory);
|
||||
}
|
||||
|
||||
[TearDown]
|
||||
public void DeleteDirectory()
|
||||
{
|
||||
if (Directory.Exists(_directory))
|
||||
Directory.Delete(_directory, recursive: true);
|
||||
}
|
||||
|
||||
private static CultCache Open(string path, bool directory)
|
||||
{
|
||||
var cache = new CultCache(Registry, CultCacheMessagePack.CreateCodec(Registry));
|
||||
cache.AddBackingStore(directory ? new DirectoryMessagePackBackingStore(path) : new SingleFileMessagePackBackingStore(path));
|
||||
cache.PullAllBackingStoresAsync().GetAwaiter().GetResult();
|
||||
return cache;
|
||||
}
|
||||
|
||||
private string Seed(bool directory, int value)
|
||||
{
|
||||
var path = Path.Combine(_directory, "held.cc");
|
||||
using var cache = Open(path, directory);
|
||||
Assert.That(cache.Commit(batch => batch.Upsert(new HeldCounter { Name = "counter", Value = value }, new CultRecordHandle<HeldCounter>(Key))), Is.True);
|
||||
return path;
|
||||
}
|
||||
|
||||
private static int StoredValue(string path, bool directory)
|
||||
{
|
||||
using var reader = Open(path, directory);
|
||||
return reader.Get<HeldCounter>(Key)!.Value;
|
||||
}
|
||||
|
||||
private static HeldCounter Plus(HeldCounter? counter, int by)
|
||||
{
|
||||
counter!.Value += by;
|
||||
return counter;
|
||||
}
|
||||
|
||||
[TestCase(false)]
|
||||
[TestCase(true)]
|
||||
public void InPlaceChangeToHeldInstanceIsNeitherAppliedNorPersisted(bool directory)
|
||||
{
|
||||
var path = Seed(directory, 1);
|
||||
using var cache = Open(path, directory);
|
||||
cache.Get<HeldCounter>(Key)!.Value = 999;
|
||||
var seen = new List<int>();
|
||||
|
||||
var result = cache.Update<HeldCounter>(Key, counter => { seen.Add(counter!.Value); return Plus(counter, 1); });
|
||||
|
||||
Assert.That(result.Outcome, Is.EqualTo(CultUpdateOutcome.Updated));
|
||||
Assert.That(seen, Is.EqualTo(new[] { 1 }), "apply saw the in-place change");
|
||||
Assert.That(result.Document!.Value, Is.EqualTo(2));
|
||||
Assert.That(StoredValue(path, directory), Is.EqualTo(2), "the in-place 999 was persisted");
|
||||
}
|
||||
|
||||
[TestCase(false)]
|
||||
[TestCase(true)]
|
||||
public void InPlaceChangeToAPreviousResultIsNotCommitted(bool directory)
|
||||
{
|
||||
var path = Seed(directory, 1);
|
||||
using var cache = Open(path, directory);
|
||||
var first = cache.Update<HeldCounter>(Key, counter => Plus(counter, 1));
|
||||
Assert.That(first.Outcome, Is.EqualTo(CultUpdateOutcome.Updated));
|
||||
first.Document!.Value = 999;
|
||||
var seen = new List<int>();
|
||||
|
||||
var second = cache.Update<HeldCounter>(Key, counter => { seen.Add(counter!.Value); return Plus(counter, 1); });
|
||||
|
||||
Assert.That(second.Outcome, Is.EqualTo(CultUpdateOutcome.Updated));
|
||||
Assert.That(seen, Is.EqualTo(new[] { 2 }), "apply ran more than once or saw the in-place change");
|
||||
Assert.That(StoredValue(path, directory), Is.EqualTo(3));
|
||||
}
|
||||
|
||||
[TestCase(false)]
|
||||
[TestCase(true)]
|
||||
public void DeclineAfterInPlaceChangeWritesNothing(bool directory)
|
||||
{
|
||||
var path = Seed(directory, 1);
|
||||
using var cache = Open(path, directory);
|
||||
cache.Get<HeldCounter>(Key)!.Value = 999;
|
||||
var before = Snapshot(path);
|
||||
|
||||
Assert.That(cache.Update<HeldCounter>(Key, _ => null).Outcome, Is.EqualTo(CultUpdateOutcome.Declined));
|
||||
|
||||
Assert.That(Snapshot(path), Is.EqualTo(before));
|
||||
Assert.That(StoredValue(path, directory), Is.EqualTo(1));
|
||||
}
|
||||
|
||||
// Every file of the store: the single file, or the manifest and its pages.
|
||||
private static Dictionary<string, byte[]> Snapshot(string path)
|
||||
{
|
||||
var files = new Dictionary<string, byte[]> { [path] = File.ReadAllBytes(path) };
|
||||
var pages = path + ".records";
|
||||
if (Directory.Exists(pages))
|
||||
foreach (var file in Directory.GetFiles(pages, "*", SearchOption.AllDirectories))
|
||||
files[file] = File.ReadAllBytes(file);
|
||||
return files;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -364,5 +364,18 @@ namespace GameCult.Caching.OutsideStoreTests
|
|||
Assert.That(() => cache.Commit(batch => batch.Remove(K)), Throws.InstanceOf<CultSchemaConflictException>(), "removal");
|
||||
Assert.That(disk.Records["k"], Is.EqualTo(before));
|
||||
}
|
||||
|
||||
// Only the library can stamp a version on a decoded record, so a store kind outside it cannot read one honestly: Update refuses before apply.
|
||||
[Test]
|
||||
public void UpdateOverAnOutsideStoreIsNotSupported()
|
||||
{
|
||||
var disk = new Disk();
|
||||
PutOnDisk(disk, AB, "k", new OutA { Name = "k", Value = 1 });
|
||||
using var cache = Open(disk, AB);
|
||||
var calls = 0;
|
||||
|
||||
Assert.That(() => cache.Update<OutA>(K, a => { calls++; return a; }), Throws.TypeOf<NotSupportedException>());
|
||||
Assert.That(calls, Is.EqualTo(0), "apply ran for a store that cannot serve Update");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -224,6 +224,28 @@ namespace GameCult.Caching.Tests
|
|||
Assert.That(AllIds(reread), Is.EqualTo(first), "the ids minted at load are the ids written, and stay through repeated writes");
|
||||
}
|
||||
|
||||
// Update reads the store's bytes, which hold no ids; it mints the ids the load minted, so the write keeps the ids the cache showed.
|
||||
[Test]
|
||||
public void UpdateOfAPreIdRecordPersistsTheIdsTheLoadMinted()
|
||||
{
|
||||
var path = PathOf("old.cc");
|
||||
WritePreIdStore(path, null, ("a", "a", 2));
|
||||
var key = new CultRecordKey("a");
|
||||
|
||||
string[] shown;
|
||||
using (var cache = Open(path))
|
||||
{
|
||||
shown = AllIds(cache.Get<IdDeck>(key)!);
|
||||
Assert.That(shown, Has.All.Not.Empty);
|
||||
var result = cache.Update<IdDeck>(key, deck => { deck!.Name = "a2"; return deck; });
|
||||
Assert.That(result.Outcome, Is.EqualTo(CultUpdateOutcome.Updated));
|
||||
}
|
||||
|
||||
var written = MessagePackSerializer.Deserialize<IdDeck>(DiskRecord(path, "a").Payload);
|
||||
Assert.That(written.Name, Is.EqualTo("a2"));
|
||||
Assert.That(AllIds(written), Is.EqualTo(shown), "the write replaced the ids the cache showed with random ones");
|
||||
}
|
||||
|
||||
[Test]
|
||||
public void MintElementIdsRewritesEveryPreIdRecordAndNothingElse()
|
||||
{
|
||||
|
|
|
|||
|
|
@ -19,7 +19,7 @@ namespace GameCult.Caching.Tests
|
|||
{
|
||||
private static readonly CultRecordKey Key = new("counter");
|
||||
private static readonly CultDocumentRegistry Registry =
|
||||
CultDocumentRegistry.ForTypes(new[] { typeof(UpdateCounter), typeof(UpdateTally), typeof(UpdateGear) });
|
||||
CultDocumentRegistry.ForTypes(new[] { typeof(UpdateCounter), typeof(UpdateTally), typeof(UpdateGear), typeof(UpdateBase), typeof(UpdateDerived) });
|
||||
|
||||
private string _directory = string.Empty;
|
||||
|
||||
|
|
@ -78,16 +78,22 @@ namespace GameCult.Caching.Tests
|
|||
using var a = Open(path, directory);
|
||||
using var b = Open(path, directory);
|
||||
var applied = 0;
|
||||
// Each round starts both writers together: an attempt reads fresh from the store, so only overlapping read-commit windows lose.
|
||||
var round = new Barrier(2);
|
||||
|
||||
var racers = new[] { a, b }.Select(cache => Task.Run(() =>
|
||||
{
|
||||
var results = new List<CultUpdateOutcome>();
|
||||
for (var i = 0; i < n; i++)
|
||||
{
|
||||
round.SignalAndWait();
|
||||
results.Add(cache.Update<UpdateCounter>(Key, counter =>
|
||||
{
|
||||
Interlocked.Increment(ref applied);
|
||||
return Plus(counter, 1);
|
||||
}, maxAttempts: 200).Outcome);
|
||||
}
|
||||
|
||||
return results;
|
||||
})).ToArray();
|
||||
Assert.That(Task.WaitAll(racers, TimeSpan.FromMinutes(2)), Is.True, "the racers did not finish");
|
||||
|
|
@ -172,6 +178,193 @@ namespace GameCult.Caching.Tests
|
|||
Assert.That(StoredValue(path, false), Is.EqualTo(1005));
|
||||
}
|
||||
|
||||
// No publication in the middle of an Update: a rival's write that cost this cache an attempt is never published on its own,
|
||||
// only the value this Update commits (a pull after every lost attempt would publish both).
|
||||
[TestCase(false)]
|
||||
[TestCase(true)]
|
||||
public void LostAttemptPublishesNothingMidUpdate(bool directory)
|
||||
{
|
||||
var path = PathOf("midupdate.cc");
|
||||
Seed(path, directory, 1);
|
||||
using var cache = Open(path, directory);
|
||||
using var rival = Open(path, directory);
|
||||
var published = new List<int>();
|
||||
using var subscription = cache.Watch<UpdateCounter>().Subscribe(change => published.Add(change.Document!.Value));
|
||||
var calls = 0;
|
||||
|
||||
var result = cache.Update<UpdateCounter>(Key, counter =>
|
||||
{
|
||||
if (++calls == 1)
|
||||
Rival(rival, 5);
|
||||
return Plus(counter, 1);
|
||||
});
|
||||
|
||||
Assert.That(result.Outcome, Is.EqualTo(CultUpdateOutcome.Updated));
|
||||
Assert.That(calls, Is.EqualTo(2));
|
||||
Assert.That(published, Is.EqualTo(new[] { 6 }), "the rival's 5 was published in the middle of the Update");
|
||||
}
|
||||
|
||||
// The directory store reads only the key's own page, so a damaged page of another record does not stop an Update.
|
||||
[Test]
|
||||
public void DirectoryUpdateReadsOnlyItsOwnPage()
|
||||
{
|
||||
var path = PathOf("pages.cc");
|
||||
var other = new CultRecordKey("other");
|
||||
using (var seed = Open(path, true))
|
||||
Assert.That(seed.Commit(batch => batch.Upsert(new UpdateCounter { Name = "other", Value = 9 }, new CultRecordHandle<UpdateCounter>(other))), Is.True);
|
||||
var otherPages = Directory.GetFiles(path + ".records", "*.msgpack");
|
||||
Assert.That(otherPages, Has.Length.EqualTo(1));
|
||||
Seed(path, true, 1);
|
||||
using var cache = Open(path, true);
|
||||
File.WriteAllBytes(otherPages[0], new byte[] { 1, 2, 3, 4, 5 });
|
||||
|
||||
var result = cache.Update<UpdateCounter>(Key, counter => Plus(counter, 1));
|
||||
|
||||
Assert.That(result.Outcome, Is.EqualTo(CultUpdateOutcome.Updated));
|
||||
Assert.That(result.Document!.Value, Is.EqualTo(2));
|
||||
}
|
||||
|
||||
// The durable read is not a load: it leaves the store's foreign list, its migration reports and what the cache publishes alone.
|
||||
[TestCase(false)]
|
||||
[TestCase(true)]
|
||||
public void ReadDurableLeavesTheStoreViewAlone(bool directory)
|
||||
{
|
||||
var path = PathOf("view.cc");
|
||||
Seed(path, directory, 1);
|
||||
using (var writer = Open(path, directory))
|
||||
Assert.That(writer.Commit(batch => batch.Upsert(new UpdateGear { Name = "gear", Power = 3 }, new CultRecordHandle<UpdateGear>(new CultRecordKey("gear")))), Is.True);
|
||||
var narrow = CultDocumentRegistry.ForTypes(new[] { typeof(UpdateCounter) });
|
||||
using var cache = new CultCache(narrow);
|
||||
cache.AddBackingStore(directory ? new DirectoryMessagePackBackingStore(path) : new SingleFileMessagePackBackingStore(path));
|
||||
using var rival = Open(path, directory);
|
||||
var store = cache.BackingStores.Single();
|
||||
var foreign = store.ForeignRecords;
|
||||
var reports = store.LastSchemaMigrationReports;
|
||||
Assert.That(foreign.Select(record => record.Key), Is.EqualTo(new[] { "gear" }), "the fixture holds a foreign record");
|
||||
Assert.That(reports, Is.Not.Empty, "the fixture holds a migration report");
|
||||
var foreignKeys = foreign.Select(record => (record.Key, record.SchemaId, record.StoredAt)).ToArray();
|
||||
var reportKinds = reports.Select(report => (report.PersistedSchemaName, report.Kind)).ToArray();
|
||||
var changes = 0;
|
||||
using var subscription = cache.Watch<UpdateCounter>().Subscribe(_ => changes++);
|
||||
var calls = 0;
|
||||
|
||||
var result = cache.Update<UpdateCounter>(Key, counter =>
|
||||
{
|
||||
if (++calls == 1)
|
||||
Rival(rival, 5);
|
||||
return Plus(counter, 1);
|
||||
});
|
||||
|
||||
Assert.That(result.Outcome, Is.EqualTo(CultUpdateOutcome.Updated));
|
||||
Assert.That(calls, Is.EqualTo(2), "the Update reread the key");
|
||||
Assert.That(store.ForeignRecords, Is.SameAs(foreign));
|
||||
Assert.That(store.LastSchemaMigrationReports, Is.SameAs(reports));
|
||||
Assert.That(store.ForeignRecords.Select(record => (record.Key, record.SchemaId, record.StoredAt)), Is.EqualTo(foreignKeys));
|
||||
Assert.That(store.LastSchemaMigrationReports.Select(report => (report.PersistedSchemaName, report.Kind)), Is.EqualTo(reportKinds));
|
||||
Assert.That(changes, Is.EqualTo(1), "only the committed write was published");
|
||||
}
|
||||
|
||||
// Routing: store A is home to UpdateBase, store B to UpdateDerived : UpdateBase. Land judges a derived record at B.
|
||||
private CultCache OpenRouted()
|
||||
{
|
||||
var cache = new CultCache(Registry);
|
||||
cache.AddBackingStore(new SingleFileMessagePackBackingStore(PathOf("a.cc")), typeof(UpdateBase));
|
||||
cache.AddBackingStore(new SingleFileMessagePackBackingStore(PathOf("b.cc")), typeof(UpdateDerived));
|
||||
return cache;
|
||||
}
|
||||
|
||||
private CultCache OpenOneRouted(string file, Type home)
|
||||
{
|
||||
var cache = new CultCache(Registry);
|
||||
cache.AddBackingStore(new SingleFileMessagePackBackingStore(PathOf(file)), home);
|
||||
return cache;
|
||||
}
|
||||
|
||||
private byte[]? BytesOf(string file) => File.Exists(PathOf(file)) ? File.ReadAllBytes(PathOf(file)) : null;
|
||||
|
||||
private static readonly CultRecordHandle<UpdateBase> BaseHandle = new(Key);
|
||||
|
||||
private static void RivalDerived(CultCache cache, int value) =>
|
||||
Assert.That(cache.Commit(batch => batch.Upsert<UpdateBase>(new UpdateDerived { Name = "counter", Value = value }, BaseHandle)), Is.True);
|
||||
|
||||
[Test]
|
||||
public void RoutedSubtypeStaleCacheConverges()
|
||||
{
|
||||
using var cache = OpenRouted();
|
||||
using var rival = OpenRouted();
|
||||
RivalDerived(rival, 1);
|
||||
var seen = new List<int?>();
|
||||
|
||||
var result = cache.Update<UpdateBase>(Key, document =>
|
||||
{
|
||||
seen.Add((document as UpdateDerived)?.Value);
|
||||
var derived = (UpdateDerived)document!;
|
||||
derived.Value += 1;
|
||||
return derived;
|
||||
});
|
||||
|
||||
Assert.That(seen, Is.EqualTo(new int?[] { 1 }), "apply did not see the rival's record in its home store, once");
|
||||
Assert.That(result.Outcome, Is.EqualTo(CultUpdateOutcome.Updated));
|
||||
using var reader = OpenRouted();
|
||||
Assert.That(((UpdateDerived)reader.Get<UpdateBase>(Key)!).Value, Is.EqualTo(2));
|
||||
}
|
||||
|
||||
[Test]
|
||||
public void RoutedSubtypeCreateLandsInItsHome()
|
||||
{
|
||||
using (var cache = OpenRouted())
|
||||
{
|
||||
var calls = 0;
|
||||
var result = cache.Update<UpdateBase>(Key, document =>
|
||||
{
|
||||
calls++;
|
||||
Assert.That(document, Is.Null);
|
||||
return new UpdateDerived { Name = "counter", Value = 1 };
|
||||
});
|
||||
Assert.That(calls, Is.EqualTo(1));
|
||||
Assert.That(result.Outcome, Is.EqualTo(CultUpdateOutcome.Updated));
|
||||
}
|
||||
|
||||
using (var inB = OpenOneRouted("b.cc", typeof(UpdateDerived)))
|
||||
Assert.That(inB.Get<UpdateBase>(Key), Is.InstanceOf<UpdateDerived>());
|
||||
using (var inA = OpenOneRouted("a.cc", typeof(UpdateBase)))
|
||||
Assert.That(inA.Get<UpdateBase>(Key), Is.Null, "the record landed in the wrong store");
|
||||
}
|
||||
|
||||
[Test]
|
||||
public void MovingTheRecordToAnotherHomeRefuses()
|
||||
{
|
||||
using (var rival = OpenRouted())
|
||||
RivalDerived(rival, 1);
|
||||
var before = (A: BytesOf("a.cc"), B: BytesOf("b.cc"));
|
||||
using var cache = OpenRouted();
|
||||
var calls = 0;
|
||||
|
||||
var refusal = Assert.Throws<InvalidOperationException>(() =>
|
||||
cache.Update<UpdateBase>(Key, _ => { calls++; return new UpdateBase { Name = "counter" }; }))!;
|
||||
|
||||
Assert.That(calls, Is.EqualTo(1), "a refusal was retried");
|
||||
Assert.That(refusal.Message, Does.Contain("a.cc").And.Contain("b.cc").And.Contain("move"));
|
||||
Assert.That(BytesOf("a.cc"), Is.EqualTo(before.A));
|
||||
Assert.That(BytesOf("b.cc"), Is.EqualTo(before.B));
|
||||
}
|
||||
|
||||
[Test]
|
||||
public void ARecordHeldByTwoStoresRefusesBeforeApply()
|
||||
{
|
||||
using var cache = OpenRouted();
|
||||
using (var inA = OpenOneRouted("a.cc", typeof(UpdateBase)))
|
||||
Assert.That(inA.Commit(batch => batch.Upsert(new UpdateBase { Name = "counter" }, BaseHandle)), Is.True);
|
||||
using (var inB = OpenOneRouted("b.cc", typeof(UpdateDerived)))
|
||||
Assert.That(inB.Commit(batch => batch.Upsert<UpdateBase>(new UpdateDerived { Name = "counter", Value = 1 }, BaseHandle)), Is.True);
|
||||
var calls = 0;
|
||||
|
||||
var refusal = Assert.Throws<InvalidOperationException>(() => cache.Update<UpdateBase>(Key, document => { calls++; return document; }))!;
|
||||
|
||||
Assert.That(calls, Is.EqualTo(0));
|
||||
Assert.That(refusal.Message, Does.Contain("held by two stores"));
|
||||
}
|
||||
|
||||
[Test]
|
||||
public void NullFromApplyDeclines()
|
||||
{
|
||||
|
|
@ -219,8 +412,19 @@ namespace GameCult.Caching.Tests
|
|||
var calls = 0;
|
||||
Func<UpdateCounter?, UpdateCounter?> apply = counter => { calls++; return Plus(counter, 1); };
|
||||
|
||||
using (var codecLess = Open(path, false, codec: false))
|
||||
Assert.That(() => codecLess.Update(Key, apply), Throws.InvalidOperationException);
|
||||
// A cache with no store copies its held instance, so it needs a codec.
|
||||
using (var storeLess = new CultCache(Registry))
|
||||
{
|
||||
storeLess.Commit(batch => batch.Upsert(new UpdateCounter { Name = "counter", Value = 1 }, new CultRecordHandle<UpdateCounter>(Key)));
|
||||
Assert.That(() => storeLess.Update(Key, apply), Throws.InvalidOperationException.With.Message.Contains("no codec"));
|
||||
}
|
||||
|
||||
// Stores are attached, but none is home to the type: refused before apply.
|
||||
using (var noHome = new CultCache(Registry, CultCacheMessagePack.CreateCodec(Registry)))
|
||||
{
|
||||
noHome.AddBackingStore(new SingleFileMessagePackBackingStore(PathOf("nohome.cc")), typeof(UpdateTally));
|
||||
Assert.That(() => noHome.Update(Key, apply), Throws.InvalidOperationException.With.Message.Contains("names its home store"));
|
||||
}
|
||||
using (var cache = Open(path, false))
|
||||
{
|
||||
Assert.That(() => cache.Update(Key, apply, maxAttempts: 0), Throws.TypeOf<ArgumentOutOfRangeException>());
|
||||
|
|
@ -241,7 +445,7 @@ namespace GameCult.Caching.Tests
|
|||
before = File.ReadAllBytes(path);
|
||||
var gearCalls = 0;
|
||||
using (var cache = Open(path, false))
|
||||
Assert.That(() => cache.Update<UpdateGear>(new CultRecordKey("gear-big"), gear => { gearCalls++; return gear; }), Throws.InvalidOperationException);
|
||||
Assert.That(() => cache.Update<UpdateGear>(new CultRecordKey("gear-big"), gear => { gearCalls++; return gear; }), Throws.InvalidOperationException.With.Message.Contains("Update cannot copy it"));
|
||||
Assert.That(gearCalls, Is.EqualTo(0), "apply ran for a variant");
|
||||
Assert.That(calls, Is.EqualTo(0), "apply ran for a refused call");
|
||||
Assert.That(File.ReadAllBytes(path), Is.EqualTo(before));
|
||||
|
|
@ -479,6 +683,73 @@ namespace GameCult.Caching.Tests
|
|||
Assert.That(cache.Get<UpdateCounter>(Key)!.Value, Is.EqualTo(200));
|
||||
}
|
||||
|
||||
[TestCase(false)]
|
||||
[TestCase(true)]
|
||||
public void StaleCacheFirstAttemptSeesTheStore(bool directory)
|
||||
{
|
||||
var path = PathOf("stale.cc");
|
||||
Seed(path, directory, 1);
|
||||
using var a = Open(path, directory);
|
||||
using var b = Open(path, directory);
|
||||
Assert.That(b.Get<UpdateCounter>(Key)!.Value, Is.EqualTo(1));
|
||||
Rival(a, 7);
|
||||
var seen = new List<int>();
|
||||
|
||||
var result = b.Update<UpdateCounter>(Key, counter => { seen.Add(counter!.Value); return Plus(counter, 1); });
|
||||
|
||||
Assert.That(result.Outcome, Is.EqualTo(CultUpdateOutcome.Updated));
|
||||
Assert.That(seen, Is.EqualTo(new[] { 7 }), "the first attempt read the cache's stale copy");
|
||||
Assert.That(StoredValue(path, directory), Is.EqualTo(8));
|
||||
}
|
||||
|
||||
// The durable read is not a load: it leaves the store's view of what the cache holds alone, so a later pull still delivers the rival's write.
|
||||
[TestCase(false)]
|
||||
[TestCase(true)]
|
||||
public void ADeclinedUpdateDoesNotHideTheRivalsWriteFromALaterPull(bool directory)
|
||||
{
|
||||
var path = PathOf("hidden.cc");
|
||||
Seed(path, directory, 1);
|
||||
using var a = Open(path, directory);
|
||||
using var b = Open(path, directory);
|
||||
Assert.That(b.Get<UpdateCounter>(Key)!.Value, Is.EqualTo(1));
|
||||
Rival(a, 7);
|
||||
|
||||
Assert.That(b.Update<UpdateCounter>(Key, _ => null).Outcome, Is.EqualTo(CultUpdateOutcome.Declined));
|
||||
Assert.That(b.Get<UpdateCounter>(Key)!.Value, Is.EqualTo(1), "a declined Update changed what the cache holds");
|
||||
b.PullAllBackingStoresAsync().GetAwaiter().GetResult();
|
||||
|
||||
Assert.That(b.Get<UpdateCounter>(Key)!.Value, Is.EqualTo(7));
|
||||
}
|
||||
|
||||
[Test]
|
||||
public void CodecLessStoreBackedCacheUpdates()
|
||||
{
|
||||
var path = PathOf("codecless.cc");
|
||||
Seed(path, false, 1);
|
||||
var calls = 0;
|
||||
using (var cache = Open(path, false, codec: false))
|
||||
Assert.That(cache.Update<UpdateCounter>(Key, counter => { calls++; return Plus(counter, 1); }).Outcome, Is.EqualTo(CultUpdateOutcome.Updated));
|
||||
|
||||
Assert.That(calls, Is.EqualTo(1));
|
||||
Assert.That(StoredValue(path, false), Is.EqualTo(2));
|
||||
}
|
||||
|
||||
[TestCase(false)]
|
||||
[TestCase(true)]
|
||||
public void AStoreHoldingNothingAtTheKeyIsAbsentEvenIfTheCacheHoldsIt(bool directory)
|
||||
{
|
||||
var path = PathOf("gone.cc");
|
||||
Seed(path, directory, 5);
|
||||
using var a = Open(path, directory);
|
||||
using var b = Open(path, directory);
|
||||
Assert.That(b.Get<UpdateCounter>(Key), Is.Not.Null);
|
||||
Assert.That(a.Commit(batch => batch.Remove(Key)), Is.True);
|
||||
var seen = new List<UpdateCounter?>();
|
||||
|
||||
Assert.That(b.Update<UpdateCounter>(Key, counter => { seen.Add(counter); return null; }).Outcome, Is.EqualTo(CultUpdateOutcome.Declined));
|
||||
Assert.That(seen, Is.EqualTo(new UpdateCounter?[] { null }));
|
||||
}
|
||||
|
||||
[CultDocument("tests.update_counter", "tests.update_counter.v1")]
|
||||
[MessagePackObject(AllowPrivate = true)]
|
||||
internal sealed class UpdateCounter
|
||||
|
|
@ -511,5 +782,22 @@ namespace GameCult.Caching.Tests
|
|||
[Key(1)]
|
||||
public int Power;
|
||||
}
|
||||
|
||||
[CultDocument("tests.update_base", "tests.update_base.v1")]
|
||||
[MessagePackObject(AllowPrivate = true)]
|
||||
internal class UpdateBase
|
||||
{
|
||||
[Key(0)]
|
||||
[CultName]
|
||||
public string Name = string.Empty;
|
||||
}
|
||||
|
||||
[CultDocument("tests.update_derived", "tests.update_derived.v1")]
|
||||
[MessagePackObject(AllowPrivate = true)]
|
||||
internal sealed class UpdateDerived : UpdateBase
|
||||
{
|
||||
[Key(1)]
|
||||
public int Value;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue