Merge variants C1: the C# core resolves document variants
A variant record (v2 store, slot 4 [baseKey, overrides], empty payload) holds top-level Set overrides and is resolved in the admitting hold. Readers see complete documents. A base edit re-resolves dependents in the same commit. Rulings honoured: Q1 a plain Upsert at a variant key is refused; Q4 an override follows the plain-record outcome (a dropped slot soft-drifts and is shed, an undecodable value refuses); Q5 the v2 header only with variants; Q6 the directory store and CultNet refuse variants (after selection, typed, subscriptions withdrawn). Self's defaults: invalid overrides are refused at write; R6 is extended to indexes (no variant shares an indexed value with another record). The held index is truthful (value to holders, smallest key wins among plain duplicates). A conditional merge onto disk is judged on the merged set. Soul made five passes. The held index keeping one key per value caused three rounds of store-bricking findings until the owner was fixed. Caching 249, Networking 293, Mesh 262, cross-runtime interop 36/36. Follow-ups: TS/Python plain-duplicate parity, a subscription id on per-change refusals, a removal-order test. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens
commit
6ffe17acf4
27 changed files with 3057 additions and 90 deletions
|
|
@ -28,7 +28,7 @@
|
|||
"code": {
|
||||
"description": "Machine-readable refusal code (R-N). Additive: a v0 peer that does not know this field ignores it.",
|
||||
"type": ["string", "null"],
|
||||
"enum": [null, "selection_invalid", "cursor_stale", "cursor_invalid", "reference_outside_target"]
|
||||
"enum": [null, "selection_invalid", "cursor_stale", "cursor_invalid", "reference_outside_target", "variant_unsupported"]
|
||||
},
|
||||
"details": {
|
||||
"description": "Code-specific structured detail (R-N). Every field is populated only for the code that defines it; the others are null, not omitted.",
|
||||
|
|
@ -36,8 +36,8 @@
|
|||
"additionalProperties": false,
|
||||
"required": ["field", "value", "asOf", "current"],
|
||||
"properties": {
|
||||
"field": { "description": "selection_invalid: the selection field that failed.", "type": ["string", "null"] },
|
||||
"value": { "description": "selection_invalid: the offending value, when there is one string worth naming.", "type": ["string", "null"] },
|
||||
"field": { "description": "selection_invalid: the selection field that failed. variant_unsupported: \"subscriptionId\", when a subscription was ended.", "type": ["string", "null"] },
|
||||
"value": { "description": "selection_invalid: the offending value, when there is one string worth naming. variant_unsupported: the id of the subscription that ended.", "type": ["string", "null"] },
|
||||
"asOf": { "description": "cursor_stale: the cursor's own minted asOf.", "type": ["integer", "null"], "minimum": 0 },
|
||||
"current": { "description": "cursor_stale: the answering server's current asOf.", "type": ["integer", "null"], "minimum": 0 }
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -829,6 +829,12 @@ class CultCacheTests(unittest.TestCase):
|
|||
message = str(caught.exception)
|
||||
self.assertTrue("cultcache.store.v2" in message or "item:anvil-big" in message, message)
|
||||
|
||||
def test_single_file_refuses_the_csharp_written_variant_store_by_version_or_record(self) -> None:
|
||||
with self.assertRaises(ValueError) as caught:
|
||||
self._pull_vector("../document-variants-c1/variant-store.msgpack")
|
||||
message = str(caught.exception)
|
||||
self.assertTrue("cultcache.store.v2" in message or "laser-big" in message, message)
|
||||
|
||||
def test_v1_store_written_at_the_base_commit_still_reads_byte_for_byte(self) -> None:
|
||||
envelopes = self._pull_vector("v1-base.msgpack")
|
||||
self.assertEqual([(e.key, e.type) for e in envelopes], [("alpha", "vectors.item"), ("beta", "vectors.item")])
|
||||
|
|
|
|||
|
|
@ -4679,6 +4679,16 @@ mod tests {
|
|||
);
|
||||
}
|
||||
|
||||
// The store the C# runtime wrote through its own variant API (tests/vectors/document-variants-c1).
|
||||
#[test]
|
||||
fn single_file_refuses_the_csharp_written_variant_store_by_version_or_record() {
|
||||
let message = refusal("../document-variants-c1/variant-store.msgpack");
|
||||
assert!(
|
||||
message.contains("cultcache.store.v2") || message.contains("laser-big"),
|
||||
"{message}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn v1_store_written_at_the_base_commit_still_reads_byte_for_byte() -> Result<()> {
|
||||
let envelopes = pull_vector("v1-base.msgpack")?;
|
||||
|
|
|
|||
|
|
@ -1443,6 +1443,13 @@ test("SingleFileMessagePackBackingStore refuses a variant store by version or re
|
|||
);
|
||||
});
|
||||
|
||||
test("SingleFileMessagePackBackingStore refuses the C#-written variant store by version or record", async () => {
|
||||
await assert.rejects(
|
||||
() => pullVector("../document-variants-c1/variant-store.msgpack"),
|
||||
(error: Error) => error.message.includes("cultcache.store.v2") || error.message.includes("laser-big"),
|
||||
);
|
||||
});
|
||||
|
||||
test("CultCache inspector refuses the same vectors by name", async () => {
|
||||
const inspect = async (name: string) => inspectCultCacheBytes(name, await readFile(join(variantVectors, name)));
|
||||
await assert.rejects(() => inspect("unknown-header.msgpack"), /cultcache\.store\.v9/u);
|
||||
|
|
|
|||
|
|
@ -28,7 +28,7 @@
|
|||
"code": {
|
||||
"description": "Machine-readable refusal code (R-N). Additive: a v0 peer that does not know this field ignores it.",
|
||||
"type": ["string", "null"],
|
||||
"enum": [null, "selection_invalid", "cursor_stale", "cursor_invalid", "reference_outside_target"]
|
||||
"enum": [null, "selection_invalid", "cursor_stale", "cursor_invalid", "reference_outside_target", "variant_unsupported"]
|
||||
},
|
||||
"details": {
|
||||
"description": "Code-specific structured detail (R-N). Every field is populated only for the code that defines it; the others are null, not omitted.",
|
||||
|
|
@ -36,8 +36,8 @@
|
|||
"additionalProperties": false,
|
||||
"required": ["field", "value", "asOf", "current"],
|
||||
"properties": {
|
||||
"field": { "description": "selection_invalid: the selection field that failed.", "type": ["string", "null"] },
|
||||
"value": { "description": "selection_invalid: the offending value, when there is one string worth naming.", "type": ["string", "null"] },
|
||||
"field": { "description": "selection_invalid: the selection field that failed. variant_unsupported: \"subscriptionId\", when a subscription was ended.", "type": ["string", "null"] },
|
||||
"value": { "description": "selection_invalid: the offending value, when there is one string worth naming. variant_unsupported: the id of the subscription that ended.", "type": ["string", "null"] },
|
||||
"asOf": { "description": "cursor_stale: the cursor's own minted asOf.", "type": ["integer", "null"], "minimum": 0 },
|
||||
"current": { "description": "cursor_stale: the answering server's current asOf.", "type": ["integer", "null"], "minimum": 0 }
|
||||
}
|
||||
|
|
|
|||
|
|
@ -250,7 +250,7 @@ impl CultNetShardDescriptor {
|
|||
}
|
||||
}
|
||||
|
||||
/// Machine-readable `cultnet.error.v0` refusal code (R-N). Mirrors the four strings
|
||||
/// Machine-readable `cultnet.error.v0` refusal code (R-N). Mirrors the five strings
|
||||
/// `contracts/cultnet/cultnet.error.schema.json` enumerates for `code`; a peer that receives a fifth
|
||||
/// string it does not know refuses to decode rather than guess a meaning for it.
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
|
||||
|
|
@ -260,6 +260,7 @@ pub enum CultNetErrorCode {
|
|||
CursorStale,
|
||||
CursorInvalid,
|
||||
ReferenceOutsideTarget,
|
||||
VariantUnsupported,
|
||||
}
|
||||
|
||||
impl CultNetErrorCode {
|
||||
|
|
@ -269,6 +270,7 @@ impl CultNetErrorCode {
|
|||
Self::CursorStale => "cursor_stale",
|
||||
Self::CursorInvalid => "cursor_invalid",
|
||||
Self::ReferenceOutsideTarget => "reference_outside_target",
|
||||
Self::VariantUnsupported => "variant_unsupported",
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -278,6 +280,7 @@ impl CultNetErrorCode {
|
|||
"cursor_stale" => Ok(Self::CursorStale),
|
||||
"cursor_invalid" => Ok(Self::CursorInvalid),
|
||||
"reference_outside_target" => Ok(Self::ReferenceOutsideTarget),
|
||||
"variant_unsupported" => Ok(Self::VariantUnsupported),
|
||||
other => Err(anyhow!("ErrorMessage.Code {other:?} is not a recognized cultnet.error.v0 code")),
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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!);
|
||||
|
|
@ -328,11 +339,11 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
|
|||
var started = tracePages ? Stopwatch.GetTimestamp() : 0L;
|
||||
var metadata = records[index];
|
||||
var record = ReadPersistedRecordPage(metadata, out var pagePayload);
|
||||
recordReports[index] = Registry.ResolvePersistedSchemaReport(record.SchemaId, catalogEntries);
|
||||
storedRecords[index] = ToStoredDocument(
|
||||
record,
|
||||
catalogEntries,
|
||||
(type, payload) => CultDocumentMessagePackSerialization.DeserializeUntyped(type, payload, Registry));
|
||||
(type, payload) => CultDocumentMessagePackSerialization.DeserializeUntyped(type, payload, Registry),
|
||||
out recordReports[index]);
|
||||
if (tracePages)
|
||||
{
|
||||
pageBytes[index] = pagePayload.LongLength;
|
||||
|
|
@ -491,6 +502,10 @@ public sealed class DirectoryMessagePackBackingStore : CacheBackingStore
|
|||
var record = CultDocumentMessagePackSerialization.DeserializePersistedRecord(pagePayload);
|
||||
if (!string.Equals(record.Key, metadata.Key, StringComparison.Ordinal))
|
||||
throw new InvalidDataException($"Record page '{path}' contains key '{record.Key}', expected '{metadata.Key}'.");
|
||||
// The shared decoder reads a variant page; this store does not hold variants, so a load refuses it as a write does.
|
||||
if (record.Variant != null)
|
||||
throw new NotSupportedException(
|
||||
$"Record '{record.Key}' is a variant of '{record.Variant.BaseKey}'; the directory store does not hold document variants yet.");
|
||||
return record;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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,20 @@ 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 follows the same outcome as the same member on a plain record; there
|
||||
is no separate rule for variants, and no separate report.
|
||||
|
||||
- An override naming a slot the type no longer has is ignored, exactly as a plain record
|
||||
ignores a dropped member. The drift is reported once, by the store's schema resolution
|
||||
(`LastSchemaMigrationReports`), for the variant's record as for any other record of the
|
||||
schema. The variant inherits the base value.
|
||||
- An override holding a value that no longer decodes into the member's type refuses the
|
||||
load, exactly as a plain record holding such a value does; the refusal names the variant
|
||||
and the member.
|
||||
|
||||
Drift is for type changes after the fact. A write is validated: an override for a slot the
|
||||
type does not have, a value that does not decode as the member, or two overrides of one
|
||||
slot is refused when written.
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
}
|
||||
|
||||
|
|
@ -1728,8 +1799,12 @@ namespace GameCult.Caching
|
|||
private readonly List<(CacheBackingStore Store, Type[] Homes)> _stores = new();
|
||||
private readonly Dictionary<string, CultStoredDocument> _entries = new(StringComparer.Ordinal);
|
||||
private readonly Dictionary<Type, Dictionary<string, string>> _names = new();
|
||||
private readonly Dictionary<(Type Type, string Alias), Dictionary<string, string>> _indexes = new();
|
||||
// Every key holding a value (unordered): the index is truthful about duplicates, so removing one holder never hides another.
|
||||
private readonly Dictionary<(Type Type, string Alias), Dictionary<string, List<string>>> _indexes = new();
|
||||
private readonly Dictionary<Type, string> _globals = new();
|
||||
private readonly CultCodec? _codec;
|
||||
// Keys holding a variant: derived from the entries by Apply, never decided anywhere else.
|
||||
private readonly HashSet<string> _variantKeys = new(StringComparer.Ordinal);
|
||||
private readonly ConditionalWeakTable<object, KeyBox> _handles = new();
|
||||
private readonly Subject<Change> _changes = new();
|
||||
private readonly object _gate = new();
|
||||
|
|
@ -1738,13 +1813,34 @@ 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;
|
||||
|
||||
// Read under the gate, by a store judging a merge.
|
||||
internal bool HoldsVariants => _variantKeys.Count > 0;
|
||||
|
||||
public CultCodec? Codec => _codec;
|
||||
|
||||
// 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 +1937,8 @@ 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);
|
||||
store.Judging = (arriving, departing) => { Resolve(arriving, departing, store); };
|
||||
_stores.Add(candidate);
|
||||
try
|
||||
{
|
||||
|
|
@ -1851,6 +1948,7 @@ namespace GameCult.Caching
|
|||
{
|
||||
_stores.Remove(candidate);
|
||||
store.Loaded = null;
|
||||
store.Judging = null;
|
||||
store.Cache = null;
|
||||
throw;
|
||||
}
|
||||
|
|
@ -1915,11 +2013,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 +2076,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)
|
||||
|
|
@ -1994,11 +2127,16 @@ namespace GameCult.Caching
|
|||
.SelectMany(pair => pair.Value.TryGetValue(name, out var key) ? new[] { key } : Array.Empty<string>()));
|
||||
}
|
||||
|
||||
// A value held by one record finds it. Records of one type sharing a value (plain duplicates) find the holder with the
|
||||
// smallest key (ordinal), whatever order they were written or loaded in: the choice is a function of the held set, so
|
||||
// it is the same live, after a flush and after a reopen, and removing that holder finds the next.
|
||||
public T? GetByIndex<T>(string alias, string value) where T : class
|
||||
{
|
||||
return Single<T>($"index {alias}='{value}'", () => _indexes
|
||||
.Where(pair => pair.Key.Alias == alias && typeof(T).IsAssignableFrom(pair.Key.Type))
|
||||
.SelectMany(pair => pair.Value.TryGetValue(value, out var key) ? new[] { key } : Array.Empty<string>()));
|
||||
.SelectMany(pair => pair.Value.TryGetValue(value, out var keys)
|
||||
? new[] { keys.Aggregate((best, next) => string.CompareOrdinal(next, best) < 0 ? next : best) }
|
||||
: Array.Empty<string>()));
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
|
|
@ -2039,7 +2177,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 +2252,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 +2278,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 +2314,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 +2453,264 @@ 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();
|
||||
}
|
||||
|
||||
// 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);
|
||||
// A flatten resolves like the variant it replaces, against the post-batch set, so a base edited in the same
|
||||
// batch is in it; it lands as the plain record of that resolution.
|
||||
var toResolve = new List<string>(admitted.Where(stored => stored.Variant != null || stored.Flattens).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)
|
||||
{
|
||||
var unchanged = new VariantPlan(admitted);
|
||||
RefuseVariantIndexSharing(unchanged, incoming, gone);
|
||||
return unchanged;
|
||||
}
|
||||
|
||||
var pending = new HashSet<string>(toResolve, StringComparer.Ordinal);
|
||||
var done = new Dictionary<string, CultStoredDocument>(StringComparer.Ordinal);
|
||||
var stack = new List<string>();
|
||||
foreach (var key in toResolve)
|
||||
ResolveOne(key);
|
||||
|
||||
var plan = new VariantPlan(admitted.Select(stored => done.TryGetValue(stored.Key.Value, out var resolved) ? resolved : stored).ToList());
|
||||
foreach (var key in toResolve.Where(key => !incoming.ContainsKey(key)))
|
||||
plan.Dependents.Add(done[key]);
|
||||
RefuseVariantIndexSharing(plan, incoming, gone);
|
||||
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);
|
||||
|
||||
// A flatten carries no delta of its own; the variant it replaces does.
|
||||
var isVariant = variant.Variant != null;
|
||||
var delta = variant.Variant ?? _entries[key].Variant!;
|
||||
var baseStored = Lookup(delta.BaseKey) ?? throw MissingBaseRefusal(key, delta.BaseKey, gone.Contains(delta.BaseKey));
|
||||
if (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 seen = new HashSet<int>();
|
||||
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;
|
||||
if (!seen.Add(slot))
|
||||
throw new InvalidOperationException($"Variant {key} overrides slot {slot} more than once; a variant has one override per slot.");
|
||||
var member = descriptor.RichMembers.FirstOrDefault(candidate => candidate.Slot == slot);
|
||||
// Drift is the load's business (ToStoredDocument sheds an override of a slot the type lacks), so a delta
|
||||
// reaching here names only slots the type has, unless someone is writing one it never had.
|
||||
if (member == null)
|
||||
throw new InvalidOperationException($"Variant {key} overrides slot {slot}, which {type.Name} does not have.");
|
||||
|
||||
try
|
||||
{
|
||||
_codec.Deserialize(member.MemberType, type, entry.Value);
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
// A value that no longer decodes refuses exactly as a plain record holding it does.
|
||||
throw new InvalidOperationException(
|
||||
$"Variant {key} overrides '{member.Member.Name}' (slot {slot}) with a value that does not decode as {member.MemberType.Name}: " +
|
||||
exception.GetBaseException().Message, exception);
|
||||
}
|
||||
|
||||
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.");
|
||||
|
||||
if (isVariant)
|
||||
{
|
||||
// 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.");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
stack.RemoveAt(stack.Count - 1);
|
||||
return done[key] = isVariant
|
||||
? variant.Resolved(document)
|
||||
: new CultStoredDocument(variant.Key, variant.StoredAt, variant.Descriptor, document) { Flattens = true };
|
||||
}
|
||||
}
|
||||
|
||||
// The one index rule, judged on the commit's result whatever the write order: a variant may not share an indexed
|
||||
// value with another record. It reads the truthful held index (every holder of a value), minus the records this
|
||||
// commit replaces or removes, plus the records it lands. Plain-vs-plain sharing is not this rule's business.
|
||||
// A unique index maps a value to one key at lookup; a variant sharing it would be found in place of its
|
||||
// holder or lose the value, and the store would refuse to reopen.
|
||||
private void RefuseVariantIndexSharing(
|
||||
VariantPlan plan,
|
||||
IReadOnlyDictionary<string, CultStoredDocument> incoming,
|
||||
HashSet<string> gone)
|
||||
{
|
||||
var landing = plan.Admitted.Concat(plan.Dependents).ToList();
|
||||
if (landing.All(stored => stored.Descriptor.IndexAccessors.Count == 0))
|
||||
return;
|
||||
var replaced = new HashSet<string>(incoming.Keys.Concat(gone).Concat(plan.Dependents.Select(stored => stored.Key.Value)), StringComparer.Ordinal);
|
||||
var landingByKey = landing.ToDictionary(stored => stored.Key.Value, StringComparer.Ordinal);
|
||||
|
||||
bool IsVariant(string key) =>
|
||||
landingByKey.TryGetValue(key, out var landed) ? landed.Variant != null : _variantKeys.Contains(key);
|
||||
|
||||
// One pass over what lands, one lookup per (index, value): the rule is linear in the records landing.
|
||||
var landedValues = new List<(CultStoredDocument Stored, (Type Type, string Alias) Index, string Value)>();
|
||||
var landed = new Dictionary<((Type Type, string Alias) Index, string Value), List<string>>();
|
||||
foreach (var stored in landing)
|
||||
{
|
||||
foreach (var pair in stored.Descriptor.IndexAccessors)
|
||||
{
|
||||
var value = pair.Value(stored.Document);
|
||||
if (string.IsNullOrWhiteSpace(value))
|
||||
continue;
|
||||
var index = (stored.Descriptor.DocumentType, pair.Key);
|
||||
landedValues.Add((stored, index, value));
|
||||
if (!landed.TryGetValue((index, value), out var landedKeys))
|
||||
landed[(index, value)] = landedKeys = new List<string>();
|
||||
landedKeys.Add(stored.Key.Value);
|
||||
}
|
||||
}
|
||||
|
||||
var holdersOf = new Dictionary<((Type Type, string Alias) Index, string Value), (string[] All, string[] Variants)>();
|
||||
foreach (var (stored, index, value) in landedValues)
|
||||
{
|
||||
if (!holdersOf.TryGetValue((index, value), out var known))
|
||||
{
|
||||
var all = (_indexes.TryGetValue(index, out var held) && held.TryGetValue(value, out var heldKeys)
|
||||
? heldKeys.Where(heldKey => !replaced.Contains(heldKey))
|
||||
: Enumerable.Empty<string>())
|
||||
.Concat(landed[(index, value)])
|
||||
.Distinct(StringComparer.Ordinal)
|
||||
.ToArray();
|
||||
holdersOf[(index, value)] = known = (all, all.Where(IsVariant).ToArray());
|
||||
}
|
||||
|
||||
var self = stored.Key.Value;
|
||||
var holder = known.All.FirstOrDefault(candidate => candidate != self);
|
||||
if (stored.Variant != null && holder != null)
|
||||
throw new InvalidOperationException(
|
||||
$"Variant {self} and {holder} both hold index {index.Alias} '{value}'; a variant must override its indexed member.");
|
||||
var variantHolder = known.Variants.FirstOrDefault(candidate => candidate != self);
|
||||
if (stored.Variant == null && variantHolder != null)
|
||||
throw new InvalidOperationException(
|
||||
$"Record {self} and variant {variantHolder} both hold index {index.Alias} '{value}'; " +
|
||||
"a variant may not share an indexed value with another record.");
|
||||
}
|
||||
}
|
||||
|
||||
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 +2718,11 @@ namespace GameCult.Caching
|
|||
Unindex(existing);
|
||||
_entries.Remove(stored.Key.Value);
|
||||
_handles.Remove(existing.Document);
|
||||
_variantKeys.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 +2734,10 @@ 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);
|
||||
changes.Add(new Change(
|
||||
previous == null ? CultCacheDocumentChangeKind.Added : CultCacheDocumentChangeKind.Updated,
|
||||
stored,
|
||||
|
|
@ -2353,7 +2801,7 @@ namespace GameCult.Caching
|
|||
{
|
||||
var value = pair.Value(stored.Document);
|
||||
if (!string.IsNullOrWhiteSpace(value))
|
||||
MapOf(_indexes, (type, pair.Key))[value] = stored.Key.Value;
|
||||
AddHolder(_indexes, (type, pair.Key), value, stored.Key.Value);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -2366,7 +2814,26 @@ namespace GameCult.Caching
|
|||
if (_names.TryGetValue(type, out var names))
|
||||
RemoveKey(names, stored.Key.Value);
|
||||
foreach (var pair in _indexes.Where(pair => pair.Key.Type == type))
|
||||
RemoveKey(pair.Value, stored.Key.Value);
|
||||
RemoveHolder(pair.Value, stored.Key.Value);
|
||||
}
|
||||
|
||||
private static void AddHolder(Dictionary<(Type Type, string Alias), Dictionary<string, List<string>>> indexes, (Type, string) index, string value, string key)
|
||||
{
|
||||
if (!indexes.TryGetValue(index, out var map))
|
||||
indexes[index] = map = new Dictionary<string, List<string>>(StringComparer.Ordinal);
|
||||
if (!map.TryGetValue(value, out var keys))
|
||||
map[value] = keys = new List<string>();
|
||||
keys.Remove(key);
|
||||
keys.Add(key);
|
||||
}
|
||||
|
||||
private static void RemoveHolder(Dictionary<string, List<string>> map, string key)
|
||||
{
|
||||
foreach (var pair in map.ToArray())
|
||||
{
|
||||
if (pair.Value.Remove(key) && pair.Value.Count == 0)
|
||||
map.Remove(pair.Key);
|
||||
}
|
||||
}
|
||||
|
||||
private static Dictionary<string, string> MapOf<TKey>(Dictionary<TKey, Dictionary<string, string>> maps, TKey key)
|
||||
|
|
@ -2440,6 +2907,11 @@ namespace GameCult.Caching
|
|||
// any of it; if the cache refuses a record the call throws and the store keeps its previous view.
|
||||
protected internal Action<IReadOnlyList<CultStoredDocument>, IReadOnlyList<CultStoredDocument>>? Loaded;
|
||||
|
||||
// Set by the cache at attach: the cache's own resolution (variants, their indexed values, their bases) run over a
|
||||
// set of records a store is about to hold, mutating nothing. It throws the refusal the cache would give the same
|
||||
// set arriving as a load.
|
||||
protected internal Action<IReadOnlyList<CultStoredDocument>, IReadOnlyList<CultStoredDocument>>? Judging;
|
||||
|
||||
private readonly object _detachedGate = new();
|
||||
internal CultCache? Cache;
|
||||
|
||||
|
|
@ -2486,21 +2958,62 @@ 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
|
||||
};
|
||||
}
|
||||
|
||||
// The one load surface: a persisted record resolved against the current type, with the report of what that
|
||||
// resolution shed. A plain record sheds a member its type dropped by never carrying it into the document; a variant
|
||||
// sheds the override of such a slot by never carrying it into the delta, so a later type reusing the slot number
|
||||
// cannot resurrect it and a flush needs no rule of its own. The report is the schema diff's ignored slots plus the
|
||||
// overrides whose slot the type does not have, including one the persisted schema never listed.
|
||||
protected CultStoredDocument ToStoredDocument(
|
||||
CultPersistedRecord record,
|
||||
IReadOnlyCollection<CultSchemaCatalogEntry> catalog,
|
||||
Func<Type, byte[], object> deserializePayload)
|
||||
Func<Type, byte[], object> deserializePayload,
|
||||
out CultSchemaMigrationReport report)
|
||||
{
|
||||
var resolution = Registry.ResolvePersistedSchemaDetailed(record.SchemaId, catalog);
|
||||
report = resolution.Report;
|
||||
if (record.Variant != null)
|
||||
{
|
||||
var descriptor = resolution.Descriptor;
|
||||
bool Has(CultVariantOverride entryOverride) =>
|
||||
entryOverride.Path.Count != 1 || descriptor.RichMembers.Any(member => member.Slot == entryOverride.Path[0].Slot);
|
||||
var shed = record.Variant.Overrides
|
||||
.Where(entryOverride => !Has(entryOverride))
|
||||
.Select(entryOverride => entryOverride.Path[0].Slot)
|
||||
.Distinct()
|
||||
.OrderBy(slot => slot)
|
||||
.ToArray();
|
||||
if (shed.Length > 0)
|
||||
{
|
||||
// A slot the schema diff already reported is not reported twice.
|
||||
var unreported = shed.Except(report.IgnoredExtraSlots).ToArray();
|
||||
report.IgnoredExtraSlots = report.IgnoredExtraSlots.Concat(unreported).OrderBy(slot => slot).ToArray();
|
||||
report.Warnings = report.Warnings.Concat(unreported.Select(slot => new CultSchemaMigrationWarning
|
||||
{
|
||||
Code = "ignored_extra_slot",
|
||||
Message = $"Variant {record.Key} overrides slot {slot}, which {descriptor.DocumentType.Name} does not have; the override is ignored and shed on flush."
|
||||
})).ToArray();
|
||||
}
|
||||
|
||||
// The cache resolves it against its base when it admits the load.
|
||||
return new CultStoredDocument(
|
||||
new CultRecordKey(record.Key),
|
||||
record.StoredAt,
|
||||
descriptor,
|
||||
new CultVariantDelta(record.Variant.BaseKey, record.Variant.Overrides.Where(Has).ToArray()),
|
||||
null);
|
||||
}
|
||||
|
||||
var document = deserializePayload(resolution.Descriptor.DocumentType, record.Payload);
|
||||
return new CultStoredDocument(
|
||||
new CultRecordKey(record.Key),
|
||||
|
|
@ -2556,8 +3069,8 @@ namespace GameCult.Caching
|
|||
var persisted = new Dictionary<string, CultStoredDocument>(StringComparer.Ordinal);
|
||||
foreach (var record in snapshot.Records)
|
||||
{
|
||||
reports.Add(Registry.ResolvePersistedSchemaReport(record.SchemaId, snapshot.SchemaCatalog));
|
||||
var stored = ToStoredDocument(record, snapshot.SchemaCatalog, DeserializePayload);
|
||||
var stored = ToStoredDocument(record, snapshot.SchemaCatalog, DeserializePayload, out var report);
|
||||
reports.Add(report);
|
||||
persisted[stored.Key.Value] = stored;
|
||||
}
|
||||
|
||||
|
|
@ -2640,6 +3153,11 @@ namespace GameCult.Caching
|
|||
var catalog = ontoDisk
|
||||
? disk.SchemaCatalog
|
||||
: Entries.Values.Select(entry => entry.Descriptor.ToCatalogEntry());
|
||||
// The committer judged the batch against its own view; the file may have moved since. A merge is judged on the
|
||||
// set the file will hold, exactly as that set would be judged loading: what other processes wrote (arriving),
|
||||
// what they removed (departing), and this batch.
|
||||
if (ontoDisk && Judging != null && MergeInvolvesVariants(request, disk))
|
||||
JudgeMerge(request, disk);
|
||||
foreach (var entry in request.Deletes)
|
||||
records.Remove(entry.Key.Value);
|
||||
foreach (var entry in request.Upserts)
|
||||
|
|
@ -2654,6 +3172,35 @@ namespace GameCult.Caching
|
|||
return CultCommitOutcome.Committed;
|
||||
}
|
||||
|
||||
// Only a variant makes a merge more than the committer's own judgement: a variant it lands, one another writer wrote
|
||||
// (read from the record's header, never its payload), or one this cache holds. A plain-only store never decodes
|
||||
// another writer's records, so a schema this cache does not register cannot fail a commit that involves no variant.
|
||||
private bool MergeInvolvesVariants(CultCommitRequest request, CultPersistedStoreSnapshot disk) =>
|
||||
Cache?.HoldsVariants == true ||
|
||||
request.Upserts.Any(entry => entry.Variant != null) ||
|
||||
disk.Records.Any(record => record.Variant != null);
|
||||
|
||||
private void JudgeMerge(CultCommitRequest request, CultPersistedStoreSnapshot disk)
|
||||
{
|
||||
var landing = request.Upserts.Select(entry => entry.Key.Value).ToHashSet(StringComparer.Ordinal);
|
||||
var removing = request.Deletes.Select(entry => entry.Key.Value).ToHashSet(StringComparer.Ordinal);
|
||||
var onDisk = disk.Records.Select(record => record.Key).ToHashSet(StringComparer.Ordinal);
|
||||
var arriving = new List<CultStoredDocument>(request.Upserts);
|
||||
foreach (var record in disk.Records)
|
||||
{
|
||||
if (landing.Contains(record.Key) || removing.Contains(record.Key))
|
||||
continue;
|
||||
if (Entries.TryGetValue(record.Key, out var known) && known.StoredAt == record.StoredAt && known.Descriptor.SchemaId == record.SchemaId)
|
||||
continue;
|
||||
arriving.Add(ToStoredDocument(record, disk.SchemaCatalog, DeserializePayload, out _));
|
||||
}
|
||||
|
||||
var departing = new List<CultStoredDocument>(request.Deletes);
|
||||
departing.AddRange(Entries.Values.Where(known =>
|
||||
!onDisk.Contains(known.Key.Value) && !landing.Contains(known.Key.Value) && !removing.Contains(known.Key.Value)));
|
||||
Judging!(arriving, departing);
|
||||
}
|
||||
|
||||
private CultPersistedStoreSnapshot? ReadSnapshot()
|
||||
{
|
||||
FileInfo.Refresh();
|
||||
|
|
@ -2675,6 +3222,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)
|
||||
|
|
|
|||
91
src/GameCult.Caching/CultVariants.cs
Normal file
91
src/GameCult.Caching/CultVariants.cs
Normal file
|
|
@ -0,0 +1,91 @@
|
|||
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; }
|
||||
}
|
||||
}
|
||||
|
|
@ -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(
|
||||
|
|
@ -198,6 +200,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,
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
using System;
|
||||
using System.Collections.Concurrent;
|
||||
using System.Collections.Generic;
|
||||
using System.Linq;
|
||||
using System.Reflection;
|
||||
using System.Threading.Tasks;
|
||||
|
|
@ -502,16 +503,48 @@ namespace GameCult.Networking
|
|||
string subscriptionId,
|
||||
CultNetServerPeer peer)
|
||||
{
|
||||
// Keys this subscriber only ever received as a refusal: it was never given them as records.
|
||||
var refused = new HashSet<string>(StringComparer.Ordinal);
|
||||
return _database.WatchAllChanges().Subscribe(change =>
|
||||
{
|
||||
var outbound = CreateChangeMessage(change, subscriptionId, selection);
|
||||
var outbound = DeliverChange(change, subscriptionId, selection, refused, error => peer.SendCultNet(error));
|
||||
if (outbound != null)
|
||||
{
|
||||
peer.SendCultNet(outbound);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
// One subscriber's view of one change. A variant is not a record CultNet can carry (Q6): the subscriber is told, by key, and
|
||||
// is not later told the variant was removed, because it never held it.
|
||||
internal CultNetDatabaseChangeRawMessage? DeliverChange(
|
||||
object change,
|
||||
string subscriptionId,
|
||||
CultNetSelection selection,
|
||||
HashSet<string> refused,
|
||||
Action<CultNetErrorMessage> sendError)
|
||||
{
|
||||
CultNetDatabaseChangeRawMessage? outbound;
|
||||
try
|
||||
{
|
||||
outbound = CreateChangeMessage(change, subscriptionId, selection);
|
||||
}
|
||||
catch (NotSupportedException refusal)
|
||||
{
|
||||
var refusedKey = change.GetType().GetProperty("Key")?.GetValue(change) is CultRecordKey changed ? changed.Value : string.Empty;
|
||||
lock (refused)
|
||||
refused.Add(refusedKey);
|
||||
sendError(new CultNetErrorMessage { Error = refusal.Message });
|
||||
return null;
|
||||
}
|
||||
|
||||
if (outbound == null)
|
||||
return null;
|
||||
var recordKey = outbound.RecordKey ?? outbound.Document?.RecordKey ?? string.Empty;
|
||||
bool wasRefused;
|
||||
lock (refused)
|
||||
wasRefused = refused.Remove(recordKey);
|
||||
return wasRefused && outbound.ChangeKind == "removed" ? null : outbound;
|
||||
}
|
||||
|
||||
internal CultNetDatabaseChangeRawMessage? CreateChangeMessage(
|
||||
object change,
|
||||
string subscriptionId,
|
||||
|
|
@ -549,6 +582,7 @@ namespace GameCult.Networking
|
|||
};
|
||||
}
|
||||
|
||||
CultNetDocumentRegistry.RefuseVariantChange(_database.Cache, key);
|
||||
return new CultNetDatabaseChangeRawMessage
|
||||
{
|
||||
MessageId = Guid.NewGuid().ToString("N"),
|
||||
|
|
|
|||
|
|
@ -204,6 +204,13 @@ namespace GameCult.Networking
|
|||
peer.SendCultNet(CultNetErrorMessage.ForReferenceOutsideTarget(ex));
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
catch (NotSupportedException refusal)
|
||||
{
|
||||
// Q6: a snapshot that selects a variant is refused to this subscriber as a typed error.
|
||||
Withdraw(key, sendRemovals: false, forgetRequest: true);
|
||||
peer.SendCultNet(CultNetErrorMessage.ForVariantUnsupported(refusal.Message, subscriptionId));
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
catch
|
||||
{
|
||||
Withdraw(key, sendRemovals: false, forgetRequest: true);
|
||||
|
|
@ -315,7 +322,18 @@ namespace GameCult.Networking
|
|||
}
|
||||
|
||||
var sourceRecordKey = ResolveChangeRecordKey(change);
|
||||
var matched = CreateMatchedRecord(change, request, peer);
|
||||
CultNetRawDocumentRecord? matched;
|
||||
try
|
||||
{
|
||||
matched = CreateMatchedRecord(change, request, peer);
|
||||
}
|
||||
catch (NotSupportedException refusal)
|
||||
{
|
||||
// Q6: a variant is not a record CultNet can carry; the subscriber that would have received it is told, by key.
|
||||
peer.SendCultNet(CultNetErrorMessage.ForVariantUnsupported(refusal.Message));
|
||||
return;
|
||||
}
|
||||
|
||||
ApplyProjectedChange(
|
||||
projection,
|
||||
sourceRecordKey,
|
||||
|
|
@ -331,9 +349,23 @@ namespace GameCult.Networking
|
|||
if (!_projections.TryGetValue(key, out var projection)) return;
|
||||
var legacyRequest = request.ToLegacyShape(key.Id);
|
||||
var authorized = _authorizeRequest?.Invoke(legacyRequest, key.Peer) != false;
|
||||
var next = authorized
|
||||
? CreateProjectedSnapshot(request, key.Peer)
|
||||
: ProjectedSnapshot.Empty;
|
||||
ProjectedSnapshot next;
|
||||
try
|
||||
{
|
||||
next = authorized
|
||||
? CreateProjectedSnapshot(request, key.Peer)
|
||||
: ProjectedSnapshot.Empty;
|
||||
}
|
||||
catch (NotSupportedException refusal)
|
||||
{
|
||||
// Q6: the selection selects a variant, which CultNet cannot carry. The subscription is withdrawn, so the peer is
|
||||
// told once, not once per store change; the rows it delivered are withdrawn too (removals first), so the peer
|
||||
// does not keep them as live, and the typed error names the subscription that ended. A change never throws out of
|
||||
// the change handler.
|
||||
Withdraw(key, sendRemovals: true, forgetRequest: true);
|
||||
key.Peer.SendCultNet(CultNetErrorMessage.ForVariantUnsupported(refusal.Message, key.Id));
|
||||
return;
|
||||
}
|
||||
|
||||
// With _projectRecord gone (docs/cultnet-selection-cut.md, S2-2), a projection's dictionary
|
||||
// key is always the record's own RecordKey/SchemaId, so the two can never disagree here;
|
||||
|
|
@ -541,6 +573,7 @@ namespace GameCult.Networking
|
|||
if (kind == CultNetDatabaseChangeKind.Removed || document == null)
|
||||
return null;
|
||||
|
||||
CultNetDocumentRegistry.RefuseVariantChange(_database.Cache, key);
|
||||
return _database.Documents.ToRawRecord(descriptor, key, document, DateTimeOffset.UtcNow.ToString("O"));
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -391,11 +391,13 @@ namespace GameCult.Networking
|
|||
var stored = cache.AllStoredDocuments;
|
||||
if (rowFilter != null)
|
||||
stored = stored.Where(entry => rowFilter(entry.Descriptor, entry.Key));
|
||||
var rows = stored
|
||||
var entries = stored.ToArray();
|
||||
var rows = entries
|
||||
.Select(entry => new CultNetSelectionEvaluator.Row(
|
||||
entry.Descriptor, entry.Key, entry.Document, ordinalOf(entry.Descriptor.SchemaId, entry.Key), entry.StoredAt))
|
||||
.ToArray();
|
||||
var full = CultNetSelectionEvaluator.EvaluateAll(_documents, rows, selection);
|
||||
RefuseVariants(entries, full.Ordered.Select(row => row.Key));
|
||||
// R-Q: a page's asOf is exact only when every matched row is committed through one shard's
|
||||
// log - a caller that knows about shards (CultNetDatabaseServer) supplies shardIdOf; one
|
||||
// that does not (the raw v0/shard-scoped rowFilter paths, which are already bounded to a
|
||||
|
|
@ -415,6 +417,7 @@ namespace GameCult.Networking
|
|||
}
|
||||
|
||||
var evaluation = CultNetSelectionEvaluator.Page(full, selection, asOf, cursorKey);
|
||||
RefuseVariants(entries, evaluation.Edges.Select(edge => edge.From.Key));
|
||||
var records = evaluation.Rows.Select(row => ToRawRecord(row, options)).ToArray();
|
||||
var wantDocument = selection.Projection == CultNetSelectionProjections.Document;
|
||||
|
||||
|
|
@ -485,17 +488,20 @@ namespace GameCult.Networking
|
|||
var stored = cache.AllStoredDocuments;
|
||||
if (rowFilter != null)
|
||||
stored = stored.Where(entry => rowFilter(entry.Descriptor, entry.Key));
|
||||
var rows = stored
|
||||
var entries = stored.ToArray();
|
||||
var rows = entries
|
||||
.Select(entry => new CultNetSelectionEvaluator.Row(
|
||||
entry.Descriptor, entry.Key, entry.Document, ordinalOf(entry.Descriptor.SchemaId, entry.Key), entry.StoredAt))
|
||||
.ToArray();
|
||||
|
||||
var full = CultNetSelectionEvaluator.EvaluateAll(_documents, rows, selection);
|
||||
RefuseVariants(entries, full.Ordered.Select(row => row.Key));
|
||||
var records = full.Ordered.Select(row => ToRawRecord(row, options)).ToArray();
|
||||
var wantDocument = selection.Projection == CultNetSelectionProjections.Document;
|
||||
var edges = selection.HasHop
|
||||
? CultNetSelectionEvaluator.EdgesFor(full.Ordered, full).Select(edge => ToEdge(edge, wantDocument)).ToArray()
|
||||
: null;
|
||||
var hopEdges = selection.HasHop ? CultNetSelectionEvaluator.EdgesFor(full.Ordered, full) : null;
|
||||
if (hopEdges != null)
|
||||
RefuseVariants(entries, hopEdges.Select(edge => edge.From.Key));
|
||||
var edges = hopEdges?.Select(edge => ToEdge(edge, wantDocument)).ToArray();
|
||||
|
||||
return new CultNetSelectionPage
|
||||
{
|
||||
|
|
@ -549,6 +555,34 @@ 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.
|
||||
// The refusal follows the selection: a variant is refused only when the selection selects it, never for existing. An
|
||||
// edge carries its citer's reference payload, read from the citer's resolved document, so a variant citing is selected
|
||||
// by the selection's hop exactly as a variant row is.
|
||||
private static void RefuseVariants(IReadOnlyList<CultStoredDocument> entries, IEnumerable<CultRecordKey> selected)
|
||||
{
|
||||
if (!entries.Any(entry => entry.Variant != null))
|
||||
return;
|
||||
var variants = entries.Where(entry => entry.Variant != null).ToDictionary(entry => entry.Key.Value, StringComparer.Ordinal);
|
||||
foreach (var key in selected)
|
||||
{
|
||||
if (variants.TryGetValue(key.Value, out var variant))
|
||||
throw VariantRefusal(variant);
|
||||
}
|
||||
}
|
||||
|
||||
// The live-change funnel's refusal: a load or write that lands a variant must not reach a subscriber as a record.
|
||||
internal static void RefuseVariantChange(CultCache cache, CultRecordKey key)
|
||||
{
|
||||
// Recorded, not fixed: a Flatten committed on another thread between the change's admission and this read makes
|
||||
// the record plain here, and its resolved view is delivered. The window is one thread hop wide.
|
||||
if (cache.GetStored(key) is { Variant: not null } entry)
|
||||
throw VariantRefusal(entry);
|
||||
}
|
||||
|
||||
private static NotSupportedException VariantRefusal(CultStoredDocument entry) =>
|
||||
new($"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").
|
||||
|
|
|
|||
|
|
@ -546,6 +546,18 @@ namespace GameCult.Networking
|
|||
: null
|
||||
};
|
||||
|
||||
/// <summary>
|
||||
/// Builds the wire error for a selection that selects a document variant, which CultNet does not carry yet: code
|
||||
/// <c>variant_unsupported</c>, details <c>{ field: "subscriptionId", value }</c> when a subscription was ended by it.
|
||||
/// </summary>
|
||||
public static CultNetErrorMessage ForVariantUnsupported(string message, string? subscriptionId = null) =>
|
||||
new CultNetErrorMessage
|
||||
{
|
||||
Error = message,
|
||||
Code = "variant_unsupported",
|
||||
Details = subscriptionId == null ? null : new CultNetErrorDetails { Field = "subscriptionId", Value = subscriptionId }
|
||||
};
|
||||
|
||||
/// <summary>
|
||||
/// Builds the wire error for an out-of-target reference refusal (R-N): code
|
||||
/// <c>reference_outside_target</c>, no details.
|
||||
|
|
|
|||
|
|
@ -131,11 +131,30 @@ namespace GameCult.Caching.Tests
|
|||
Assert.That(message, Does.Contain(ItemSchemaId));
|
||||
}
|
||||
|
||||
// C1 reads v2. This vector was written by hand before C1 existed and its variant keeps its base's name, so a
|
||||
// cache with a codec refuses it under R6, naming both records; a cache without one cannot resolve it at all.
|
||||
[Test]
|
||||
public void SingleFileRefusesVariantStoreByVersionOrRecord()
|
||||
public void SingleFileRefusesTheHandWrittenVariantVectorNamingTheRecords()
|
||||
{
|
||||
Assert.That(Refusal("variant-v2.msgpack"),
|
||||
Does.Contain("cultcache.store.v2").Or.Contain("item:anvil-big"));
|
||||
var root = Path.Combine(Path.GetTempPath(), $"cultlib-vector-{Guid.NewGuid():N}");
|
||||
Directory.CreateDirectory(root);
|
||||
try
|
||||
{
|
||||
var file = Path.Combine(root, "store.msgpack");
|
||||
File.Copy(VectorPath("variant-v2.msgpack"), file);
|
||||
|
||||
using var withCodec = new CultCache(CultDocumentRegistry.Shared, CultCacheMessagePack.CreateCodec(CultDocumentRegistry.Shared));
|
||||
var named = Assert.Throws<InvalidOperationException>(() => withCodec.AddBackingStore(new SingleFileMessagePackBackingStore(file)))!;
|
||||
Assert.That(named.Message, Does.Contain("item:anvil-big").And.Contain("item:anvil"));
|
||||
|
||||
using var without = new CultCache();
|
||||
var codecless = Assert.Throws<InvalidOperationException>(() => without.AddBackingStore(new SingleFileMessagePackBackingStore(file)))!;
|
||||
Assert.That(codecless.Message, Does.Contain("item:anvil-big").And.Contain("codec"));
|
||||
}
|
||||
finally
|
||||
{
|
||||
if (Directory.Exists(root)) Directory.Delete(root, recursive: true);
|
||||
}
|
||||
}
|
||||
|
||||
[Test]
|
||||
|
|
|
|||
1461
tests/GameCult.Caching.Tests/VariantTests.cs
Normal file
1461
tests/GameCult.Caching.Tests/VariantTests.cs
Normal file
File diff suppressed because it is too large
Load diff
|
|
@ -45,10 +45,16 @@ public sealed class CultMeshVariantRefusalTests
|
|||
message.Should().Contain("item:anvil").And.Contain(ItemSchemaId);
|
||||
}
|
||||
|
||||
// Q6: CultMesh does not resolve variants. It refuses a variant key naming it and its base, and still reads the plain
|
||||
// records of the same v2 store.
|
||||
[Test]
|
||||
public void ReadSingleFileDocument_RefusesVariantStoreByVersionOrRecord()
|
||||
public void ReadSingleFileDocument_RefusesAVariantKeyByNameButReadsItsPlainNeighbours()
|
||||
{
|
||||
var message = Refusal("variant-v2.msgpack").Message;
|
||||
(message.Contains("cultcache.store.v2") || message.Contains("item:anvil-big")).Should().BeTrue(message);
|
||||
var refusal = Assert.Throws<NotSupportedException>(() =>
|
||||
CultMesh.ReadSingleFileDocumentPayload(VectorPath("variant-v2.msgpack"), new CultRecordKey("item:anvil-big"), ItemSchemaId))!;
|
||||
refusal.Message.Should().Contain("item:anvil-big").And.Contain("item:anvil'");
|
||||
|
||||
CultMesh.ReadSingleFileDocumentPayload(VectorPath("variant-v2.msgpack"), new CultRecordKey("item:bellows"), ItemSchemaId)
|
||||
.Should().NotBeEmpty();
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,470 @@
|
|||
#nullable enable
|
||||
using System;
|
||||
using System.Collections.Concurrent;
|
||||
using System.IO;
|
||||
using System.Linq;
|
||||
using System.Net;
|
||||
using System.Net.Sockets;
|
||||
using System.Reflection;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using GameCult.Caching;
|
||||
using GameCult.Caching.MessagePack;
|
||||
using MessagePack;
|
||||
using NUnit.Framework;
|
||||
|
||||
namespace GameCult.Networking.Tests
|
||||
{
|
||||
// Q6 (docs/document-variants-cut.md C1): CultNet carries no variant deltas yet, so building a row for one is refused
|
||||
// loudly, on the v1 selection path and on the v0 snapshot path alike.
|
||||
public sealed class CultNetDocumentRegistryVariantTests
|
||||
{
|
||||
private static readonly CultRecordKey BaseKey = new("variant-fixture-base");
|
||||
private static readonly CultRecordKey VariantKey = new("variant-fixture-big");
|
||||
|
||||
private static async Task<(CultCache Cache, CultNetDocumentRegistry Documents)> BuildFixtureAsync(bool withVariant)
|
||||
{
|
||||
var registry = CultDocumentRegistry.Shared;
|
||||
var cache = new CultCache(registry, CultCacheMessagePack.CreateCodec(registry));
|
||||
var documents = new CultNetDocumentRegistry(registry)
|
||||
.Register(CultNetDocumentBinding.ForDocument<VariantWireFixture>(registry));
|
||||
await cache.UpsertAsync(new VariantWireFixture { Name = "base", Power = 1 }, new CultRecordHandle<VariantWireFixture>(BaseKey));
|
||||
if (withVariant)
|
||||
await cache.UpsertVariantAsync(VariantKey, BaseKey, new[]
|
||||
{
|
||||
cache.Override<VariantWireFixture>(nameof(VariantWireFixture.Name), "big"),
|
||||
cache.Override<VariantWireFixture>(nameof(VariantWireFixture.Power), 9)
|
||||
});
|
||||
return (cache, documents);
|
||||
}
|
||||
|
||||
[Test]
|
||||
public async Task SelectionResponseRefusesAVariantRowNamingIt()
|
||||
{
|
||||
var (cache, documents) = await BuildFixtureAsync(withVariant: true);
|
||||
var selection = new CultNetSelection { Projection = CultNetSelectionProjections.Document };
|
||||
|
||||
var error = Assert.Throws<NotSupportedException>(() =>
|
||||
documents.CreateSelectionResponse(cache, "m1", selection, ordinalOf: (_, _) => 1, asOf: 1))!;
|
||||
|
||||
Assert.That(error.Message, Does.Contain(VariantKey.Value).And.Contain(BaseKey.Value));
|
||||
}
|
||||
|
||||
[Test]
|
||||
public async Task RawSnapshotResponseRefusesAVariantRowNamingIt()
|
||||
{
|
||||
var (cache, documents) = await BuildFixtureAsync(withVariant: true);
|
||||
|
||||
var error = Assert.Throws<NotSupportedException>(() => documents.CreateRawSnapshotResponse(cache, "m1"))!;
|
||||
|
||||
Assert.That(error.Message, Does.Contain(VariantKey.Value));
|
||||
}
|
||||
|
||||
[Test]
|
||||
public async Task ACacheWithoutVariantsStillAnswersBothPaths()
|
||||
{
|
||||
var (cache, documents) = await BuildFixtureAsync(withVariant: false);
|
||||
var selection = new CultNetSelection { Projection = CultNetSelectionProjections.Document };
|
||||
|
||||
Assert.That(documents.CreateSelectionResponse(cache, "m1", selection, ordinalOf: (_, _) => 1, asOf: 1).Documents, Has.Length.EqualTo(1));
|
||||
Assert.That(documents.CreateRawSnapshotResponse(cache, "m2").Documents, Has.Length.EqualTo(1));
|
||||
}
|
||||
|
||||
// The live-change funnel: a load that lands a variant reaches a subscriber as neither a record nor silence.
|
||||
[Test]
|
||||
public async Task ALoadThatLandsAVariantNeverReachesASubscriberAsARecordAndTheSubscriberIsToldByKey()
|
||||
{
|
||||
var registry = CultDocumentRegistry.Shared;
|
||||
var directory = Path.Combine(Path.GetTempPath(), $"cultlib-variant-live-{Guid.NewGuid():N}");
|
||||
Directory.CreateDirectory(directory);
|
||||
try
|
||||
{
|
||||
var path = Path.Combine(directory, "wire.cc");
|
||||
using (var writer = CultCacheMessagePack.Create(path, new CultCacheOpenOptions { Registry = registry }))
|
||||
{
|
||||
writer.Commit(batch => batch.Upsert(typeof(VariantWireFixture), new VariantWireFixture { Name = "base", Power = 1 }, BaseKey));
|
||||
writer.Commit(batch => batch.UpsertVariant(VariantKey, BaseKey, new[]
|
||||
{
|
||||
writer.Override<VariantWireFixture>(nameof(VariantWireFixture.Name), "big"),
|
||||
writer.Override<VariantWireFixture>(nameof(VariantWireFixture.Power), 9)
|
||||
}));
|
||||
}
|
||||
|
||||
var cache = new CultCache(registry, CultCacheMessagePack.CreateCodec(registry));
|
||||
var documents = new CultNetDocumentRegistry(registry)
|
||||
.Register(CultNetDocumentBinding.ForDocument<VariantWireFixture>(registry));
|
||||
var database = new CultNetDatabase(cache, new CultNetDatabaseOptions { DocumentRegistry = documents });
|
||||
var socket = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp);
|
||||
socket.Bind(new IPEndPoint(IPAddress.Loopback, 0));
|
||||
socket.ReceiveTimeout = 20;
|
||||
using var server = new RudpCultNetSchemaServer(new RudpCultNetSchemaServerOptions
|
||||
{
|
||||
RuntimeId = "variant-live-server",
|
||||
Socket = socket
|
||||
});
|
||||
using var subscriptions = new CultNetDatabaseSubscriptionServer(server, database);
|
||||
using var cancellation = new CancellationTokenSource();
|
||||
var serverThread = new Thread(() =>
|
||||
{
|
||||
while (!cancellation.IsCancellationRequested)
|
||||
{
|
||||
_ = server.PollOnceAsync().GetAwaiter().GetResult();
|
||||
Thread.Sleep(1);
|
||||
}
|
||||
}) { IsBackground = true };
|
||||
serverThread.Start();
|
||||
try
|
||||
{
|
||||
using var client = CultNetSchemaClients.CreateRudp("variant-live-client");
|
||||
var snapshot = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
var changes = new ConcurrentQueue<CultNetDatabaseChangeRawMessage>();
|
||||
var errors = new ConcurrentQueue<CultNetErrorMessage>();
|
||||
client.OnCultNet<CultNetSnapshotResponseRawMessage>(_ => snapshot.TrySetResult(true));
|
||||
client.OnCultNet<CultNetDatabaseChangeRawMessage>(changes.Enqueue);
|
||||
client.OnCultNet<CultNetErrorMessage>(errors.Enqueue);
|
||||
client.Connect("127.0.0.1", server.LocalEndPoint.Port);
|
||||
await WaitUntilAsync(() => client.Connected);
|
||||
client.SendCultNet(new CultNetDatabaseSubscribeMessage
|
||||
{
|
||||
MessageId = "subscribe-all",
|
||||
SubscriptionId = "all",
|
||||
RecordKeys = new[] { BaseKey.Value, VariantKey.Value },
|
||||
IncludeSnapshot = true
|
||||
});
|
||||
await WaitUntilAsync(() => snapshot.Task.IsCompleted);
|
||||
|
||||
cache.AddBackingStore(new SingleFileMessagePackBackingStore(path));
|
||||
|
||||
await WaitUntilAsync(() => !errors.IsEmpty || changes.Any(change => change.Document?.RecordKey == VariantKey.Value),
|
||||
() => $"changes={string.Join(",", changes.Select(c => c.ChangeKind + ":" + c.Document?.RecordKey))} errors={string.Join(",", errors.Select(e => e.Error))} cacheKeys={cache.GetStored(BaseKey) != null}/{cache.GetStored(VariantKey) != null}");
|
||||
await Task.Delay(200); // let anything else already in flight arrive
|
||||
|
||||
Assert.That(changes.Select(change => change.Document?.RecordKey), Does.Not.Contain(VariantKey.Value),
|
||||
"a variant must not reach a subscriber as a plain record");
|
||||
Assert.That(errors.Select(error => error.Error), Has.Some.Contains(VariantKey.Value),
|
||||
"the subscriber is told, naming the variant");
|
||||
}
|
||||
finally
|
||||
{
|
||||
cancellation.Cancel();
|
||||
serverThread.Join();
|
||||
}
|
||||
}
|
||||
finally
|
||||
{
|
||||
Directory.Delete(directory, recursive: true);
|
||||
}
|
||||
}
|
||||
|
||||
[Test]
|
||||
public async Task TheDatabaseServersLiveChangeForAVariantIsRefusedNamingIt()
|
||||
{
|
||||
var (cache, documents) = await BuildFixtureAsync(withVariant: true);
|
||||
var database = new CultNetDatabase(cache, new CultNetDatabaseOptions { DocumentRegistry = documents });
|
||||
using var server = new Server(cache, ServerSecurityOptions.Development());
|
||||
using var databaseServer = new CultNetDatabaseServer(server, database);
|
||||
var schemaId = cache.Registry.GetRequired<VariantWireFixture>().SchemaId;
|
||||
var change = new CultNetDatabaseChange<VariantWireFixture>(
|
||||
CultNetDatabaseChangeKind.Added,
|
||||
VariantKey,
|
||||
schemaId,
|
||||
database.Shards[0],
|
||||
cache.Get<VariantWireFixture>(VariantKey)!,
|
||||
previousDocument: null);
|
||||
var method = typeof(CultNetDatabaseServer).GetMethod("CreateChangeMessage", BindingFlags.Instance | BindingFlags.NonPublic)!;
|
||||
|
||||
var refused = Assert.Throws<TargetInvocationException>(() => method.Invoke(databaseServer, new object[]
|
||||
{
|
||||
change, "sub-1", new CultNetSelection { Projection = CultNetSelectionProjections.Document }
|
||||
}))!;
|
||||
|
||||
Assert.That(refused.InnerException, Is.TypeOf<NotSupportedException>());
|
||||
Assert.That(refused.InnerException!.Message, Does.Contain(VariantKey.Value).And.Contain(BaseKey.Value));
|
||||
}
|
||||
|
||||
|
||||
// A hop-bearing selection reconciles the whole store on every change. The refusal follows the selection: a subscriber
|
||||
// whose selection never selects the variant keeps receiving; one that does is told, as a typed error, not a silent stop.
|
||||
[Test]
|
||||
public async Task AHopBearingSubscriptionReconcilesPastAVariantItDoesNotSelectAndIsToldWhenItDoes()
|
||||
{
|
||||
var registry = CultDocumentRegistry.Shared;
|
||||
var cache = new CultCache(registry, CultCacheMessagePack.CreateCodec(registry));
|
||||
var documents = new CultNetDocumentRegistry(registry)
|
||||
.Register(CultNetDocumentBinding.ForDocument<VariantWireFixture>(registry))
|
||||
.Register(CultNetDocumentBinding.ForDocument<VariantCiterFixture>(registry));
|
||||
var wireSchemaId = registry.GetRequired<VariantWireFixture>().SchemaId;
|
||||
cache.Commit(batch =>
|
||||
{
|
||||
batch.Upsert(typeof(VariantWireFixture), new VariantWireFixture { Name = "base", Power = 1 }, BaseKey);
|
||||
batch.Upsert(typeof(VariantCiterFixture), Citer("citer-one", BaseKey), new CultRecordKey("citer-one"));
|
||||
});
|
||||
var database = new CultNetDatabase(cache, new CultNetDatabaseOptions { DocumentRegistry = documents });
|
||||
var socket = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp);
|
||||
socket.Bind(new IPEndPoint(IPAddress.Loopback, 0));
|
||||
socket.ReceiveTimeout = 20;
|
||||
using var server = new RudpCultNetSchemaServer(new RudpCultNetSchemaServerOptions { RuntimeId = "variant-hop-server", Socket = socket });
|
||||
using var subscriptions = new CultNetDatabaseSubscriptionServer(server, database);
|
||||
using var cancellation = new CancellationTokenSource();
|
||||
var serverThread = new Thread(() =>
|
||||
{
|
||||
while (!cancellation.IsCancellationRequested)
|
||||
{
|
||||
_ = server.PollOnceAsync().GetAwaiter().GetResult();
|
||||
Thread.Sleep(1);
|
||||
}
|
||||
}) { IsBackground = true };
|
||||
serverThread.Start();
|
||||
try
|
||||
{
|
||||
using var client = CultNetSchemaClients.CreateRudp("variant-hop-client");
|
||||
var snapshots = new ConcurrentQueue<CultNetSnapshotResponseRawV1Message>();
|
||||
var changes = new ConcurrentQueue<CultNetDatabaseChangeRawMessage>();
|
||||
var errors = new ConcurrentQueue<CultNetErrorMessage>();
|
||||
client.OnCultNet<CultNetSnapshotResponseRawV1Message>(snapshots.Enqueue);
|
||||
client.OnCultNet<CultNetDatabaseChangeRawMessage>(changes.Enqueue);
|
||||
client.OnCultNet<CultNetErrorMessage>(errors.Enqueue);
|
||||
client.Connect("127.0.0.1", server.LocalEndPoint.Port);
|
||||
await WaitUntilAsync(() => client.Connected);
|
||||
string Diagnostic() =>
|
||||
$"changes={string.Join(",", changes.Select(c => c.SubscriptionId + ":" + c.ChangeKind + ":" + (c.Document?.RecordKey ?? c.RecordKey)))} " +
|
||||
$"errors={string.Join(",", errors.Select(e => e.Error))}";
|
||||
|
||||
// "citers": rows citing the base. It never selects the variant.
|
||||
client.SendCultNet(new CultNetDatabaseSubscribeV1Message
|
||||
{
|
||||
MessageId = "citers",
|
||||
SubscriptionId = "citers",
|
||||
IncludeSnapshot = true,
|
||||
Selection = new CultNetSelection
|
||||
{
|
||||
Cites = new CultNetCitation { Target = new CultNetRecordRef { SchemaId = wireSchemaId, RecordKey = BaseKey.Value }, Role = "Ref" }
|
||||
}
|
||||
});
|
||||
await WaitUntilAsync(() => snapshots.Count == 1, Diagnostic);
|
||||
|
||||
await database.PutAsync(new CultRecordKey("citer-two"), Citer("citer-two", BaseKey));
|
||||
await WaitUntilAsync(() => changes.Any(change => change.SubscriptionId == "citers" && change.Document?.RecordKey == "citer-two"), Diagnostic);
|
||||
|
||||
// With a variant in the store, the same subscriber keeps receiving.
|
||||
await cache.UpsertVariantAsync(VariantKey, BaseKey, new[]
|
||||
{
|
||||
cache.Override<VariantWireFixture>(nameof(VariantWireFixture.Name), "big")
|
||||
});
|
||||
await database.PutAsync(new CultRecordKey("citer-three"), Citer("citer-three", BaseKey));
|
||||
await WaitUntilAsync(() => changes.Any(change => change.SubscriptionId == "citers" && change.Document?.RecordKey == "citer-three"), Diagnostic);
|
||||
Assert.That(errors, Is.Empty, "a subscriber that never selects the variant is never told about it");
|
||||
|
||||
// "cited": wire rows some citer cites. Once a citer cites the variant, it selects it.
|
||||
client.SendCultNet(new CultNetDatabaseSubscribeV1Message
|
||||
{
|
||||
MessageId = "cited",
|
||||
SubscriptionId = "cited",
|
||||
IncludeSnapshot = true,
|
||||
Selection = new CultNetSelection
|
||||
{
|
||||
Schemas = new[] { wireSchemaId },
|
||||
Cited = new CultNetIncoming { Role = "Ref", Exists = true }
|
||||
}
|
||||
});
|
||||
await WaitUntilAsync(() => snapshots.Count == 2, Diagnostic);
|
||||
Assert.That(errors, Is.Empty, "only the base is cited yet");
|
||||
await database.PutAsync(new CultRecordKey("citer-four"), Citer("citer-four", VariantKey));
|
||||
|
||||
await WaitUntilAsync(() => errors.Any(error => error.Error.Contains(VariantKey.Value)), Diagnostic);
|
||||
var refusal = errors.First(error => error.Error.Contains(VariantKey.Value));
|
||||
Assert.That(refusal.Code, Is.EqualTo("variant_unsupported"));
|
||||
Assert.That(refusal.Details?.Value, Is.EqualTo("cited"), "the peer can tell which subscription ended");
|
||||
await WaitUntilAsync(() => changes.Any(change => change.SubscriptionId == "cited" && change.ChangeKind == "removed" && change.RecordKey == BaseKey.Value), Diagnostic);
|
||||
Assert.That(changes.Where(change => change.SubscriptionId == "cited").Select(change => change.Document?.RecordKey),
|
||||
Does.Not.Contain(VariantKey.Value), "a variant never reaches a subscriber as a record");
|
||||
|
||||
// The first subscriber is still alive after the refusal; the refused one is withdrawn, not told again per change.
|
||||
var told = errors.Count;
|
||||
await database.PutAsync(new CultRecordKey("citer-five"), Citer("citer-five", BaseKey));
|
||||
await WaitUntilAsync(() => changes.Any(change => change.SubscriptionId == "citers" && change.Document?.RecordKey == "citer-five"), Diagnostic);
|
||||
await Task.Delay(200);
|
||||
Assert.That(errors.Count, Is.EqualTo(told), "a refused subscription is withdrawn: one error, not one per store change");
|
||||
}
|
||||
finally
|
||||
{
|
||||
cancellation.Cancel();
|
||||
serverThread.Join();
|
||||
}
|
||||
}
|
||||
|
||||
// An edge carries its citer's reference payload, read from the citer's resolved document. A selection whose hop
|
||||
// reaches a variant citer is refused as a selection of a variant row is, though only the cited row is selected.
|
||||
private static async Task<(CultCache Cache, CultNetDocumentRegistry Documents, CultNetSelection Selection, CultRecordKey CiterVariant)> BuildCitedByVariantFixtureAsync()
|
||||
{
|
||||
var registry = CultDocumentRegistry.Shared;
|
||||
var cache = new CultCache(registry, CultCacheMessagePack.CreateCodec(registry));
|
||||
var documents = new CultNetDocumentRegistry(registry)
|
||||
.Register(CultNetDocumentBinding.ForDocument<VariantWireFixture>(registry))
|
||||
.Register(CultNetDocumentBinding.ForDocument<VariantCiterFixture>(registry));
|
||||
cache.Commit(batch =>
|
||||
{
|
||||
batch.Upsert(typeof(VariantWireFixture), new VariantWireFixture { Name = "base", Power = 1 }, BaseKey);
|
||||
batch.Upsert(typeof(VariantCiterFixture), Citer("citer-one", BaseKey), new CultRecordKey("citer-one"));
|
||||
});
|
||||
var citerVariant = new CultRecordKey("citer-variant");
|
||||
await cache.UpsertVariantAsync(citerVariant, new CultRecordKey("citer-one"), new[]
|
||||
{
|
||||
cache.Override<VariantCiterFixture>(nameof(VariantCiterFixture.Name), "citer variant")
|
||||
});
|
||||
var selection = new CultNetSelection
|
||||
{
|
||||
Schemas = new[] { cache.Registry.GetRequired<VariantWireFixture>().SchemaId },
|
||||
Cited = new CultNetIncoming { Role = "Ref", Exists = true },
|
||||
Projection = CultNetSelectionProjections.Document
|
||||
};
|
||||
return (cache, documents, selection, citerVariant);
|
||||
}
|
||||
|
||||
[Test]
|
||||
public async Task AHopEdgeWhoseCiterIsAVariantIsRefusedNamingItAndAStoreWithoutOneAnswers()
|
||||
{
|
||||
var (cache, documents, selection, citerVariant) = await BuildCitedByVariantFixtureAsync();
|
||||
|
||||
var error = Assert.Throws<NotSupportedException>(() =>
|
||||
documents.CreateSelectionResponse(cache, "m1", selection, ordinalOf: (_, _) => 1, asOf: 1))!;
|
||||
Assert.That(error.Message, Does.Contain(citerVariant.Value));
|
||||
|
||||
cache.Commit(batch => batch.Remove(citerVariant));
|
||||
var answered = documents.CreateSelectionResponse(cache, "m2", selection, ordinalOf: (_, _) => 1, asOf: 1);
|
||||
Assert.That(answered.Edges, Has.Length.EqualTo(1));
|
||||
Assert.That(answered.Edges![0].From.RecordKey, Is.EqualTo("citer-one"));
|
||||
}
|
||||
|
||||
[Test]
|
||||
public async Task AHopSubscriptionWhoseEdgeCiterIsAVariantIsRefusedOnceAndWithdrawn()
|
||||
{
|
||||
var (cache, documents, selection, citerVariant) = await BuildCitedByVariantFixtureAsync();
|
||||
var database = new CultNetDatabase(cache, new CultNetDatabaseOptions { DocumentRegistry = documents });
|
||||
var socket = new Socket(AddressFamily.InterNetwork, SocketType.Dgram, ProtocolType.Udp);
|
||||
socket.Bind(new IPEndPoint(IPAddress.Loopback, 0));
|
||||
socket.ReceiveTimeout = 20;
|
||||
using var server = new RudpCultNetSchemaServer(new RudpCultNetSchemaServerOptions { RuntimeId = "variant-edge-server", Socket = socket });
|
||||
using var subscriptions = new CultNetDatabaseSubscriptionServer(server, database);
|
||||
using var cancellation = new CancellationTokenSource();
|
||||
var serverThread = new Thread(() =>
|
||||
{
|
||||
while (!cancellation.IsCancellationRequested)
|
||||
{
|
||||
_ = server.PollOnceAsync().GetAwaiter().GetResult();
|
||||
Thread.Sleep(1);
|
||||
}
|
||||
}) { IsBackground = true };
|
||||
serverThread.Start();
|
||||
try
|
||||
{
|
||||
using var client = CultNetSchemaClients.CreateRudp("variant-edge-client");
|
||||
var snapshots = new ConcurrentQueue<CultNetSnapshotResponseRawV1Message>();
|
||||
var errors = new ConcurrentQueue<CultNetErrorMessage>();
|
||||
client.OnCultNet<CultNetSnapshotResponseRawV1Message>(snapshots.Enqueue);
|
||||
client.OnCultNet<CultNetErrorMessage>(errors.Enqueue);
|
||||
client.Connect("127.0.0.1", server.LocalEndPoint.Port);
|
||||
await WaitUntilAsync(() => client.Connected);
|
||||
|
||||
client.SendCultNet(new CultNetDatabaseSubscribeV1Message
|
||||
{
|
||||
MessageId = "edges",
|
||||
SubscriptionId = "edges",
|
||||
IncludeSnapshot = true,
|
||||
Selection = selection
|
||||
});
|
||||
await WaitUntilAsync(() => errors.Any(error => error.Error.Contains(citerVariant.Value)));
|
||||
Assert.That(snapshots, Is.Empty, "the snapshot's edges would carry the variant's resolved reference");
|
||||
Assert.That(errors, Has.Count.EqualTo(1));
|
||||
}
|
||||
finally
|
||||
{
|
||||
cancellation.Cancel();
|
||||
serverThread.Join();
|
||||
}
|
||||
}
|
||||
|
||||
private static CultNetDatabaseChangeRawMessage? Deliver(
|
||||
CultNetDatabaseServer server,
|
||||
object change,
|
||||
CultNetSelection selection,
|
||||
System.Collections.Generic.HashSet<string> refused,
|
||||
System.Collections.Generic.List<CultNetErrorMessage> errors)
|
||||
{
|
||||
var method = typeof(CultNetDatabaseServer).GetMethod("DeliverChange", BindingFlags.Instance | BindingFlags.NonPublic)!;
|
||||
return (CultNetDatabaseChangeRawMessage?)method.Invoke(server, new object[]
|
||||
{
|
||||
change, "sub", selection, refused, (Action<CultNetErrorMessage>)errors.Add
|
||||
});
|
||||
}
|
||||
|
||||
private static VariantCiterFixture Citer(string name, CultRecordKey target) =>
|
||||
new() { Name = name, Ref = new CultRecordRef<VariantWireFixture>(target) };
|
||||
|
||||
// A subscriber told a variant only as an error is not later told it was removed: it never held it.
|
||||
[Test]
|
||||
public async Task RemovingAVariantTheSubscriberWasOnlyToldAboutAsAnErrorSendsNoRemoval()
|
||||
{
|
||||
var (cache, documents) = await BuildFixtureAsync(withVariant: true);
|
||||
var database = new CultNetDatabase(cache, new CultNetDatabaseOptions { DocumentRegistry = documents });
|
||||
using var server = new Server(cache, ServerSecurityOptions.Development());
|
||||
using var databaseServer = new CultNetDatabaseServer(server, database);
|
||||
var schemaId = cache.Registry.GetRequired<VariantWireFixture>().SchemaId;
|
||||
var selection = new CultNetSelection { Projection = CultNetSelectionProjections.Document };
|
||||
var refused = new System.Collections.Generic.HashSet<string>();
|
||||
var errors = new System.Collections.Generic.List<CultNetErrorMessage>();
|
||||
|
||||
var added = new CultNetDatabaseChange<VariantWireFixture>(
|
||||
CultNetDatabaseChangeKind.Added, VariantKey, schemaId, database.Shards[0], cache.Get<VariantWireFixture>(VariantKey)!, previousDocument: null);
|
||||
Assert.That(Deliver(databaseServer, added, selection, refused, errors), Is.Null);
|
||||
Assert.That(errors, Has.Count.EqualTo(1));
|
||||
|
||||
var removed = new CultNetDatabaseChange<VariantWireFixture>(
|
||||
CultNetDatabaseChangeKind.Removed, VariantKey, schemaId, database.Shards[0], document: null, previousDocument: cache.Get<VariantWireFixture>(VariantKey)!);
|
||||
Assert.That(Deliver(databaseServer, removed, selection, refused, errors), Is.Null,
|
||||
"no removal for a key the subscriber never held as a record");
|
||||
|
||||
// A record it did hold is still told about.
|
||||
var baseRemoved = new CultNetDatabaseChange<VariantWireFixture>(
|
||||
CultNetDatabaseChangeKind.Removed, BaseKey, schemaId, database.Shards[0], document: null, previousDocument: cache.Get<VariantWireFixture>(BaseKey)!);
|
||||
Assert.That(Deliver(databaseServer, baseRemoved, selection, refused, errors)?.ChangeKind, Is.EqualTo("removed"));
|
||||
}
|
||||
|
||||
private static async Task WaitUntilAsync(Func<bool> condition, Func<string>? diagnostic = null)
|
||||
{
|
||||
var deadline = DateTimeOffset.UtcNow + TimeSpan.FromSeconds(5);
|
||||
while (!condition())
|
||||
{
|
||||
if (DateTimeOffset.UtcNow > deadline)
|
||||
Assert.Fail("Condition was not satisfied before timeout. " + diagnostic?.Invoke());
|
||||
await Task.Delay(5);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
[CultDocument("variants.wire-fixture", "variants.wire-fixture.v1")]
|
||||
[MessagePackObject(AllowPrivate = true)]
|
||||
public sealed class VariantWireFixture
|
||||
{
|
||||
[Key(0)]
|
||||
[CultName]
|
||||
public string Name = string.Empty;
|
||||
|
||||
[Key(1)]
|
||||
public int Power;
|
||||
}
|
||||
|
||||
[CultDocument("variants.wire-citer-fixture", "variants.wire-citer-fixture.v1")]
|
||||
[MessagePackObject(AllowPrivate = true)]
|
||||
public sealed class VariantCiterFixture
|
||||
{
|
||||
[Key(0)]
|
||||
[CultName]
|
||||
public string Name = string.Empty;
|
||||
|
||||
[Key(1)]
|
||||
[CultReference(typeof(VariantWireFixture))]
|
||||
public CultRecordRef<VariantWireFixture> Ref;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -11,9 +11,11 @@ at test time. Run: python generate.py (needs msgpack).
|
|||
v1-base.msgpack NOT generated here: written once by the Python runtime at CultLib
|
||||
69a21bb (SingleFileMessagePackBackingStore.push of alpha and
|
||||
beta, type vectors.item, before any C0 code existed)
|
||||
variant-v2.msgpack what C1 writes for a variant: header v2, record slot 4 =
|
||||
[baseKey, overrides[]], empty payload (override op Set = 0,
|
||||
path steps are [slot, elementId])
|
||||
variant-v2.msgpack the wire shape of a variant record, hand-built: header v2, record slot 4 =
|
||||
[baseKey, overrides[]], empty payload (override op Set = 0, path steps
|
||||
are [slot, elementId]). C1 does not write this store and refuses to load
|
||||
it: the variant resolves to its base's name (R6). It exercises readers
|
||||
that refuse a variant by version or record.
|
||||
"""
|
||||
import pathlib
|
||||
|
||||
|
|
|
|||
3
tests/vectors/document-variants-c1/.gitattributes
vendored
Normal file
3
tests/vectors/document-variants-c1/.gitattributes
vendored
Normal file
|
|
@ -0,0 +1,3 @@
|
|||
# Refusal vectors are byte-exact; never convert line endings.
|
||||
* binary
|
||||
*.py text
|
||||
BIN
tests/vectors/document-variants-c1/variant-store.msgpack
Normal file
BIN
tests/vectors/document-variants-c1/variant-store.msgpack
Normal file
Binary file not shown.
Loading…
Reference in a new issue