C# core: variant records, top-level Set overrides, resolution in the admitting hold
A variant record stores [baseKey, overrides[]] in record slot 4 and an empty payload; the cache resolves it against its base inside the hold that admits it or any record on its base chain, and re-resolves dependents in the same Apply, so a base change reaches its variants with contiguous sequences and no stamp. Plain Upsert at a variant key is refused (Flatten is the explicit way); cycle, missing base, different concrete type, evicted base with survivors and a name that shadows the base are refused naming the keys. Overrides of a dropped slot or an undecodable value are ignored with a warning and reported as compatible drift. The store writes cultcache.store.v2 only when it holds a variant and reads v1 and v2. The codec pair CultInspectorModel took becomes CultCodec, shared with the cache, which needs it to clone documents and overlay override values. The directory store, CultMesh single-file reads and CultNet row building refuse a variant by name until their own cuts. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens
parent
7584a5c08a
commit
dbc87b9447
11 changed files with 797 additions and 55 deletions
|
|
@ -15,6 +15,9 @@ Cross-runtime parity has three layers:
|
|||
without translation folklore.
|
||||
Every store reader refuses, by name, a header or record slot it does not
|
||||
understand (vectors: `tests/vectors/document-variants-c0`).
|
||||
Document variants: C# resolves them (`cultcache.store.v2`); every other runtime,
|
||||
CultMesh single-file reads, the directory store and CultNet refuse a variant by name
|
||||
(vector written by C#: `tests/vectors/document-variants-c1`).
|
||||
- Feature parity: each runtime must expose the CultCache/CultNet/CultMesh
|
||||
surfaces needed for its intended role, with unsupported production-server
|
||||
ownership called out plainly.
|
||||
|
|
|
|||
|
|
@ -19,13 +19,19 @@ namespace GameCult.Caching.MessagePack
|
|||
|
||||
public static class CultCacheMessagePack
|
||||
{
|
||||
// The inspection model over this codec. Values serialize without the document guard but with their owning document's
|
||||
// options, resolvers included, so dictionary keys compare by the bytes the store writes; clones deserialize as
|
||||
// registered documents.
|
||||
public static CultInspectorModel CreateInspectorModel(CultDocumentRegistry registry) => new CultInspectorModel(
|
||||
registry,
|
||||
// The store's encoding as one port. Values serialize without the document guard but with their owning document's
|
||||
// options, resolvers included, so dictionary keys compare by the bytes the store writes; a document type
|
||||
// deserializes as a registered document, a member type as itself.
|
||||
public static CultCodec CreateCodec(CultDocumentRegistry registry) => new CultCodec(
|
||||
(value, type, document) => global::MessagePack.MessagePackSerializer.Serialize(type, value, CultDocumentMessagePackSerialization.OptionsFor(document.Assembly)),
|
||||
(type, bytes) => CultDocumentMessagePackSerialization.DeserializeUntyped(type, bytes, registry));
|
||||
(type, document, bytes) => type == document
|
||||
? CultDocumentMessagePackSerialization.DeserializeUntyped(type, bytes, registry)
|
||||
: global::MessagePack.MessagePackSerializer.Deserialize(type, bytes, CultDocumentMessagePackSerialization.OptionsFor(document.Assembly)),
|
||||
CultDocumentMessagePackSerialization.OverlaySlots);
|
||||
|
||||
// The inspection model over that codec.
|
||||
public static CultInspectorModel CreateInspectorModel(CultDocumentRegistry registry) =>
|
||||
new CultInspectorModel(registry, CreateCodec(registry));
|
||||
|
||||
// Opening hydrates. A consumer that wants a fresh store removes the records it does not keep, upserts, and
|
||||
// flushes; the atomic replace does the rest.
|
||||
|
|
@ -37,7 +43,8 @@ namespace GameCult.Caching.MessagePack
|
|||
}
|
||||
|
||||
options ??= new CultCacheOpenOptions();
|
||||
var cache = new CultCache(options.Registry)
|
||||
var registry = options.Registry ?? CultDocumentRegistry.Shared;
|
||||
var cache = new CultCache(registry, CreateCodec(registry))
|
||||
{
|
||||
FlushAttachedStoresOnDispose = options.FlushOnDispose
|
||||
};
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
using System;
|
||||
using System.Buffers;
|
||||
using System.Collections.Concurrent;
|
||||
using System.Collections.Generic;
|
||||
using System.Linq;
|
||||
using System.Reflection;
|
||||
using GameCult.Caching;
|
||||
|
|
@ -30,8 +31,8 @@ public sealed class CultDocumentResolver : IFormatterResolver
|
|||
|
||||
public static class CultDocumentMessagePackSerialization
|
||||
{
|
||||
private const string SingleFileFormatVersion = "cultcache.store.v1";
|
||||
private const int PersistedRecordFieldCount = 4;
|
||||
private const int VariantRecordFieldCount = 5;
|
||||
private const int SchemaCatalogEntryFieldCount = 7;
|
||||
private const int SchemaCatalogMemberFieldCount = 8;
|
||||
private const int StoreSnapshotFieldCount = 3;
|
||||
|
|
@ -95,6 +96,49 @@ public static class CultDocumentMessagePackSerialization
|
|||
?? throw new InvalidOperationException($"MessagePack returned null for Cult document type {type.FullName}.");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Replaces slots of a serialized document payload (a MessagePack array indexed by slot) with already-encoded member
|
||||
/// values, leaving every other slot byte for byte as it was. A slot past the end extends the array with nil.
|
||||
/// </summary>
|
||||
public static byte[] OverlaySlots(byte[] payload, IReadOnlyList<KeyValuePair<int, byte[]>> slots)
|
||||
{
|
||||
var reader = new MessagePackReader(payload);
|
||||
if (reader.NextMessagePackType != MessagePackType.Array)
|
||||
{
|
||||
throw new InvalidOperationException("A document payload is a MessagePack array; this one is not, so its slots cannot be overridden.");
|
||||
}
|
||||
|
||||
var count = reader.ReadArrayHeader();
|
||||
var elements = new List<byte[]>(count);
|
||||
for (var index = 0; index < count; index++)
|
||||
{
|
||||
var start = reader.Position;
|
||||
reader.Skip();
|
||||
elements.Add(reader.Sequence.Slice(start, reader.Position).ToArray());
|
||||
}
|
||||
|
||||
foreach (var (slot, value) in slots)
|
||||
{
|
||||
while (elements.Count <= slot)
|
||||
{
|
||||
elements.Add(new byte[] { 0xc0 });
|
||||
}
|
||||
|
||||
elements[slot] = value;
|
||||
}
|
||||
|
||||
var buffer = new ArrayBufferWriter<byte>();
|
||||
var writer = new MessagePackWriter(buffer);
|
||||
writer.WriteArrayHeader(elements.Count);
|
||||
foreach (var element in elements)
|
||||
{
|
||||
writer.WriteRaw(element);
|
||||
}
|
||||
|
||||
writer.Flush();
|
||||
return buffer.WrittenSpan.ToArray();
|
||||
}
|
||||
|
||||
public static byte[] SerializePersistedRecord(CultPersistedRecord record)
|
||||
{
|
||||
var buffer = new global::System.Buffers.ArrayBufferWriter<byte>();
|
||||
|
|
@ -141,7 +185,7 @@ public static class CultDocumentMessagePackSerialization
|
|||
if (fieldCount > 0)
|
||||
{
|
||||
snapshot.FormatVersion = reader.ReadString()
|
||||
?? throw new NotSupportedException("Store snapshot declares no format version; this runtime reads " + SingleFileFormatVersion + " only.");
|
||||
?? throw new NotSupportedException("Store snapshot declares no format version; this runtime reads " + CultPersistedStoreSnapshot.FormatV1 + " and " + CultPersistedStoreSnapshot.FormatV2 + ".");
|
||||
}
|
||||
|
||||
if (fieldCount > 1)
|
||||
|
|
@ -172,24 +216,110 @@ public static class CultDocumentMessagePackSerialization
|
|||
return snapshot;
|
||||
}
|
||||
|
||||
/// <summary>Refuses a single-file snapshot this runtime cannot read, naming the format it found.</summary>
|
||||
/// <summary>Refuses a single-file snapshot this runtime cannot read, naming the format or record it found.</summary>
|
||||
public static void RequireSingleFileFormat(CultPersistedStoreSnapshot snapshot)
|
||||
{
|
||||
if (!string.Equals(snapshot.FormatVersion, SingleFileFormatVersion, StringComparison.Ordinal))
|
||||
var version = snapshot.FormatVersion;
|
||||
if (!string.Equals(version, CultPersistedStoreSnapshot.FormatV1, StringComparison.Ordinal) &&
|
||||
!string.Equals(version, CultPersistedStoreSnapshot.FormatV2, StringComparison.Ordinal))
|
||||
{
|
||||
throw new NotSupportedException(
|
||||
$"Store format {snapshot.FormatVersion} is not readable; this runtime reads {SingleFileFormatVersion} only. " +
|
||||
"The store needs a runtime that resolves document variants.");
|
||||
$"Store format {version} is not readable; this runtime reads {CultPersistedStoreSnapshot.FormatV1} and {CultPersistedStoreSnapshot.FormatV2}.");
|
||||
}
|
||||
|
||||
// Only a v2 store may hold a variant: a v1 header over a variant record is a store no reader could trust.
|
||||
if (string.Equals(version, CultPersistedStoreSnapshot.FormatV1, StringComparison.Ordinal))
|
||||
{
|
||||
var variant = snapshot.Records.FirstOrDefault(record => record.Variant != null);
|
||||
if (variant != null)
|
||||
{
|
||||
throw new NotSupportedException(
|
||||
$"Record '{variant.Key}' (schema '{variant.SchemaId}') is a variant but the store declares {version}; variants need {CultPersistedStoreSnapshot.FormatV2}.");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private static void WritePersistedRecord(ref MessagePackWriter writer, CultPersistedRecord record)
|
||||
{
|
||||
writer.WriteArrayHeader(PersistedRecordFieldCount);
|
||||
writer.WriteArrayHeader(record.Variant == null ? PersistedRecordFieldCount : VariantRecordFieldCount);
|
||||
writer.Write(record.Key);
|
||||
writer.Write(record.SchemaId);
|
||||
writer.Write(record.StoredAt);
|
||||
writer.Write(record.Payload);
|
||||
if (record.Variant != null)
|
||||
{
|
||||
// Slot 4: [baseKey, overrides[]]; an override is [op, path[], id, value], a path step [slot, elementId],
|
||||
// and the value is the member's own MessagePack encoding, inline.
|
||||
writer.WriteArrayHeader(2);
|
||||
writer.Write(record.Variant.BaseKey);
|
||||
writer.WriteArrayHeader(record.Variant.Overrides.Count);
|
||||
foreach (var entry in record.Variant.Overrides)
|
||||
{
|
||||
writer.WriteArrayHeader(4);
|
||||
writer.Write((int)entry.Op);
|
||||
writer.WriteArrayHeader(entry.Path.Count);
|
||||
foreach (var step in entry.Path)
|
||||
{
|
||||
writer.WriteArrayHeader(2);
|
||||
writer.Write(step.Slot);
|
||||
writer.Write(step.ElementId);
|
||||
}
|
||||
|
||||
writer.Write(entry.Id);
|
||||
writer.WriteRaw(entry.Value);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private static CultVariantDelta ReadVariant(ref MessagePackReader reader, string key, string schemaId)
|
||||
{
|
||||
var slots = reader.ReadArrayHeader();
|
||||
if (slots != 2)
|
||||
{
|
||||
throw new NotSupportedException(
|
||||
$"Record '{key}' (schema '{schemaId}') has a variant slot of {slots} entries; this runtime reads 2 (base key, overrides).");
|
||||
}
|
||||
|
||||
var baseKey = reader.ReadString() ?? string.Empty;
|
||||
var count = reader.ReadArrayHeader();
|
||||
var overrides = new CultVariantOverride[count];
|
||||
for (var index = 0; index < count; index++)
|
||||
{
|
||||
var fields = reader.ReadArrayHeader();
|
||||
if (fields != 4)
|
||||
{
|
||||
throw new NotSupportedException(
|
||||
$"Record '{key}' (schema '{schemaId}') has an override of {fields} entries; this runtime reads 4 (op, path, id, value).");
|
||||
}
|
||||
|
||||
var op = reader.ReadInt32();
|
||||
if (op is < 0 or > 2)
|
||||
{
|
||||
throw new NotSupportedException($"Record '{key}' (schema '{schemaId}') has an override with unknown op {op}.");
|
||||
}
|
||||
|
||||
var stepCount = reader.ReadArrayHeader();
|
||||
var path = new CultPathStep[stepCount];
|
||||
for (var step = 0; step < stepCount; step++)
|
||||
{
|
||||
var stepFields = reader.ReadArrayHeader();
|
||||
if (stepFields != 2)
|
||||
{
|
||||
throw new NotSupportedException(
|
||||
$"Record '{key}' (schema '{schemaId}') has a path step of {stepFields} entries; this runtime reads 2 (slot, elementId).");
|
||||
}
|
||||
|
||||
path[step] = new CultPathStep(reader.ReadInt32(), reader.ReadString() ?? string.Empty);
|
||||
}
|
||||
|
||||
var id = reader.ReadString() ?? string.Empty;
|
||||
var start = reader.Position;
|
||||
reader.Skip();
|
||||
var value = reader.Sequence.Slice(start, reader.Position).ToArray();
|
||||
overrides[index] = new CultVariantOverride((CultOverrideOp)op, path, id, value);
|
||||
}
|
||||
|
||||
return new CultVariantDelta(baseKey, overrides);
|
||||
}
|
||||
|
||||
public static CultPersistedRecord ReadPersistedRecord(ref MessagePackReader reader)
|
||||
|
|
@ -217,11 +347,21 @@ public static class CultDocumentMessagePackSerialization
|
|||
record.Payload = reader.ReadBytes()?.ToArray() ?? Array.Empty<byte>();
|
||||
}
|
||||
|
||||
if (fieldCount > PersistedRecordFieldCount)
|
||||
if (fieldCount > VariantRecordFieldCount)
|
||||
{
|
||||
throw new NotSupportedException(
|
||||
$"Record '{record.Key}' (schema '{record.SchemaId}') has {fieldCount} slots; this runtime reads {PersistedRecordFieldCount}. " +
|
||||
"The store needs a runtime that resolves document variants.");
|
||||
$"Record '{record.Key}' (schema '{record.SchemaId}') has {fieldCount} slots; this runtime reads {PersistedRecordFieldCount} " +
|
||||
$"(or {VariantRecordFieldCount} for a variant).");
|
||||
}
|
||||
|
||||
if (fieldCount == VariantRecordFieldCount)
|
||||
{
|
||||
record.Variant = ReadVariant(ref reader, record.Key, record.SchemaId);
|
||||
if (record.Payload.Length != 0)
|
||||
{
|
||||
throw new NotSupportedException(
|
||||
$"Record '{record.Key}' (schema '{record.SchemaId}') is a variant and must carry an empty payload.");
|
||||
}
|
||||
}
|
||||
|
||||
return record;
|
||||
|
|
|
|||
|
|
@ -135,9 +135,18 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
|
|||
Trace("publish");
|
||||
}
|
||||
|
||||
// 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)
|
||||
{
|
||||
if (entry.Variant != null)
|
||||
throw new NotSupportedException(
|
||||
$"Record '{entry.Key.Value}' is a variant of '{entry.Variant.BaseKey}'; the directory store does not hold document variants yet.");
|
||||
}
|
||||
|
||||
public override void Push(CultStoredDocument entry)
|
||||
{
|
||||
ThrowIfReadOnly();
|
||||
RefuseVariant(entry);
|
||||
Held(() =>
|
||||
{
|
||||
Entries[entry.Key.Value] = entry;
|
||||
|
|
@ -162,6 +171,8 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
|
|||
public override CultCommitOutcome CommitBatch(CultCommitRequest request, bool wait)
|
||||
{
|
||||
ThrowIfReadOnly();
|
||||
foreach (var entry in request.Upserts)
|
||||
RefuseVariant(entry);
|
||||
return Held(() =>
|
||||
{
|
||||
Directory.CreateDirectory(_manifestFile.DirectoryName!);
|
||||
|
|
|
|||
|
|
@ -213,9 +213,39 @@ The payload is the domain object only. It does not repeat the key, schema id,
|
|||
or stored timestamp.
|
||||
|
||||
Under `cultcache.store.v1` a record has exactly these four slots. A reader refuses any
|
||||
other `cultcache.store.*` header and any record with more than four slots, naming the
|
||||
version or the record's key and schema id. It never skips a slot it does not
|
||||
understand.
|
||||
`cultcache.store.*` header it does not know and any record with more slots than its
|
||||
header allows, naming the version or the record's key and schema id. It never skips a
|
||||
slot it does not understand.
|
||||
|
||||
## Variant Records
|
||||
|
||||
A variant record is a record whose document is a base record's document with some
|
||||
members replaced. It stores the difference, never the resolved document. It carries a
|
||||
fifth slot:
|
||||
|
||||
```
|
||||
[key, schemaId, storedAt, payload, [baseKey, overrides[]]]
|
||||
```
|
||||
|
||||
- `payload` is empty. An older reader that ignores the fifth slot reads an empty
|
||||
payload as a corrupt document and refuses, which is why the delta lives outside it.
|
||||
- `schemaId` is the type's schema id, the same as the base's. A variant has its base's
|
||||
concrete type, and lives in the same store (one home store per type).
|
||||
- An override is `[op, path[], id, value]`. `op` is `0` Set, `1` Insert or `2` Remove.
|
||||
A path step is `[slot, elementId]`: `slot` is the member's MessagePack `[Key]` and
|
||||
`elementId` is `""` unless the step enters an object-list element. `value` is the
|
||||
member's own MessagePack encoding, inline. A top-level Set has one step and `id` `""`;
|
||||
a reader that resolves only that shape refuses the others, naming the variant.
|
||||
- A store holding at least one variant declares `cultcache.store.v2`; a store holding
|
||||
none declares `cultcache.store.v1` and is byte-identical to what a runtime without
|
||||
variants writes. A `v1` header over a variant record is refused.
|
||||
- A variant's resolved document is derived in memory when the cache admits the record
|
||||
or any record on its base chain, and is never written back. A plain write at a
|
||||
variant's key is refused; `Flatten` replaces a variant with a plain record holding its
|
||||
resolved document.
|
||||
- The C# runtime resolves variants. Every other runtime refuses a `v2` store by name.
|
||||
CultMesh single-file reads, the directory store and CultNet refuse a variant record by
|
||||
name until each has its own cut.
|
||||
|
||||
## Canonical Semantic Schema Hash
|
||||
|
||||
|
|
|
|||
|
|
@ -105,3 +105,10 @@ lie about the payload:
|
|||
- no compatible local schema candidate
|
||||
|
||||
This is not negotiable. Better a loud refusal than quiet bit-rot.
|
||||
|
||||
## Variant overrides
|
||||
|
||||
A variant's override drifts exactly as the same member would on a plain record: an
|
||||
override naming a slot the type no longer has, or holding a value that no longer decodes
|
||||
into the member's type, is ignored with a warning naming the variant and the member, and
|
||||
the variant inherits the base value. There is no separate rule for variants.
|
||||
|
|
|
|||
|
|
@ -55,11 +55,17 @@ namespace GameCult.Caching
|
|||
public string SchemaId { get; set; } = string.Empty;
|
||||
public string StoredAt { get; set; } = string.Empty;
|
||||
public byte[] Payload { get; set; } = Array.Empty<byte>();
|
||||
|
||||
// Non-null for a variant record (slot 4): its base and overrides. Its payload is empty.
|
||||
public CultVariantDelta? Variant { get; set; }
|
||||
}
|
||||
|
||||
public sealed class CultPersistedStoreSnapshot
|
||||
{
|
||||
public string FormatVersion { get; set; } = "cultcache.store.v1";
|
||||
public const string FormatV1 = "cultcache.store.v1";
|
||||
public const string FormatV2 = "cultcache.store.v2";
|
||||
|
||||
public string FormatVersion { get; set; } = FormatV1;
|
||||
public CultSchemaCatalogEntry[] SchemaCatalog { get; set; } = Array.Empty<CultSchemaCatalogEntry>();
|
||||
public CultPersistedRecord[] Records { get; set; } = Array.Empty<CultPersistedRecord>();
|
||||
}
|
||||
|
|
@ -336,13 +342,43 @@ namespace GameCult.Caching
|
|||
Key = key;
|
||||
StoredAt = storedAt;
|
||||
Descriptor = descriptor;
|
||||
Document = document;
|
||||
_document = document;
|
||||
}
|
||||
|
||||
// A variant: the delta is what is stored; the cache resolves it against its base in the hold that admits it.
|
||||
internal CultStoredDocument(
|
||||
CultRecordKey key,
|
||||
string storedAt,
|
||||
CultDocumentDescriptor descriptor,
|
||||
CultVariantDelta variant,
|
||||
object? resolved)
|
||||
{
|
||||
Key = key;
|
||||
StoredAt = storedAt;
|
||||
Descriptor = descriptor;
|
||||
Variant = variant;
|
||||
_document = resolved;
|
||||
}
|
||||
|
||||
private readonly object? _document;
|
||||
|
||||
public CultRecordKey Key { get; }
|
||||
public string StoredAt { get; }
|
||||
public CultDocumentDescriptor Descriptor { get; }
|
||||
public object Document { get; }
|
||||
|
||||
// The complete document. For a variant, the resolved view: derived, never written back.
|
||||
public object Document => _document ??
|
||||
throw new InvalidOperationException($"Variant {Key.Value} has not been resolved; only the cache resolves a variant.");
|
||||
|
||||
// The base and overrides a variant record stores; null for a plain record.
|
||||
public CultVariantDelta? Variant { get; }
|
||||
|
||||
// A plain record staged by Flatten: the one way a plain record may replace a variant at its key.
|
||||
internal bool Flattens { get; set; }
|
||||
|
||||
internal bool IsResolved => _document != null;
|
||||
|
||||
internal CultStoredDocument Resolved(object document) => new(Key, StoredAt, Descriptor, Variant!, document);
|
||||
}
|
||||
|
||||
public sealed class CultDocumentRegistry
|
||||
|
|
@ -1662,11 +1698,29 @@ namespace GameCult.Caching
|
|||
private static string Identity(string key, string schemaId, string storedAt) => $"{key}\n{schemaId}\n{storedAt}";
|
||||
}
|
||||
|
||||
// One staged write: a plain record (Stored), a variant (Variant), a flatten, or, with none of those, a removal.
|
||||
internal sealed class StagedOperation
|
||||
{
|
||||
public StagedOperation(CultRecordKey key, CultStoredDocument? stored, CultVariantDelta? variant, bool flatten)
|
||||
{
|
||||
Key = key;
|
||||
Stored = stored;
|
||||
Variant = variant;
|
||||
Flatten = flatten;
|
||||
}
|
||||
|
||||
public CultRecordKey Key { get; }
|
||||
public CultStoredDocument? Stored { get; }
|
||||
public CultVariantDelta? Variant { get; }
|
||||
public bool Flatten { get; }
|
||||
public bool IsRemoval => Stored == null && Variant == null && !Flatten;
|
||||
}
|
||||
|
||||
// An explicit value: nothing staged here is visible to anyone until Commit returns.
|
||||
public sealed class CultCacheBatch
|
||||
{
|
||||
private readonly CultCache _cache;
|
||||
internal readonly Dictionary<string, (CultRecordKey Key, CultStoredDocument? Stored)> Operations = new(StringComparer.Ordinal);
|
||||
internal readonly Dictionary<string, StagedOperation> Operations = new(StringComparer.Ordinal);
|
||||
internal readonly List<(CultRecordKey Key, string? SchemaId, string? StoredAt)> Expected = new();
|
||||
internal bool ExpectsUnchanged;
|
||||
internal bool Sealed;
|
||||
|
|
@ -1691,7 +1745,24 @@ namespace GameCult.Caching
|
|||
public void Remove(CultRecordKey key)
|
||||
{
|
||||
ThrowIfSealed();
|
||||
Operations[key.Value] = (key, null);
|
||||
Operations[key.Value] = new StagedOperation(key, null, null, false);
|
||||
}
|
||||
|
||||
// Stages a variant record: its base and a sparse set of overrides, resolved by the cache when the batch commits.
|
||||
// A variant changes only through here; rebasing is this call with another base.
|
||||
public CultRecordKey UpsertVariant(CultRecordKey key, CultRecordKey baseKey, IEnumerable<CultVariantOverride>? overrides = null)
|
||||
{
|
||||
ThrowIfSealed();
|
||||
Operations[key.Value] = new StagedOperation(
|
||||
key, null, new CultVariantDelta(baseKey.Value, (overrides ?? Array.Empty<CultVariantOverride>()).ToArray()), false);
|
||||
return key;
|
||||
}
|
||||
|
||||
// Replaces a variant with a plain record holding its resolved document: the explicit, separate operation (Q1a).
|
||||
public void Flatten(CultRecordKey key)
|
||||
{
|
||||
ThrowIfSealed();
|
||||
Operations[key.Value] = new StagedOperation(key, null, null, true);
|
||||
}
|
||||
|
||||
public void Expect(CultRecordKey key, object? current)
|
||||
|
|
@ -1711,7 +1782,7 @@ namespace GameCult.Caching
|
|||
{
|
||||
ThrowIfSealed();
|
||||
var stored = _cache.CreateStoredDocument(document, key);
|
||||
Operations[stored.Key.Value] = (stored.Key, stored);
|
||||
Operations[stored.Key.Value] = new StagedOperation(stored.Key, stored, null, false);
|
||||
return stored.Key;
|
||||
}
|
||||
|
||||
|
|
@ -1730,6 +1801,11 @@ namespace GameCult.Caching
|
|||
private readonly Dictionary<Type, Dictionary<string, string>> _names = new();
|
||||
private readonly Dictionary<(Type Type, string Alias), Dictionary<string, string>> _indexes = new();
|
||||
private readonly Dictionary<Type, string> _globals = new();
|
||||
private readonly CultCodec? _codec;
|
||||
// Keys holding a variant, and the drift report of each variant whose overrides were partly ignored: both derived
|
||||
// from the entries by Apply, never decided anywhere else.
|
||||
private readonly HashSet<string> _variantKeys = new(StringComparer.Ordinal);
|
||||
private readonly Dictionary<string, CultSchemaMigrationReport> _variantReports = new(StringComparer.Ordinal);
|
||||
private readonly ConditionalWeakTable<object, KeyBox> _handles = new();
|
||||
private readonly Subject<Change> _changes = new();
|
||||
private readonly object _gate = new();
|
||||
|
|
@ -1738,13 +1814,42 @@ namespace GameCult.Caching
|
|||
[ThreadStatic] private static List<(Change Change, bool Loaded)>? _held;
|
||||
private bool _dirtyInMemory;
|
||||
|
||||
public CultCache(CultDocumentRegistry? registry = null)
|
||||
// The codec is the store's encoding (CultCacheMessagePack.Create supplies it). A cache without one holds no variants:
|
||||
// resolving a variant needs to clone documents and decode override values.
|
||||
public CultCache(CultDocumentRegistry? registry = null, CultCodec? codec = null)
|
||||
{
|
||||
_registry = registry ?? CultDocumentRegistry.Shared;
|
||||
_codec = codec;
|
||||
}
|
||||
|
||||
public CultDocumentRegistry Registry => _registry;
|
||||
|
||||
public CultCodec? Codec => _codec;
|
||||
|
||||
// Overrides a variant carries that no longer resolve (a slot the type dropped, a value that no longer decodes),
|
||||
// reported as compatible drift: the override is ignored and the variant inherits the base value.
|
||||
public IReadOnlyList<CultSchemaMigrationReport> VariantMigrationReports
|
||||
{
|
||||
get
|
||||
{
|
||||
lock (_gate)
|
||||
return _variantReports.Values.ToArray();
|
||||
}
|
||||
}
|
||||
|
||||
// The whole override of one member, encoded with this cache's codec, ready for UpsertVariant.
|
||||
public CultVariantOverride Override(Type documentType, string member, object? value)
|
||||
{
|
||||
if (_codec == null) throw new InvalidOperationException("This cache has no codec; create it with CultCacheMessagePack.Create.");
|
||||
var found = _registry.GetRequired(documentType).RichMembers.FirstOrDefault(entry => entry.Member.Name == member)
|
||||
?? throw new ArgumentException($"{documentType.Name} has no persisted member '{member}'.", nameof(member));
|
||||
return CultVariantOverride.Set(
|
||||
found.Slot,
|
||||
_codec.Serialize(value!, found.MemberType, documentType));
|
||||
}
|
||||
|
||||
public CultVariantOverride Override<T>(string member, object? value) where T : class => Override(typeof(T), member, value);
|
||||
|
||||
public bool IsDirty
|
||||
{
|
||||
get
|
||||
|
|
@ -1841,7 +1946,7 @@ namespace GameCult.Caching
|
|||
|
||||
store.AttachRegistry(_registry);
|
||||
store.Cache = this;
|
||||
store.Loaded = (loaded, dropped) => Admit(loaded, dropped, store, _ => CultCommitOutcome.Committed);
|
||||
store.Loaded = (loaded, dropped) => Admit(loaded, dropped, store, (_, _) => CultCommitOutcome.Committed);
|
||||
_stores.Add(candidate);
|
||||
try
|
||||
{
|
||||
|
|
@ -1915,11 +2020,39 @@ namespace GameCult.Caching
|
|||
return Task.FromResult(Write(document, key));
|
||||
}
|
||||
|
||||
// A single-call variant write; a batch stages the same thing with CultCacheBatch.UpsertVariant.
|
||||
public Task<CultRecordKey> UpsertVariantAsync(CultRecordKey key, CultRecordKey baseKey, IEnumerable<CultVariantOverride>? overrides = null)
|
||||
{
|
||||
var delta = new CultVariantDelta(baseKey.Value, (overrides ?? Array.Empty<CultVariantOverride>()).ToArray());
|
||||
return Task.FromResult(Held(() =>
|
||||
{
|
||||
var stored = StampVariant(key, DescriptorOfBase(key, delta.BaseKey, null), delta);
|
||||
Admit(new[] { stored }, Array.Empty<CultStoredDocument>(), null, (home, docs) =>
|
||||
{
|
||||
home?.Push(docs[0]);
|
||||
return CultCommitOutcome.Committed;
|
||||
});
|
||||
return key;
|
||||
}));
|
||||
}
|
||||
|
||||
// Replaces a variant with a plain record holding its resolved document (Q1a's explicit operation).
|
||||
public Task<CultRecordKey> FlattenAsync(CultRecordKey key) => Task.FromResult(Held(() =>
|
||||
{
|
||||
var stored = StampFlatten(key);
|
||||
Admit(new[] { stored }, Array.Empty<CultStoredDocument>(), null, (home, docs) =>
|
||||
{
|
||||
home?.Push(docs[0]);
|
||||
return CultCommitOutcome.Committed;
|
||||
});
|
||||
return key;
|
||||
}));
|
||||
|
||||
public bool Remove(CultRecordKey key) => Held(() =>
|
||||
{
|
||||
if (!_entries.TryGetValue(key.Value, out var existing))
|
||||
return false;
|
||||
Admit(Array.Empty<CultStoredDocument>(), new[] { existing }, null, home =>
|
||||
Admit(Array.Empty<CultStoredDocument>(), new[] { existing }, null, (home, _) =>
|
||||
{
|
||||
home?.Delete(existing);
|
||||
return CultCommitOutcome.Committed;
|
||||
|
|
@ -1950,6 +2083,13 @@ namespace GameCult.Caching
|
|||
return _handles.TryGetValue(document, out var box) ? new CultRecordHandle<T>(box.Key) : null;
|
||||
}
|
||||
|
||||
// The stored form of a record: for a variant, the delta beside the resolved document.
|
||||
public CultStoredDocument? GetStored(CultRecordKey key)
|
||||
{
|
||||
lock (_gate)
|
||||
return _entries.TryGetValue(key.Value, out var stored) ? stored : null;
|
||||
}
|
||||
|
||||
public object? Get(CultRecordKey key)
|
||||
{
|
||||
lock (_gate)
|
||||
|
|
@ -2039,7 +2179,53 @@ namespace GameCult.Caching
|
|||
|
||||
// Call under the gate that admits the result, so no load can land a later storedAt in between.
|
||||
private CultStoredDocument Stamp(CultRecordKey key, CultDocumentDescriptor descriptor, object document) =>
|
||||
new(key, MintStoredAt(_entries.TryGetValue(key.Value, out var previous) ? previous.StoredAt : null), descriptor, document);
|
||||
new(key, MintStoredAt(PreviousStoredAt(key)), descriptor, document);
|
||||
|
||||
// An unresolved variant: the admitting hold resolves it against its base before anything lands.
|
||||
private CultStoredDocument StampVariant(CultRecordKey key, CultDocumentDescriptor descriptor, CultVariantDelta delta) =>
|
||||
new(key, MintStoredAt(PreviousStoredAt(key)), descriptor, delta, null);
|
||||
|
||||
// The plain record a Flatten stages: the variant's resolved document, under the variant's own key and schema.
|
||||
private CultStoredDocument StampFlatten(CultRecordKey key)
|
||||
{
|
||||
if (!_entries.TryGetValue(key.Value, out var existing) || existing.Variant == null)
|
||||
throw new InvalidOperationException($"Flatten({key.Value}): the record is not a variant.");
|
||||
return new CultStoredDocument(key, MintStoredAt(existing.StoredAt), existing.Descriptor, existing.Document) { Flattens = true };
|
||||
}
|
||||
|
||||
private string? PreviousStoredAt(CultRecordKey key) => _entries.TryGetValue(key.Value, out var previous) ? previous.StoredAt : null;
|
||||
|
||||
// The concrete type a variant takes from its base, walking bases through what the batch stages and then the cache.
|
||||
// Refuses a cycle or a base that is not there, naming the keys, exactly as the admitting hold would.
|
||||
private CultDocumentDescriptor DescriptorOfBase(CultRecordKey variant, string baseKey, CultCacheBatch? batch)
|
||||
{
|
||||
var seen = new List<string> { variant.Value };
|
||||
var next = baseKey;
|
||||
while (true)
|
||||
{
|
||||
if (seen.Contains(next))
|
||||
throw CycleRefusal(seen.Append(next));
|
||||
var staged = batch != null && batch.Operations.TryGetValue(next, out var op) ? op : null;
|
||||
if (staged != null)
|
||||
{
|
||||
if (staged.Stored != null)
|
||||
return staged.Stored.Descriptor;
|
||||
if (staged.Variant != null)
|
||||
{
|
||||
seen.Add(next);
|
||||
next = staged.Variant.BaseKey;
|
||||
continue;
|
||||
}
|
||||
|
||||
if (!staged.Flatten)
|
||||
throw MissingBaseRefusal(variant.Value, baseKey, removed: true);
|
||||
}
|
||||
|
||||
if (!_entries.TryGetValue(next, out var existing))
|
||||
throw MissingBaseRefusal(variant.Value, baseKey, removed: false);
|
||||
return existing.Descriptor;
|
||||
}
|
||||
}
|
||||
|
||||
internal CultStoredDocument? Observe(CultRecordKey key, object? current)
|
||||
{
|
||||
|
|
@ -2068,9 +2254,9 @@ namespace GameCult.Caching
|
|||
private CultRecordKey Write(object document, CultRecordKey? key) => Held(() =>
|
||||
{
|
||||
var stored = CreateStoredDocument(document, key);
|
||||
Admit(new[] { stored }, Array.Empty<CultStoredDocument>(), null, home =>
|
||||
Admit(new[] { stored }, Array.Empty<CultStoredDocument>(), null, (home, docs) =>
|
||||
{
|
||||
home?.Push(stored);
|
||||
home?.Push(docs[0]);
|
||||
return CultCommitOutcome.Committed;
|
||||
});
|
||||
return stored.Key;
|
||||
|
|
@ -2094,16 +2280,19 @@ namespace GameCult.Caching
|
|||
{
|
||||
// Staging stamped outside this gate; a pull since then may hold a later storedAt, so stamp again.
|
||||
var upserts = batch.Operations.Values
|
||||
.Where(op => op.Stored != null)
|
||||
.Select(op => Stamp(op.Key, op.Stored!.Descriptor, op.Stored.Document))
|
||||
.Where(op => !op.IsRemoval)
|
||||
.Select(op => op.Stored != null ? Stamp(op.Key, op.Stored.Descriptor, op.Stored.Document)
|
||||
: op.Variant != null ? StampVariant(op.Key, DescriptorOfBase(op.Key, op.Variant.BaseKey, batch), op.Variant)
|
||||
: StampFlatten(op.Key))
|
||||
.ToArray();
|
||||
var deletes = batch.Operations.Values
|
||||
.Where(op => op.Stored == null && _entries.ContainsKey(op.Key.Value))
|
||||
.Where(op => op.IsRemoval && _entries.ContainsKey(op.Key.Value))
|
||||
.Select(op => _entries[op.Key.Value])
|
||||
.ToArray();
|
||||
var request = new CultCommitRequest(upserts, deletes, batch.Expected.ToArray(), batch.ExpectsUnchanged);
|
||||
return Admit(upserts, deletes, null, home =>
|
||||
return Admit(upserts, deletes, null, (home, resolved) =>
|
||||
{
|
||||
// What the store holds is what the hold resolved: a variant's delta beside its resolved view.
|
||||
var request = new CultCommitRequest(resolved, deletes, batch.Expected.ToArray(), batch.ExpectsUnchanged);
|
||||
if (home != null)
|
||||
{
|
||||
if (request.HasConditions && home.IsDirty)
|
||||
|
|
@ -2127,14 +2316,18 @@ namespace GameCult.Caching
|
|||
IReadOnlyList<CultStoredDocument> admitted,
|
||||
IReadOnlyList<CultStoredDocument> evicted,
|
||||
CacheBackingStore? source,
|
||||
Func<CacheBackingStore?, CultCommitOutcome> land) => Held(() =>
|
||||
Func<CacheBackingStore?, IReadOnlyList<CultStoredDocument>, CultCommitOutcome> land) => Held(() =>
|
||||
{
|
||||
if (_held == null)
|
||||
throw new InvalidOperationException("An admission reached the cache under a plain lock on its gate; every admission runs in a hold.");
|
||||
var outcome = land(Validate(admitted, evicted, source));
|
||||
var home = Validate(admitted, evicted, source);
|
||||
// The one resolution: a variant admitted, a base admitted or evicted, all resolved here against the post-batch
|
||||
// set, before land and before memory. land receives the admitted records with variants resolved.
|
||||
var plan = Resolve(admitted, evicted, source);
|
||||
var outcome = land(home, plan.Admitted);
|
||||
if (outcome != CultCommitOutcome.Committed)
|
||||
return outcome;
|
||||
foreach (var change in Apply(admitted, evicted, source))
|
||||
foreach (var change in Apply(plan, evicted, source))
|
||||
_held.Add((change, source != null));
|
||||
return CultCommitOutcome.Committed;
|
||||
});
|
||||
|
|
@ -2262,12 +2455,215 @@ namespace GameCult.Caching
|
|||
}
|
||||
}
|
||||
|
||||
private List<Change> Apply(
|
||||
private sealed class VariantPlan
|
||||
{
|
||||
public VariantPlan(IReadOnlyList<CultStoredDocument> admitted)
|
||||
{
|
||||
Admitted = admitted;
|
||||
}
|
||||
|
||||
// The admitted records, variants resolved.
|
||||
public IReadOnlyList<CultStoredDocument> Admitted { get; }
|
||||
|
||||
// Variants this admission did not write whose base chain it changed: re-resolved, same delta, same storedAt.
|
||||
public List<CultStoredDocument> Dependents { get; } = new();
|
||||
|
||||
// The drift report of every variant resolved here; null when nothing was ignored.
|
||||
public Dictionary<string, CultSchemaMigrationReport?> Reports { get; } = new(StringComparer.Ordinal);
|
||||
}
|
||||
|
||||
// Resolves every admitted variant and every variant whose base chain this admission changes, base first, against
|
||||
// the post-admission set. It refuses a broken variant naming the keys, and mutates nothing.
|
||||
private VariantPlan Resolve(
|
||||
IReadOnlyList<CultStoredDocument> admitted,
|
||||
IReadOnlyList<CultStoredDocument> evicted,
|
||||
CacheBackingStore? source)
|
||||
{
|
||||
var changes = new List<Change>(admitted.Count + evicted.Count);
|
||||
if (_variantKeys.Count == 0 && admitted.All(stored => stored.Variant == null))
|
||||
return new VariantPlan(admitted);
|
||||
|
||||
var incoming = admitted.ToDictionary(stored => stored.Key.Value, StringComparer.Ordinal);
|
||||
var gone = evicted.Select(stored => stored.Key.Value).ToHashSet(StringComparer.Ordinal);
|
||||
CultStoredDocument? Lookup(string key) =>
|
||||
incoming.TryGetValue(key, out var found) ? found
|
||||
: gone.Contains(key) ? null
|
||||
: _entries.TryGetValue(key, out var held) ? held : null;
|
||||
|
||||
// A variant is written only through the variant API: a plain record never lands on a variant's key, and the
|
||||
// reverse is a rebase or a flatten, never an accident. A load reports what is on disk and is not a write.
|
||||
if (source == null)
|
||||
{
|
||||
foreach (var stored in admitted)
|
||||
{
|
||||
_entries.TryGetValue(stored.Key.Value, out var present);
|
||||
if (stored.Variant == null && !stored.Flattens && present?.Variant != null)
|
||||
throw new InvalidOperationException(
|
||||
$"Record {stored.Key.Value} is a variant of {present.Variant.BaseKey}; a plain write cannot replace it. " +
|
||||
"Change it through the variant API, or Flatten it.");
|
||||
if (stored.Variant != null && present != null && present.Variant == null)
|
||||
throw new InvalidOperationException(
|
||||
$"Record {stored.Key.Value} is a plain record; it cannot become a variant of {stored.Variant.BaseKey} in place. Use a new key.");
|
||||
if (stored.Variant != null && present?.Variant != null &&
|
||||
present.Descriptor.DocumentType != stored.Descriptor.DocumentType)
|
||||
throw new InvalidOperationException(
|
||||
$"Variant {stored.Key.Value} is a {present.Descriptor.DocumentType.Name}; " +
|
||||
$"rebasing it onto {stored.Variant.BaseKey} would make it a {stored.Descriptor.DocumentType.Name}.");
|
||||
}
|
||||
}
|
||||
|
||||
var changed = new HashSet<string>(incoming.Keys.Concat(gone), StringComparer.Ordinal);
|
||||
var toResolve = new List<string>(admitted.Where(stored => stored.Variant != null).Select(stored => stored.Key.Value));
|
||||
foreach (var variantKey in _variantKeys)
|
||||
{
|
||||
if (incoming.ContainsKey(variantKey) || gone.Contains(variantKey))
|
||||
continue;
|
||||
var next = _entries[variantKey].Variant!.BaseKey;
|
||||
for (var steps = 0; steps <= _entries.Count; steps++)
|
||||
{
|
||||
if (changed.Contains(next))
|
||||
{
|
||||
toResolve.Add(variantKey);
|
||||
break;
|
||||
}
|
||||
|
||||
var link = Lookup(next);
|
||||
if (link?.Variant == null)
|
||||
break;
|
||||
next = link.Variant.BaseKey;
|
||||
}
|
||||
}
|
||||
|
||||
if (toResolve.Count == 0)
|
||||
return new VariantPlan(admitted);
|
||||
|
||||
var pending = new HashSet<string>(toResolve, StringComparer.Ordinal);
|
||||
var done = new Dictionary<string, CultStoredDocument>(StringComparer.Ordinal);
|
||||
var reports = new Dictionary<string, CultSchemaMigrationReport?>(StringComparer.Ordinal);
|
||||
var stack = new List<string>();
|
||||
foreach (var key in toResolve)
|
||||
ResolveOne(key);
|
||||
|
||||
var plan = new VariantPlan(admitted.Select(stored => stored.Variant != null ? done[stored.Key.Value] : stored).ToList());
|
||||
foreach (var key in toResolve.Where(key => !incoming.ContainsKey(key)))
|
||||
plan.Dependents.Add(done[key]);
|
||||
foreach (var pair in reports)
|
||||
plan.Reports[pair.Key] = pair.Value;
|
||||
return plan;
|
||||
|
||||
CultStoredDocument ResolveOne(string key)
|
||||
{
|
||||
if (done.TryGetValue(key, out var finished))
|
||||
return finished;
|
||||
var variant = Lookup(key)!;
|
||||
if (stack.Contains(key))
|
||||
throw CycleRefusal(stack.SkipWhile(entry => entry != key).Append(key));
|
||||
stack.Add(key);
|
||||
|
||||
var delta = variant.Variant!;
|
||||
var baseStored = Lookup(delta.BaseKey) ?? throw MissingBaseRefusal(key, delta.BaseKey, gone.Contains(delta.BaseKey));
|
||||
if (baseStored.Variant != null && pending.Contains(delta.BaseKey))
|
||||
baseStored = ResolveOne(delta.BaseKey);
|
||||
|
||||
var descriptor = variant.Descriptor;
|
||||
if (_codec == null)
|
||||
throw new InvalidOperationException(
|
||||
$"Variant {key} needs a codec to resolve; this cache has none. Create it with CultCacheMessagePack.Create.");
|
||||
if (baseStored.Descriptor.DocumentType != descriptor.DocumentType)
|
||||
throw new InvalidOperationException(
|
||||
$"Variant {key} is a {descriptor.DocumentType.Name} but its base {delta.BaseKey} is a {baseStored.Descriptor.DocumentType.Name}; " +
|
||||
"a variant has its base's concrete type.");
|
||||
|
||||
var type = descriptor.DocumentType;
|
||||
var kept = new List<KeyValuePair<int, byte[]>>();
|
||||
var ignored = new List<int>();
|
||||
var warnings = new List<CultSchemaMigrationWarning>();
|
||||
foreach (var entry in delta.Overrides)
|
||||
{
|
||||
if (entry.Op != CultOverrideOp.Set || entry.Path.Count != 1 || entry.Path[0].ElementId.Length != 0 || entry.Id.Length != 0)
|
||||
throw new InvalidOperationException(
|
||||
$"Variant {key} carries {entry.Op} at {entry.PathText}; this runtime resolves top-level Set overrides only.");
|
||||
var slot = entry.Path[0].Slot;
|
||||
var member = descriptor.RichMembers.FirstOrDefault(candidate => candidate.Slot == slot);
|
||||
if (member == null)
|
||||
{
|
||||
var name = delta.SlotNames != null && delta.SlotNames.TryGetValue(slot, out var persistedName)
|
||||
? $"'{persistedName}'" : "unnamed";
|
||||
ignored.Add(slot);
|
||||
warnings.Add(new CultSchemaMigrationWarning
|
||||
{
|
||||
Code = "ignored_extra_slot",
|
||||
Message = $"Variant {key} overrides slot {slot} ({name}), which {type.Name} no longer has; the override is ignored and the variant inherits the base value."
|
||||
});
|
||||
continue;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
_codec.Deserialize(member.MemberType, type, entry.Value);
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
ignored.Add(slot);
|
||||
warnings.Add(new CultSchemaMigrationWarning
|
||||
{
|
||||
Code = "ignored_undecodable_override",
|
||||
Message = $"Variant {key} overrides slot {slot} ('{member.Member.Name}') with a value that no longer decodes as {member.MemberType.Name} " +
|
||||
$"({exception.GetBaseException().Message}); the override is ignored and the variant inherits the base value."
|
||||
});
|
||||
continue;
|
||||
}
|
||||
|
||||
kept.Add(new KeyValuePair<int, byte[]>(slot, entry.Value));
|
||||
}
|
||||
|
||||
var payload = _codec.Serialize(baseStored.Document, type, type);
|
||||
var document = _codec.Deserialize(type, type, kept.Count == 0 ? payload : _codec.Overlay(payload, kept))
|
||||
?? throw new InvalidOperationException($"Variant {key} resolved to nothing.");
|
||||
|
||||
// R6: the name index maps one name to one key, so a variant that kept its base's name would shadow it.
|
||||
if (descriptor.NameAccessor?.Invoke(document) is { Length: > 0 } variantName)
|
||||
{
|
||||
for (var ancestor = baseStored; ancestor != null;
|
||||
ancestor = ancestor.Variant == null ? null : done.TryGetValue(ancestor.Variant.BaseKey, out var above) ? above : Lookup(ancestor.Variant.BaseKey))
|
||||
{
|
||||
if (descriptor.NameAccessor(ancestor.Document) == variantName)
|
||||
throw new InvalidOperationException(
|
||||
$"Variant {key} and its base {ancestor.Key.Value} are both named '{variantName}'; a variant must override its name.");
|
||||
}
|
||||
}
|
||||
|
||||
reports[key] = ignored.Count == 0
|
||||
? null
|
||||
: new CultSchemaMigrationReport
|
||||
{
|
||||
PersistedSchemaId = descriptor.SchemaId,
|
||||
LocalSchemaId = descriptor.SchemaId,
|
||||
PersistedSchemaName = descriptor.SchemaName,
|
||||
LocalSchemaName = descriptor.SchemaName,
|
||||
Kind = CultSchemaMigrationKind.CompatibleDrift,
|
||||
IgnoredExtraSlots = ignored.OrderBy(slot => slot).ToArray(),
|
||||
Warnings = warnings.ToArray()
|
||||
};
|
||||
stack.RemoveAt(stack.Count - 1);
|
||||
return done[key] = variant.Resolved(document);
|
||||
}
|
||||
}
|
||||
|
||||
private static InvalidOperationException CycleRefusal(IEnumerable<string> keys) =>
|
||||
new($"Variant bases form a cycle: {string.Join(" -> ", keys)}.");
|
||||
|
||||
private static InvalidOperationException MissingBaseRefusal(string variant, string baseKey, bool removed) =>
|
||||
new(removed
|
||||
? $"Record {baseKey} cannot go: variant {variant} is based on it. Remove or rebase the variant in the same commit."
|
||||
: $"Variant {variant} is based on {baseKey}, which does not exist.");
|
||||
|
||||
private List<Change> Apply(
|
||||
VariantPlan plan,
|
||||
IReadOnlyList<CultStoredDocument> evicted,
|
||||
CacheBackingStore? source)
|
||||
{
|
||||
var admitted = plan.Admitted;
|
||||
var changes = new List<Change>(admitted.Count + evicted.Count + plan.Dependents.Count);
|
||||
foreach (var stored in evicted)
|
||||
{
|
||||
if (!_entries.TryGetValue(stored.Key.Value, out var existing))
|
||||
|
|
@ -2275,10 +2671,12 @@ namespace GameCult.Caching
|
|||
Unindex(existing);
|
||||
_entries.Remove(stored.Key.Value);
|
||||
_handles.Remove(existing.Document);
|
||||
_variantKeys.Remove(stored.Key.Value);
|
||||
_variantReports.Remove(stored.Key.Value);
|
||||
changes.Add(new Change(CultCacheDocumentChangeKind.Removed, existing, null, existing.Document, ++_sequence));
|
||||
}
|
||||
|
||||
foreach (var stored in admitted)
|
||||
foreach (var stored in admitted.Concat(plan.Dependents))
|
||||
{
|
||||
_entries.TryGetValue(stored.Key.Value, out var previous);
|
||||
if (previous != null)
|
||||
|
|
@ -2290,6 +2688,14 @@ namespace GameCult.Caching
|
|||
_entries[stored.Key.Value] = stored;
|
||||
_handles.AddOrUpdate(stored.Document, new KeyBox(stored.Key));
|
||||
Index(stored);
|
||||
if (stored.Variant == null)
|
||||
_variantKeys.Remove(stored.Key.Value);
|
||||
else
|
||||
_variantKeys.Add(stored.Key.Value);
|
||||
if (plan.Reports.TryGetValue(stored.Key.Value, out var report) && report != null)
|
||||
_variantReports[stored.Key.Value] = report;
|
||||
else
|
||||
_variantReports.Remove(stored.Key.Value);
|
||||
changes.Add(new Change(
|
||||
previous == null ? CultCacheDocumentChangeKind.Added : CultCacheDocumentChangeKind.Updated,
|
||||
stored,
|
||||
|
|
@ -2486,12 +2892,14 @@ namespace GameCult.Caching
|
|||
|
||||
protected CultPersistedRecord ToPersistedRecord(CultStoredDocument entry, Func<object, byte[]> serializePayload)
|
||||
{
|
||||
// A variant persists its delta and an empty payload; the resolved view is never written.
|
||||
return new CultPersistedRecord
|
||||
{
|
||||
Key = entry.Key.Value,
|
||||
SchemaId = entry.Descriptor.SchemaId,
|
||||
StoredAt = entry.StoredAt,
|
||||
Payload = serializePayload(entry.Document)
|
||||
Payload = entry.Variant != null ? Array.Empty<byte>() : serializePayload(entry.Document),
|
||||
Variant = entry.Variant == null ? null : new CultVariantDelta(entry.Variant.BaseKey, entry.Variant.Overrides)
|
||||
};
|
||||
}
|
||||
|
||||
|
|
@ -2501,6 +2909,21 @@ namespace GameCult.Caching
|
|||
Func<Type, byte[], object> deserializePayload)
|
||||
{
|
||||
var resolution = Registry.ResolvePersistedSchemaDetailed(record.SchemaId, catalog);
|
||||
if (record.Variant != null)
|
||||
{
|
||||
// The cache resolves it against its base when it admits the load; the catalog names the slots the type may have dropped.
|
||||
var persisted = catalog.FirstOrDefault(entry => string.Equals(entry.SchemaId, record.SchemaId, StringComparison.Ordinal));
|
||||
return new CultStoredDocument(
|
||||
new CultRecordKey(record.Key),
|
||||
record.StoredAt,
|
||||
resolution.Descriptor,
|
||||
new CultVariantDelta(record.Variant.BaseKey, record.Variant.Overrides)
|
||||
{
|
||||
SlotNames = persisted?.Members.ToDictionary(member => member.Slot, member => member.MemberName)
|
||||
},
|
||||
null);
|
||||
}
|
||||
|
||||
var document = deserializePayload(resolution.Descriptor.DocumentType, record.Payload);
|
||||
return new CultStoredDocument(
|
||||
new CultRecordKey(record.Key),
|
||||
|
|
@ -2675,6 +3098,10 @@ namespace GameCult.Caching
|
|||
var used = new HashSet<string>(ordered.Select(record => record.SchemaId), StringComparer.Ordinal);
|
||||
var snapshot = new CultPersistedStoreSnapshot
|
||||
{
|
||||
// v2 exactly when the store holds a variant: a store without one stays byte-identical v1.
|
||||
FormatVersion = ordered.Any(record => record.Variant != null)
|
||||
? CultPersistedStoreSnapshot.FormatV2
|
||||
: CultPersistedStoreSnapshot.FormatV1,
|
||||
SchemaCatalog = catalog
|
||||
.GroupBy(entry => entry.SchemaId, StringComparer.Ordinal)
|
||||
.Select(group => group.First())
|
||||
|
|
|
|||
|
|
@ -292,22 +292,24 @@ namespace GameCult.Caching
|
|||
typeof(int), typeof(long), typeof(uint), typeof(short), typeof(ushort), typeof(byte), typeof(sbyte)
|
||||
};
|
||||
|
||||
private readonly Func<object, Type, Type, byte[]> _serialize;
|
||||
private readonly Func<Type, byte[], object> _deserialize;
|
||||
private readonly CultCodec _codec;
|
||||
private readonly Dictionary<Type, CultInspectorShape> _shapes = new Dictionary<Type, CultInspectorShape>();
|
||||
private readonly Dictionary<MemberInfo, CultInspectorMetadata> _metadata = new Dictionary<MemberInfo, CultInspectorMetadata>();
|
||||
|
||||
// serialize (value, value type, owning document type) and deserialize are the store's codec
|
||||
// (CultCacheMessagePack.CreateInspectorModel for .cc stores). Edits clone documents through both; dictionary keys
|
||||
// compare by serialize under their owning document, exactly as the store writes them inside it, so serialize must
|
||||
// take any value, not only documents.
|
||||
public CultInspectorModel(CultDocumentRegistry registry, Func<object, Type, Type, byte[]> serialize, Func<Type, byte[], object> deserialize)
|
||||
// The codec is the store's (CultCacheMessagePack.CreateInspectorModel for .cc stores), the one a CultCache resolves
|
||||
// variants with. Edits clone documents through it; dictionary keys compare by its Serialize under their owning
|
||||
// document, exactly as the store writes them inside it.
|
||||
public CultInspectorModel(CultDocumentRegistry registry, CultCodec codec)
|
||||
{
|
||||
Registry = registry ?? throw new ArgumentNullException(nameof(registry));
|
||||
_serialize = serialize ?? throw new ArgumentNullException(nameof(serialize));
|
||||
_deserialize = deserialize ?? throw new ArgumentNullException(nameof(deserialize));
|
||||
_codec = codec ?? throw new ArgumentNullException(nameof(codec));
|
||||
}
|
||||
|
||||
private byte[] _serialize(object value, Type type, Type document) => _codec.Serialize(value, type, document);
|
||||
|
||||
private object _deserialize(Type type, byte[] bytes) =>
|
||||
_codec.Deserialize(type, type, bytes) ?? throw new InvalidOperationException($"The codec returned null for {type.FullName}.");
|
||||
|
||||
public CultDocumentRegistry Registry { get; }
|
||||
|
||||
public static string RecordLabel(CultStoredDocument record)
|
||||
|
|
|
|||
95
src/GameCult.Caching/CultVariants.cs
Normal file
95
src/GameCult.Caching/CultVariants.cs
Normal file
|
|
@ -0,0 +1,95 @@
|
|||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Linq;
|
||||
|
||||
namespace GameCult.Caching
|
||||
{
|
||||
// The store's encoding, as the cache needs it: the one pair CultInspectorModel and variant resolution share.
|
||||
// Serialize takes any value, not only documents, under its owning document type's options; Deserialize is its
|
||||
// inverse (a document type is checked against the registry, a member type is not); Overlay replaces slots of a
|
||||
// serialized document payload with already-encoded member values and touches nothing else.
|
||||
public sealed class CultCodec
|
||||
{
|
||||
public CultCodec(
|
||||
Func<object, Type, Type, byte[]> serialize,
|
||||
Func<Type, Type, byte[], object?> deserialize,
|
||||
Func<byte[], IReadOnlyList<KeyValuePair<int, byte[]>>, byte[]> overlay)
|
||||
{
|
||||
Serialize = serialize ?? throw new ArgumentNullException(nameof(serialize));
|
||||
Deserialize = deserialize ?? throw new ArgumentNullException(nameof(deserialize));
|
||||
Overlay = overlay ?? throw new ArgumentNullException(nameof(overlay));
|
||||
}
|
||||
|
||||
// (value, value type, owning document type)
|
||||
public Func<object, Type, Type, byte[]> Serialize { get; }
|
||||
|
||||
// (value type, owning document type, bytes)
|
||||
public Func<Type, Type, byte[], object?> Deserialize { get; }
|
||||
|
||||
// (document payload, slot and encoded value pairs)
|
||||
public Func<byte[], IReadOnlyList<KeyValuePair<int, byte[]>>, byte[]> Overlay { get; }
|
||||
}
|
||||
|
||||
public enum CultOverrideOp
|
||||
{
|
||||
Set = 0,
|
||||
Insert = 1,
|
||||
Remove = 2
|
||||
}
|
||||
|
||||
// One step of a member path: the MessagePack slot, and the id of the object-list element the step enters ("" when it
|
||||
// does not enter one).
|
||||
public sealed class CultPathStep
|
||||
{
|
||||
public CultPathStep(int slot, string elementId = "")
|
||||
{
|
||||
Slot = slot;
|
||||
ElementId = elementId ?? string.Empty;
|
||||
}
|
||||
|
||||
public int Slot { get; }
|
||||
public string ElementId { get; }
|
||||
|
||||
public override string ToString() => ElementId.Length == 0 ? $"[{Slot}]" : $"[{Slot}:{ElementId}]";
|
||||
}
|
||||
|
||||
// The wire shape [op, path[], id, value]; value is the override's already-encoded MessagePack value.
|
||||
public sealed class CultVariantOverride
|
||||
{
|
||||
public CultVariantOverride(CultOverrideOp op, IReadOnlyList<CultPathStep> path, string id, byte[] value)
|
||||
{
|
||||
Op = op;
|
||||
Path = path ?? throw new ArgumentNullException(nameof(path));
|
||||
Id = id ?? string.Empty;
|
||||
Value = value ?? throw new ArgumentNullException(nameof(value));
|
||||
}
|
||||
|
||||
public CultOverrideOp Op { get; }
|
||||
public IReadOnlyList<CultPathStep> Path { get; }
|
||||
public string Id { get; }
|
||||
public byte[] Value { get; }
|
||||
|
||||
// A top-level Set: the whole member at slot takes the encoded value.
|
||||
public static CultVariantOverride Set(int slot, byte[] value) =>
|
||||
new(CultOverrideOp.Set, new[] { new CultPathStep(slot) }, string.Empty, value);
|
||||
|
||||
internal string PathText => string.Concat(Path.Select(step => step.ToString()));
|
||||
}
|
||||
|
||||
// What a variant record stores: its base and its overrides. Never the resolved document.
|
||||
public sealed class CultVariantDelta
|
||||
{
|
||||
public CultVariantDelta(string baseKey, IReadOnlyList<CultVariantOverride> overrides)
|
||||
{
|
||||
BaseKey = baseKey ?? throw new ArgumentNullException(nameof(baseKey));
|
||||
Overrides = overrides ?? throw new ArgumentNullException(nameof(overrides));
|
||||
}
|
||||
|
||||
public string BaseKey { get; }
|
||||
public IReadOnlyList<CultVariantOverride> Overrides { get; }
|
||||
|
||||
// Member names by slot from the catalog the delta was loaded under, for warnings about slots the type dropped.
|
||||
// Not persisted.
|
||||
internal IReadOnlyDictionary<int, string>? SlotNames { get; set; }
|
||||
}
|
||||
}
|
||||
|
|
@ -156,6 +156,8 @@ namespace GameCult.Mesh
|
|||
$"CultCache document '{path}' record '{key.Value}' has schema '{record.SchemaId}', expected '{expectedSchemaId}'.");
|
||||
}
|
||||
|
||||
RefuseVariant(path, record);
|
||||
|
||||
if (!PublishesSchema(snapshot.SchemaCatalog, expectedSchemaId))
|
||||
{
|
||||
throw new InvalidDataException(
|
||||
|
|
@ -189,6 +191,8 @@ namespace GameCult.Mesh
|
|||
$"CultCache document '{path}' has schema '{record.SchemaId}', expected '{expectedSchemaId}'.");
|
||||
}
|
||||
|
||||
RefuseVariant(path, record);
|
||||
|
||||
if (!PublishesSchema(snapshot.SchemaCatalog, expectedSchemaId))
|
||||
{
|
||||
throw new InvalidDataException(
|
||||
|
|
@ -198,6 +202,15 @@ namespace GameCult.Mesh
|
|||
return record.Payload;
|
||||
}
|
||||
|
||||
// A variant record's payload is empty; its document exists only as the cache resolves it against its base.
|
||||
private static void RefuseVariant(string path, CultPersistedRecord record)
|
||||
{
|
||||
if (record.Variant != null)
|
||||
throw new NotSupportedException(
|
||||
$"CultCache document '{path}' record '{record.Key}' is a variant of '{record.Variant.BaseKey}'; " +
|
||||
"CultMesh does not resolve document variants, open the store with a CultCache.");
|
||||
}
|
||||
|
||||
private static void WriteSingleFileDocumentPayload(
|
||||
string path,
|
||||
CultRecordKey key,
|
||||
|
|
|
|||
|
|
@ -393,7 +393,7 @@ namespace GameCult.Networking
|
|||
stored = stored.Where(entry => rowFilter(entry.Descriptor, entry.Key));
|
||||
var rows = stored
|
||||
.Select(entry => new CultNetSelectionEvaluator.Row(
|
||||
entry.Descriptor, entry.Key, entry.Document, ordinalOf(entry.Descriptor.SchemaId, entry.Key), entry.StoredAt))
|
||||
entry.Descriptor, entry.Key, RefuseVariant(entry).Document, ordinalOf(entry.Descriptor.SchemaId, entry.Key), entry.StoredAt))
|
||||
.ToArray();
|
||||
var full = CultNetSelectionEvaluator.EvaluateAll(_documents, rows, selection);
|
||||
// R-Q: a page's asOf is exact only when every matched row is committed through one shard's
|
||||
|
|
@ -487,7 +487,7 @@ namespace GameCult.Networking
|
|||
stored = stored.Where(entry => rowFilter(entry.Descriptor, entry.Key));
|
||||
var rows = stored
|
||||
.Select(entry => new CultNetSelectionEvaluator.Row(
|
||||
entry.Descriptor, entry.Key, entry.Document, ordinalOf(entry.Descriptor.SchemaId, entry.Key), entry.StoredAt))
|
||||
entry.Descriptor, entry.Key, RefuseVariant(entry).Document, ordinalOf(entry.Descriptor.SchemaId, entry.Key), entry.StoredAt))
|
||||
.ToArray();
|
||||
|
||||
var full = CultNetSelectionEvaluator.EvaluateAll(_documents, rows, selection);
|
||||
|
|
@ -549,6 +549,13 @@ namespace GameCult.Networking
|
|||
};
|
||||
}
|
||||
|
||||
// CultNet carries no variant deltas yet, and a resolved view sent as a plain record would be a lie about the store.
|
||||
private static CultStoredDocument RefuseVariant(CultStoredDocument entry) =>
|
||||
entry.Variant == null
|
||||
? entry
|
||||
: throw new NotSupportedException(
|
||||
$"Record '{entry.Key.Value}' is a variant of '{entry.Variant.BaseKey}'; CultNet does not carry document variants yet.");
|
||||
|
||||
/// <summary>
|
||||
/// Builds the wire record for one evaluated row. The shared path under snapshot and change
|
||||
/// (docs/cultnet-selection-cut.md, section 8's "shared paths").
|
||||
|
|
|
|||
Loading…
Reference in a new issue