diff --git a/contracts/cultnet/cultnet-distributed-database.md b/contracts/cultnet/cultnet-distributed-database.md index 932ba861..1ddc80e8 100644 --- a/contracts/cultnet/cultnet-distributed-database.md +++ b/contracts/cultnet/cultnet-distributed-database.md @@ -140,6 +140,12 @@ A malformed put or delete is `message_invalid`, not a per-operation code: the pe - A shard snapshot represents a log sequence. Applying it replaces the local shard view and advances the replica cursor to that sequence. - Realtime subscriptions publish domain changes, not storage envelopes. +- A subscription's `database_change_raw` messages may carry `commitId` and `commitCount`: the changes of one cache + commit that this subscription delivers (`commitCount` of them, at least two). Both fields are optional and additive; a + peer that does not know them ignores them and applies each change alone. A replica that knows them applies the + commit's changes as one commit, so the replica judges the result (a swap of two unique-index values) and not the state + between records. A replica whose cache refuses a commit is faulted, never silent. `commitId` is opaque and valid only + within the subscription. The C# subscription server and database server tag commits; other runtimes adopt next. - Client-owned input documents may be predicted locally only inside explicit `CultNetClientAuthorityScope` declarations. - Predicted input state is not authoritative shared state. The shard log still diff --git a/contracts/cultnet/cultnet.database-change-raw.schema.json b/contracts/cultnet/cultnet.database-change-raw.schema.json index 2725a5af..d214ed6b 100644 --- a/contracts/cultnet/cultnet.database-change-raw.schema.json +++ b/contracts/cultnet/cultnet.database-change-raw.schema.json @@ -12,7 +12,9 @@ "changeKind": { "enum": ["added", "updated", "removed"] }, "document": { "$ref": "./cultnet.raw-document-record.schema.json" }, "recordKey": { "type": "string", "minLength": 1 }, - "schemaId": { "type": "string", "minLength": 1 } + "schemaId": { "type": "string", "minLength": 1 }, + "commitId": { "type": "string", "minLength": 1 }, + "commitCount": { "type": "integer", "minimum": 2 } }, "allOf": [ { diff --git a/docs/start/README.md b/docs/start/README.md index e033c935..c388f8b6 100644 --- a/docs/start/README.md +++ b/docs/start/README.md @@ -44,7 +44,7 @@ using MessagePack; public sealed class Item { [Key(0)] [CultName] public string Name = string.Empty; - [Key(1)] [CultIndex] public string Category = string.Empty; + [Key(1)] public string Category = string.Empty; [Key(2)] public int Value; } diff --git a/docs/studio-grouping-cut.md b/docs/studio-grouping-cut.md index 770d0698..7bc8418c 100644 --- a/docs/studio-grouping-cut.md +++ b/docs/studio-grouping-cut.md @@ -33,9 +33,9 @@ Recovered from the legacy "Database Tools" window (Aetheria `d3db1730`, **The attribute is not `[CultIndex]`.** `[CultIndex]` means unique (operator, 2026-09-30; `CultDocumentContracts.cs:34`). Grouping is `CultInspectorGroupByAttribute`, in `CultInspectorAttributes.cs`, read only by `CultInspectorModel`. It changes no schema id, -catalog entry, index, or store byte. The index cut's deletion of grouping-only `[CultIndex]` -declarations (CultMesh, Geometry) adds no `[CultInspectorGroupBy]` in their place: those are -service stores, not Studio-authored catalogs. +catalog entry, index, or store byte. The index cut converts the grouping and selection `[CultIndex]` +declarations (CultMesh, Geometry) to `[CultRole]` rather than deleting them, and adds no +`[CultInspectorGroupBy]` in their place: those are service stores, not Studio-authored catalogs. ### Invariants diff --git a/src/GameCult.Caching/Contracts/cultcache-store-composition.md b/src/GameCult.Caching/Contracts/cultcache-store-composition.md index 1a03d4b1..01f7a5c9 100644 --- a/src/GameCult.Caching/Contracts/cultcache-store-composition.md +++ b/src/GameCult.Caching/Contracts/cultcache-store-composition.md @@ -110,7 +110,10 @@ which route wrote it. already agree when publication starts. - An admission runs a record's `[CultName]` and `[CultIndex]` getters once per landing record, before the store is written and before any `Sequence` is - minted. A conditional commit on a clean file store is judged a second time + minted. An admission is refused when it would leave two records of one type + holding one `[CultIndex]` value (blank is no value), and a load of a store + holding such a pair refuses naming both keys and the index. Concurrent writers on a + directory store are not covered by this rule. A conditional commit on a clean file store is judged a second time against the file it merges onto whenever the cache holds variants in any store, the batch lands a variant, or the file holds one. That second read runs the getters again for the records the batch lands, for any variant it re-resolves, diff --git a/src/GameCult.Caching/CultCache.cs b/src/GameCult.Caching/CultCache.cs index 5a0626a6..709fc64d 100644 --- a/src/GameCult.Caching/CultCache.cs +++ b/src/GameCult.Caching/CultCache.cs @@ -885,6 +885,10 @@ namespace GameCult.Caching return indexes.BySchemaId.ContainsKey(schemaId) || indexes.ByCompatibleSchemaId.ContainsKey(schemaId); } + // Whether the type registered under this schema id declares a [CultIndex]: its records are written through a cache, which judges uniqueness. + internal bool HoldsUniqueIndex(string schemaId) => + _indexes.BySchemaId.TryGetValue(schemaId, out var descriptor) && descriptor.IndexAccessors.Count > 0; + // A persisted entry a local entry reads whole: identical content, or compatible shapes that leave no persisted slot ignored. internal static bool ReadsWhole(CultSchemaCatalogEntry persisted, CultSchemaCatalogEntry local) => string.Equals(persisted.ContentHash, local.ContentHash, StringComparison.Ordinal) || @@ -1006,10 +1010,10 @@ namespace GameCult.Caching throw new InvalidOperationException(ids.Problems[0]); var nameMember = members.FirstOrDefault(member => member.IsName); var indexAccessors = members - .Where(member => member.IndexAlias != null) + .Where(member => member.IsUnique) .ToDictionary( member => member.IndexAlias!, - member => member.Getter, + member => member.IndexValue, StringComparer.Ordinal); var descriptorMembers = members .Select(member => new CultDocumentMemberDescriptor @@ -1481,6 +1485,13 @@ namespace GameCult.Caching rejections.Add(HiddenMemberMessage(type.Name, Qualified(pair[0].Member), Qualified(pair[1].Member))); } + // A member names its role once: [CultIndex] already is a role. + foreach (var entry in keyed.OrderBy(entry => entry.Member.Name, StringComparer.Ordinal)) + { + if (entry.Member.GetCustomAttribute() != null && entry.Member.GetCustomAttribute() != null) + rejections.Add($"Cult document {type.Name} member {Qualified(entry.Member)} declares both [CultIndex] and [CultRole]; [CultIndex] already names its role."); + } + // D10: two differently-named members that independently declare the same index alias. A // hiding member sharing one slot is already caught above; this is two distinct members whose // alias map entry (BuildDescriptor's indexAccessors) would otherwise collide silently. @@ -1580,14 +1591,19 @@ namespace GameCult.Caching private static string Qualified(MemberInfo member) => member.DeclaringType!.Name.Split('`')[0] + "." + member.Name; + // The member's role alias, from [CultIndex] or [CultRole] (the member name when the alias is blank). // Shared by DiscoverMembers' D10/D11 rejections and PersistedMember.FromMember, so registration // refuses exactly the shapes the accessor it builds afterward could not have walked anyway. private static string? ResolveIndexAlias(MemberInfo member) { - var indexAttribute = member.GetCustomAttribute(); - if (indexAttribute == null) + string? alias; + if (member.GetCustomAttribute() is { } index) + alias = index.Alias; + else if (member.GetCustomAttribute() is { } role) + alias = role.Alias; + else return null; - return string.IsNullOrWhiteSpace(indexAttribute.Alias) ? member.Name : indexAttribute.Alias; + return string.IsNullOrWhiteSpace(alias) ? member.Name : alias; } private static bool IsCultRecordRefType(Type type) => @@ -1854,6 +1870,7 @@ namespace GameCult.Caching public int Slot { get; set; } public bool IsName { get; set; } public string? IndexAlias { get; set; } + public bool IsUnique { get; set; } public bool IsReference { get; set; } public bool IsMany { get; set; } public string? TargetSchemaName { get; set; } @@ -1866,6 +1883,10 @@ namespace GameCult.Caching public Func Getter { get; set; } = default!; public Func GetterNullable { get; set; } = default!; + // What a [CultIndex] value is: culture-invariant, and a number is its canonical decimal (1.0m and 1.00m are one value, + // 0.0 and -0.0 are one value), so the value that decides a refusal is the value a lookup names. + public Func IndexValue { get; set; } = default!; + // D8/Q-J (revised 2026-09-22): non-null exactly for the closed CLR numeric set (Nullable // unwrapped), rendered as the member's exact canonical decimal string - never a double, so // no CLR numeric type's range or precision is lost rendering it. docs/cultnet-selection-cut.md @@ -1896,6 +1917,7 @@ namespace GameCult.Caching Slot = slot, IsName = member.GetCustomAttribute() != null, IndexAlias = ResolveIndexAlias(member), + IsUnique = member.GetCustomAttribute() != null, IsReference = isReference, IsMany = many, TargetSchemaName = targetSchemaName, @@ -1903,6 +1925,13 @@ namespace GameCult.Caching IsNumeric = IsNumericClrType(memberType), Getter = document => getValue(document)?.ToString() ?? string.Empty, GetterNullable = document => getValue(document)?.ToString(), + IndexValue = document => getValue(document) switch + { + null => string.Empty, + var raw when IsNumericClrType(raw.GetType()) => RenderCanonicalNumber(raw) ?? string.Empty, + IFormattable formattable => formattable.ToString(null, CultureInfo.InvariantCulture), + var raw => raw.ToString() ?? string.Empty + }, NumberOf = document => { var raw = getValue(document); @@ -2200,8 +2229,8 @@ namespace GameCult.Caching private readonly List<(CacheBackingStore Store, Type[] Homes)> _stores = new(); private readonly Dictionary _entries = new(StringComparer.Ordinal); private readonly Dictionary> _names = 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>> _indexes = new(); + // One key per value: RefuseIndexSharing decides it, Apply only records it. + private readonly Dictionary<(Type Type, string Alias), Dictionary> _indexes = new(); private readonly Dictionary _globals = new(); private readonly CultCodec? _codec; // Keys holding a variant: derived from the entries by Apply, never decided anywhere else. @@ -2671,16 +2700,12 @@ namespace GameCult.Caching .SelectMany(pair => pair.Value.TryGetValue(name, out var key) ? new[] { key } : Array.Empty())); } - // 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. + // A [CultIndex] value names one record of its type; blank values are not indexed. public T? GetByIndex(string alias, string value) where T : class { return Single($"index {alias}='{value}'", () => _indexes .Where(pair => pair.Key.Alias == alias && typeof(T).IsAssignableFrom(pair.Key.Type)) - .SelectMany(pair => pair.Value.TryGetValue(value, out var keys) - ? new[] { keys.Aggregate((best, next) => string.CompareOrdinal(next, best) < 0 ? next : best) } - : Array.Empty())); + .SelectMany(pair => pair.Value.TryGetValue(value, out var key) ? new[] { key } : Array.Empty())); } public void Dispose() @@ -3270,12 +3295,10 @@ namespace GameCult.Caching { var plan = ResolveVariants(admitted, evicted, source); Project(plan); - // The variant rules are judged only when a variant is involved. + RefuseIndexSharing(plan, evicted); + // The variant name rule is judged only when a variant is involved. if (_variantKeys.Count != 0 || admitted.Any(stored => stored.Variant != null)) - { RefuseVariantNameSharing(plan); - RefuseVariantIndexSharing(plan, evicted); - } return plan; } @@ -3472,62 +3495,41 @@ namespace GameCult.Caching } } - // 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, IReadOnlyList evicted) + // The one index rule, judged on the commit's result whatever the write order: within one document type and one + // [CultIndex] alias, no two records hold one non-blank value. The holders of a value are the held key (unless this + // admission replaces or evicts it) plus every landing key projecting it; two distinct holders refuse, naming both + // keys and the index and never the value. One pass over what lands, one lookup per (index, value). + private void RefuseIndexSharing(VariantPlan plan, IReadOnlyList evicted) { var landing = plan.Admitted.Concat(plan.Dependents).ToList(); if (landing.All(stored => stored.Descriptor.IndexAccessors.Count == 0)) return; var replaced = new HashSet( plan.Admitted.Concat(evicted).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>(); + var schemaNames = landing.Select(stored => stored.Descriptor).GroupBy(descriptor => descriptor.DocumentType) + .ToDictionary(group => group.Key, group => group.First().SchemaName); foreach (var stored in landing) { foreach (var (alias, value) in plan.Projections[stored.Key.Value].Indexes) { - var index = (stored.Descriptor.DocumentType, alias); - landedValues.Add((stored, index, value)); - if (!landed.TryGetValue((index, value), out var landedKeys)) - landed[(index, value)] = landedKeys = new List(); - landedKeys.Add(stored.Key.Value); + var slot = ((stored.Descriptor.DocumentType, alias), value); + if (!landed.TryGetValue(slot, out var keys)) + landed[slot] = keys = new List(); + keys.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) + foreach (var ((index, value), landedKeys) in landed.OrderBy(pair => pair.Key.Index.Alias, StringComparer.Ordinal)) { - 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()) - .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) + var holders = landedKeys.AsEnumerable(); + if (_indexes.TryGetValue(index, out var held) && held.TryGetValue(value, out var heldKey) && !replaced.Contains(heldKey)) + holders = holders.Append(heldKey); + var distinct = holders.Distinct(StringComparer.Ordinal).OrderBy(key => key, StringComparer.Ordinal).Take(2).ToArray(); + if (distinct.Length > 1) 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."); + $"Records {distinct[0]} and {distinct[1]} ({schemaNames[index.Type]}) " + + $"both hold unique index '{index.Alias}'; an index value names one record."); } } @@ -3633,7 +3635,7 @@ namespace GameCult.Caching if (projection.Name != null) MapOf(_names, type)[projection.Name] = stored.Key.Value; foreach (var (alias, value) in projection.Indexes) - AddHolder(_indexes, (type, alias), value, stored.Key.Value); + MapOf(_indexes, (type, alias))[value] = stored.Key.Value; } // Documents are mutable, so a name or index value may have changed since it was indexed: drop by key. @@ -3645,26 +3647,7 @@ 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)) - RemoveHolder(pair.Value, stored.Key.Value); - } - - private static void AddHolder(Dictionary<(Type Type, string Alias), Dictionary>> indexes, (Type, string) index, string value, string key) - { - if (!indexes.TryGetValue(index, out var map)) - indexes[index] = map = new Dictionary>(StringComparer.Ordinal); - if (!map.TryGetValue(value, out var keys)) - map[value] = keys = new List(); - keys.Remove(key); - keys.Add(key); - } - - private static void RemoveHolder(Dictionary> map, string key) - { - foreach (var pair in map.ToArray()) - { - if (pair.Value.Remove(key) && pair.Value.Count == 0) - map.Remove(pair.Key); - } + RemoveKey(pair.Value, stored.Key.Value); } private static Dictionary MapOf(Dictionary> maps, TKey key) @@ -4648,13 +4631,17 @@ namespace GameCult.Caching /// Whether the record's payload holds element ids, or null when the caller cannot tell. /// The registry the payload was encoded with. Given, the write decides the header for copied records that registry can read. It never decides ownership: the answer is the same lent or not. alone owns the record at the key when it declares the stored id (its own or one it lists compatible), or when it is the type that reads the stored entry and reads it whole, so an older id of its own schema name it reads whole is its own and is restamped. Everything else is refused with the file left as it was: a record of another registered type included, whatever the registry holds, and a record it would read only by dropping persisted slots. Retyping a key is a removal and an add. It serves this call only: the store keeps no registry. /// The call holds the store's gate while it runs. User code it runs must not enter a cache or another store; that throws . - /// The store is attached to a cache; write through the cache. + /// The store is attached to a cache; write through the cache. Also thrown when the record's type declares a , and when the catalog entry declares an index alias and the registry lent for the call does not hold the type (none, or another): the call cannot tell a unique index from a role, so such records are written through a cache. /// The file at the path is not a store this runtime can read. public void WriteRecord(CultPersistedRecord record, CultSchemaCatalogEntry catalogEntry, bool? holdsIds = null, CultDocumentRegistry? registry = null) { if (record == null) throw new ArgumentNullException(nameof(record)); if (catalogEntry == null) throw new ArgumentNullException(nameof(catalogEntry)); - Cacheless(registry, () => WriteOne(record, catalogEntry, holdsIds)); + Cacheless(registry, () => + { + RefuseUniqueIndexedCacheless(AttachedRegistry, catalogEntry); + WriteOne(record, catalogEntry, holdsIds); + }); } /// @@ -4671,7 +4658,7 @@ namespace GameCult.Caching /// The stored-at stamp, or null for now. /// The registry that describes and encodes the document, or null for . The document type's entry is taken from it, and the registry decides the header for copied records it can read. It never decides ownership, so the answer is the same for any registry that holds the type: the document's entry owns a record whose id it declares (its own or one it lists compatible), and a record of an older id of its own schema name that it reads whole; everything else is refused, another registered type's record included, and a record it would read only by dropping persisted slots. Retyping a key is a removal and an add. It serves this call only: the store keeps no registry. /// The call holds the store's gate while it runs. User code it runs (property getters while the document is encoded, constructors while a copied record is read) must not enter a cache or another store; that throws , carried by the encoder's wrapper when the code ran inside it. - /// The store is attached to a cache; write through the cache. + /// The store is attached to a cache; write through the cache. Also thrown when the record's type declares a , and when the catalog entry declares an index alias and the registry lent for the call does not hold the type (none, or another): the call cannot tell a unique index from a role, so such records are written through a cache. /// The file at the path is not a store this runtime can read. /// The document holds an element id or list a cache write refuses. public void WriteDocument(CultRecordKey key, TDocument document, string? storedAt = null, CultDocumentRegistry? registry = null) @@ -4680,6 +4667,7 @@ namespace GameCult.Caching if (document == null) throw new ArgumentNullException(nameof(document)); var resolvedRegistry = registry ?? CultDocumentRegistry.Shared; var catalogEntry = resolvedRegistry.GetRequired().ToCatalogEntry(); + RefuseUniqueIndexedCacheless(resolvedRegistry, catalogEntry); Cacheless(resolvedRegistry, () => { var ids = CultElementIds.Plan(typeof(TDocument), document, key.Value, deterministic: false); @@ -4705,6 +4693,19 @@ namespace GameCult.Caching }); } + // A unique index is judged against every held record, which a cacheless write does not hold. A write is refused when the + // registry lent for the call holds a type with a [CultIndex] under the schema id, and also when the entry declares an index + // alias the call cannot classify (no registry, or one not holding the type): the catalog does not say unique from role. + private static void RefuseUniqueIndexedCacheless(CultDocumentRegistry? registry, CultSchemaCatalogEntry entry) + { + var unique = registry != null && registry.HoldsUniqueIndex(entry.SchemaId); + var unclassified = entry.Members.Any(member => member.IndexAlias != null) && (registry == null || !registry.Declared(entry.SchemaId)); + if (unique || unclassified) + throw new InvalidOperationException( + "A record of a type with a unique index is written through a cache; a cacheless write cannot judge uniqueness" + + (unclassified ? " and the registry lent for the call does not say whether this schema's index is unique." : ".")); + } + private void WriteOne(CultPersistedRecord record, CultSchemaCatalogEntry catalogEntry, bool? holdsIds) { using (AcquireLock(wait: true)) @@ -4727,7 +4728,7 @@ namespace GameCult.Caching RefuseUndeclaredOnDisk(durable, upserts, removals, checks); if (upserts.Count == 0 && removals.Count == 0 && (checks.Any() || StagedChecks.Count > 0)) return Array.Empty<(CultStagedRecord, string)>(); - RefuseWriteIntoMovedVariantStore(durable, upserts, removals); + RefuseWriteIntoMovedStore(durable, upserts, removals); var durableRecords = durable?.Snapshot.Records ?? Array.Empty(); // A record written over one the store holds is stored at a storedAt later than the one it replaces, whoever wrote that. var replaced = durableRecords.ToDictionary(record => record.Key, record => record.StoredAt, StringComparer.Ordinal); @@ -4812,10 +4813,11 @@ namespace GameCult.Caching return false; } - private void RefuseWriteIntoMovedVariantStore(Durable? durable, IReadOnlyCollection upserts, IReadOnlyCollection removals) + private void RefuseWriteIntoMovedStore(Durable? durable, IReadOnlyCollection upserts, IReadOnlyCollection removals) { var records = durable?.Snapshot.Records ?? Array.Empty(); - if (records.All(record => record.Variant == null) && upserts.All(staged => staged.Record.Variant == null)) + if (records.All(record => record.Variant == null) && upserts.All(staged => staged.Record.Variant == null) && + !upserts.Any(staged => AttachedRegistry?.HoldsUniqueIndex(staged.Entry.SchemaId) == true)) return; var changed = new HashSet(StringComparer.Ordinal); var versions = records.ToDictionary(record => record.Key, record => durable!.Version(record.Key), StringComparer.Ordinal); @@ -4852,9 +4854,9 @@ namespace GameCult.Caching }, inner => new CultWriteConflictException( Cache == null - ? $"{this} holds a variant and was written by a writer with no CultCache, which cannot resolve variants; nothing was written. " + + ? $"{this} holds a variant or a unique index and was written by a writer with no CultCache, which cannot resolve variants; nothing was written. " + $"Records it would have written or removed: {string.Join(", ", recordKeys.Select(key => $"'{key}'"))}; records that differ from what the writer has read: {string.Join(", ", changedKeys.Select(key => $"'{key}'"))}." - : $"{this} changed since this cache last read it, and it holds a variant, which resolves against other records; nothing was written. " + + : $"{this} changed since this cache last read it, and it holds a variant or a record with a unique index, which resolve against other records; nothing was written. " + $"Records this write would have written or removed: {string.Join(", ", recordKeys.Select(key => $"'{key}'"))}; records that changed: {string.Join(", ", changedKeys.Select(key => $"'{key}'"))}. " + (inner == null ? "The cache reloaded them; retry the write against what the store holds now." diff --git a/src/GameCult.Caching/CultDocumentContracts.cs b/src/GameCult.Caching/CultDocumentContracts.cs index b439ac9f..0f8235ce 100644 --- a/src/GameCult.Caching/CultDocumentContracts.cs +++ b/src/GameCult.Caching/CultDocumentContracts.cs @@ -38,6 +38,7 @@ namespace GameCult.Caching { } + // A unique lookup: within one document type, a non-blank value names one record. Also filterable by CultNet selection. [AttributeUsage(AttributeTargets.Field | AttributeTargets.Property, Inherited = true, AllowMultiple = false)] public sealed class CultIndexAttribute : Attribute { @@ -49,6 +50,19 @@ namespace GameCult.Caching public string? Alias { get; } } + // Names the member's selection and reference role (the alias, or the member name). Implies nothing about uniqueness: + // many records may hold one value, and a role member is never indexed for lookup. + [AttributeUsage(AttributeTargets.Field | AttributeTargets.Property, Inherited = true, AllowMultiple = false)] + public sealed class CultRoleAttribute : Attribute + { + public CultRoleAttribute(string? alias = null) + { + Alias = alias; + } + + public string? Alias { get; } + } + [AttributeUsage(AttributeTargets.Field | AttributeTargets.Property, Inherited = true)] public sealed class CultReferenceAttribute : Attribute { diff --git a/src/GameCult.Caching/README.md b/src/GameCult.Caching/README.md index fc5cd308..471361dd 100644 --- a/src/GameCult.Caching/README.md +++ b/src/GameCult.Caching/README.md @@ -22,9 +22,12 @@ public sealed class ItemData [Key(1)] [CultIndex] - public string Category = string.Empty; + public string Sku = string.Empty; [Key(2)] + public string Category = string.Empty; + + [Key(3)] public int Value; } ``` @@ -33,7 +36,8 @@ Available cache attributes: - `CultDocument(schemaName, schemaVersion)` - `CultName` -- `CultIndex(alias?)` +- `CultIndex(alias?)`: a unique lookup. Within one document type, a non-blank value names one record; a commit that would leave two records holding one is refused, and a store holding such a pair refuses to load. Uniqueness is judged by a `CultCache` over the single-file store; a cacheless `WriteDocument`/`WriteRecord` of such a type is refused (also when the catalog declares an index alias the call cannot classify), and concurrent writers on a directory store are not covered (`unique-index-directory-store`). A value is culture-invariant, and a number is its canonical decimal (`1.0m` and `1.00m`, `0.0` and `-0.0` are one value). +- `CultRole(alias?)`: a selection and reference role. Many records may share a value; it is never a lookup. - `CultGlobal` - `CultReference(targetType?, many: false)` @@ -53,6 +57,7 @@ var cache = new CultCache(); var handle = await cache.AddAsync(new ItemData { Name = "Potion", + Sku = "POT-1", Category = "Consumable", Value = 50 }); @@ -66,7 +71,7 @@ Name and index lookups are driven from attributes, not manual registration. ```csharp var byName = cache.GetByName("Potion"); -var byIndex = cache.GetByIndex("Category", "Consumable"); +var byIndex = cache.GetByIndex("Sku", "POT-1"); ``` ### Explicit references @@ -130,6 +135,7 @@ var cache = await CultCacheMessagePack.OpenAsync("Data.msgpack"); var handle = await cache.UpsertAsync(new ItemData { Name = "Potion", + Sku = "POT-1", Category = "Consumable", Value = 50 }); diff --git a/src/GameCult.Geometry/CultGeometryDocuments.cs b/src/GameCult.Geometry/CultGeometryDocuments.cs index 169b9ae1..6756d39a 100644 --- a/src/GameCult.Geometry/CultGeometryDocuments.cs +++ b/src/GameCult.Geometry/CultGeometryDocuments.cs @@ -32,7 +32,7 @@ namespace GameCult.Geometry public string DomainId { get; set; } = string.Empty; [Key(1)] - [CultIndex] + [CultRole] public string RootKey { get; set; } = string.Empty; [Key(2)] @@ -164,7 +164,7 @@ namespace GameCult.Geometry public CultRecordRef DomainKey { get; set; } [Key(2)] - [CultIndex] + [CultRole] public string WorkerGroup { get; set; } = string.Empty; [Key(3)] @@ -326,7 +326,7 @@ namespace GameCult.Geometry public CultRecordRef CutKey { get; set; } [Key(2)] - [CultIndex] + [CultRole] public string SelectedCutId { get; set; } = string.Empty; [Key(3)] diff --git a/src/GameCult.Mesh/CultMeshBodyPublications.cs b/src/GameCult.Mesh/CultMeshBodyPublications.cs index 2358c277..86e49365 100644 --- a/src/GameCult.Mesh/CultMeshBodyPublications.cs +++ b/src/GameCult.Mesh/CultMeshBodyPublications.cs @@ -34,7 +34,7 @@ namespace GameCult.Mesh public sealed class CultMeshBodyPublicationDocument { [Key(0), CultName] public string BodyId { get; set; } = string.Empty; - [Key(1), CultIndex] public string ProducerId { get; set; } = string.Empty; + [Key(1), CultRole] public string ProducerId { get; set; } = string.Empty; [Key(2)] public string SchemaId { get; set; } = string.Empty; [Key(3)] public int LayoutVersion { get; set; } [Key(4)] public long ByteSize { get; set; } diff --git a/src/GameCult.Mesh/CultMeshCdn.cs b/src/GameCult.Mesh/CultMeshCdn.cs index 7f47f291..25650f1c 100644 --- a/src/GameCult.Mesh/CultMeshCdn.cs +++ b/src/GameCult.Mesh/CultMeshCdn.cs @@ -142,17 +142,17 @@ namespace GameCult.Mesh /// Artifact kind, such as asset, build, or package. [Key(1)] - [CultIndex] + [CultRole] public string Kind { get; set; } = CultMeshCdnArtifactKinds.Asset; /// Caller-defined artifact version. [Key(2)] - [CultIndex] + [CultRole] public string Version { get; set; } = string.Empty; /// SHA-256 hash of the full materialized artifact. [Key(3)] - [CultIndex] + [CultRole] public string ContentHash { get; set; } = string.Empty; /// Full materialized artifact size. diff --git a/src/GameCult.Mesh/CultMeshEntityPrefabs.cs b/src/GameCult.Mesh/CultMeshEntityPrefabs.cs index afb0b0bf..8d3f26ef 100644 --- a/src/GameCult.Mesh/CultMeshEntityPrefabs.cs +++ b/src/GameCult.Mesh/CultMeshEntityPrefabs.cs @@ -193,12 +193,12 @@ namespace GameCult.Mesh /// Caller-defined package version. [Key(1)] - [CultIndex] + [CultRole] public string Version { get; set; } = string.Empty; /// Tool that authored the package, such as Brokkr Blender. [Key(2)] - [CultIndex] + [CultRole] public string SourceTool { get; set; } = string.Empty; /// Source scene name or path. @@ -207,12 +207,12 @@ namespace GameCult.Mesh /// Source Blender collection name. [Key(4)] - [CultIndex] + [CultRole] public string SourceCollection { get; set; } = string.Empty; /// Stable hash of graph structure and referenced assets. [Key(5)] - [CultIndex] + [CultRole] public string ContentHash { get; set; } = string.Empty; /// Package creation timestamp in round-trip UTC format. diff --git a/src/GameCult.Mesh/CultMeshSimulationFacts.cs b/src/GameCult.Mesh/CultMeshSimulationFacts.cs index 11c7d296..562be44a 100644 --- a/src/GameCult.Mesh/CultMeshSimulationFacts.cs +++ b/src/GameCult.Mesh/CultMeshSimulationFacts.cs @@ -20,14 +20,14 @@ namespace GameCult.Mesh /// Stable fact identifier. /// [Key(0)] - [CultIndex] + [CultRole] public string FactId { get; set; } = string.Empty; /// /// Shard containing the simulation. /// [Key(1)] - [CultIndex] + [CultRole] public string ShardId { get; set; } = string.Empty; /// @@ -40,21 +40,21 @@ namespace GameCult.Mesh /// Simulation frame or tick. /// [Key(3)] - [CultIndex] + [CultRole] public long Frame { get; set; } /// /// Subject the fact is about. /// [Key(4)] - [CultIndex] + [CultRole] public string SubjectId { get; set; } = string.Empty; /// /// Fact kind, such as hit, collision, or ownership. /// [Key(5)] - [CultIndex] + [CultRole] public string ClaimKind { get; set; } = string.Empty; /// diff --git a/src/GameCult.Mesh/docs/public-api.md b/src/GameCult.Mesh/docs/public-api.md index 072e6c73..31737e9c 100644 --- a/src/GameCult.Mesh/docs/public-api.md +++ b/src/GameCult.Mesh/docs/public-api.md @@ -321,15 +321,16 @@ var bySchema = documents.DocumentBySchema("gamecult.aetheria.station_refit.v1"); Typed collections use the same one-call style: ```csharp -var allies = CultMesh.CollectionByIndex( +// A [CultIndex] value is unique within its type, so the handle holds at most one record. +var flagship = CultMesh.CollectionByIndex( node, - "Faction", - "au"); + "CallSign", + "au-1"); -var currentAllies = await allies.LatestAsync(); -using var changes = allies.WatchChanges(change => UpdateRoster(change)); +var currentFlagship = await flagship.LatestAsync(); +using var changes = flagship.WatchChanges(change => UpdateRoster(change)); -var collections = CultMesh.Collections(allies); +var collections = CultMesh.Collections(flagship); var roster = await collections.LatestAsync(); using var rosterChanges = collections.WatchChanges(RenderRoster); ``` diff --git a/src/GameCult.Networking/CultNetDatabase.cs b/src/GameCult.Networking/CultNetDatabase.cs index af561ea5..9db3cd1b 100644 --- a/src/GameCult.Networking/CultNetDatabase.cs +++ b/src/GameCult.Networking/CultNetDatabase.cs @@ -315,6 +315,22 @@ namespace GameCult.Networking public CultNetSelectionCursorKey? CursorKey { get; set; } } + /// + /// The typed changes ( boxed) one cache commit made, in the order the cache admitted them. + /// is opaque and unique among the commits of one database instance. + /// + public sealed class CultNetDatabaseCommit + { + public CultNetDatabaseCommit(string id, IReadOnlyList changes) + { + Id = id; + Changes = changes; + } + + public string Id { get; } + public IReadOnlyList Changes { get; } + } + /// /// A typed domain change emitted by a CultNet database surface. /// @@ -587,6 +603,7 @@ namespace GameCult.Networking private readonly Dictionary<(string SchemaId, string RecordKey), long> _lastWriteSequence = new(); private readonly Subject _changes = new(); + private readonly Subject _commits = new(); private readonly IDisposable _cacheChanges; private readonly IDisposable _cacheJournal; // What a write door of this database knows about the change it is admitting and the cache does not: the wire @@ -1148,6 +1165,17 @@ namespace GameCult.Networking return _changes; } + /// + /// Watches changes as the cache commits that made them: one event per commit, after every change of it was published to + /// . A consumer that forwards changes to a replica forwards a commit as a unit, so the replica + /// can judge the result of the commit and not each record's intermediate state. + /// + public Observable WatchCommits() + { + ThrowIfDisposed(); + return _commits; + } + /// /// Watches one record. /// @@ -1215,6 +1243,7 @@ namespace GameCult.Networking } _changes.Dispose(); + _commits.Dispose(); } // The cache is the one owner of "a change was committed"; the journal is this database's one decision point and its @@ -1231,6 +1260,8 @@ namespace GameCult.Networking } List? failures = null; + // One journal call is one admission, so one commit; its first Sequence names it. + var commit = new CommitGroup(changes.Count == 0 ? "0" : changes[0].Sequence.ToString(System.Globalization.CultureInfo.InvariantCulture)); List? burned = null; List? burnCauses = null; foreach (var change in changes) @@ -1288,7 +1319,10 @@ namespace GameCult.Networking } lock (_logGate) - _stash[change.Sequence] = new Publication(documentType, kind, change.Key, descriptor.SchemaId, shard, change.Document, change.PreviousDocument); + { + _stash[change.Sequence] = new Publication(documentType, kind, change.Key, descriptor.SchemaId, shard, change.Document, change.PreviousDocument, commit); + commit.Unpublished++; + } } // Every change is stashed and will be published; the writer hears of a failed log afterwards, in one exception. @@ -1321,8 +1355,9 @@ namespace GameCult.Networking Publish(publication); } - private void Publish(Publication publication) => - PublishUntyped( + private void Publish(Publication publication) + { + var change = PublishUntyped( publication.DocumentType, publication.Kind, publication.Key, @@ -1330,6 +1365,19 @@ namespace GameCult.Networking publication.Shard, publication.Document, publication.PreviousDocument); + CultNetDatabaseCommit? finished = null; + lock (_logGate) + { + var commit = publication.Commit; + if (change != null) + commit.Published.Add(change); + if (--commit.Unpublished == 0 && commit.Published.Count > 0) + finished = new CultNetDatabaseCommit(commit.Id, commit.Published.ToArray()); + } + + if (finished != null) + _commits.OnNext(finished); + } private void RecordReplicatedEntry( CultNetShardDescriptor shard, @@ -1467,8 +1515,10 @@ namespace GameCult.Networking string schemaId, CultNetShardDescriptor shard, object? document, - object? previousDocument) + object? previousDocument, + CommitGroup commit) { + Commit = commit; DocumentType = documentType; Kind = kind; Key = key; @@ -1485,6 +1535,16 @@ namespace GameCult.Networking public CultNetShardDescriptor Shard { get; } public object? Document { get; } public object? PreviousDocument { get; } + public CommitGroup Commit { get; } + } + + // The changes of one cache admission, counted down as they are published. Guarded by _logGate. + private sealed class CommitGroup + { + public CommitGroup(string id) => Id = id; + public string Id { get; } + public int Unpublished { get; set; } + public List Published { get; } = new(); } private sealed class DoorContext @@ -1964,7 +2024,7 @@ namespace GameCult.Networking return $"{schemaId}:{key.Value}"; } - private void PublishUntyped( + private object? PublishUntyped( Type documentType, CultNetDatabaseChangeKind kind, CultRecordKey key, @@ -1987,6 +2047,8 @@ namespace GameCult.Networking { _changes.OnNext(change); } + + return change; } // No shard owns a schema and key that no descriptor matches, and nothing is written to one: every write path refuses. diff --git a/src/GameCult.Networking/CultNetDatabaseServer.cs b/src/GameCult.Networking/CultNetDatabaseServer.cs index e1694131..32f302fc 100644 --- a/src/GameCult.Networking/CultNetDatabaseServer.cs +++ b/src/GameCult.Networking/CultNetDatabaseServer.cs @@ -765,14 +765,41 @@ namespace GameCult.Networking { // Keys this subscriber only ever received as a refusal: it was never given them as records. var refused = new HashSet(StringComparer.Ordinal); - return _database.WatchAllChanges().Subscribe(change => + return _database.WatchCommits().Subscribe(commit => { - var outbound = DeliverChange(change, subscriptionId, selection, refused, error => peer.SendCultNet(error)); - if (outbound != null) - peer.SendCultNet(outbound); + foreach (var message in DeliverCommit(commit, subscriptionId, selection, refused, error => peer.SendCultNet(error))) + peer.SendCultNet(message); }); } + // One subscriber's view of one commit. Its changes go out together, tagged with the commit, so a replica applies them as one. + internal IReadOnlyList DeliverCommit( + CultNetDatabaseCommit commit, + string subscriptionId, + CultNetSelection selection, + HashSet refused, + Action sendError) + { + var outbound = new List(); + foreach (var change in commit.Changes) + { + var message = DeliverChange(change, subscriptionId, selection, refused, sendError); + if (message != null) + outbound.Add(message); + } + + if (outbound.Count > 1) + { + foreach (var message in outbound) + { + message.CommitId = commit.Id; + message.CommitCount = outbound.Count; + } + } + + return 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( diff --git a/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs b/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs index db1e2907..971a53c2 100644 --- a/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs +++ b/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs @@ -29,6 +29,14 @@ namespace GameCult.Networking private readonly ConcurrentDictionary _initialSnapshots = new(); private readonly ConcurrentDictionary _deliveryModes = new(StringComparer.Ordinal); + // Changes of a commit that has not fully arrived, by subscription and commit id. + // Guarded by its own lock: the queue fills it, Unsubscribe and Dispose evict. A commit that does not complete is bounded + // by MaxPendingCommits and by MaxBufferedChanges, and a violation faults the client. + private const int MaxPendingCommits = 64; + // Policy: the most changes the incomplete commits may announce between them, the bound on what the buffer can hold. It + // sits far above any commit a source makes; a commit that announces more than this could not be held at all. + private const int MaxBufferedChanges = 1_000_000; + private readonly Dictionary<(string SubscriptionId, string CommitId), PendingCommit> _commits = new(); private readonly object _queueLock = new(); private Task _queue = Task.CompletedTask; private bool _disposed; @@ -48,6 +56,15 @@ namespace GameCult.Networking /// Raised after a live change has been applied to the local typed cache. public event Action? Changed; + /// + /// Raised when the local cache refuses a replicated commit. The replica then differs from its source; the client applies + /// no further changes and reports until it is replaced by a new subscription. + /// + public event Action? Faulted; + + /// The refusal that stopped replication, or null while the replica is intact. + public Exception? Fault { get; private set; } + /// Gets the local reactive cache receiving replicated documents. public CultCache Cache => _cache; @@ -109,6 +126,8 @@ namespace GameCult.Networking CultNetDatabaseSubscriptionDeliveryMode deliveryMode = CultNetDatabaseSubscriptionDeliveryMode.ReplicateToCache) { if (_disposed) throw new ObjectDisposedException(nameof(CultNetDatabaseSubscriptionClient)); + if (Fault != null) + throw new InvalidOperationException("The replica refused a commit and differs from its source; subscribe with a new client.", Fault); if (!_client.Connected) throw new InvalidOperationException("CultNet schema client must be connected before subscribing."); if (string.IsNullOrWhiteSpace(subscriptionId)) throw new ArgumentException("Subscription id is required.", nameof(subscriptionId)); @@ -166,6 +185,9 @@ namespace GameCult.Networking if (_disposed) return; if (string.IsNullOrWhiteSpace(subscriptionId)) throw new ArgumentException("Subscription id is required.", nameof(subscriptionId)); _deliveryModes.TryRemove(subscriptionId, out _); + lock (_commits) + foreach (var held in _commits.Keys.Where(key => key.SubscriptionId == subscriptionId).ToArray()) + _commits.Remove(held); _client.SendCultNet(new CultNetDatabaseUnsubscribeMessage { MessageId = Guid.NewGuid().ToString("N"), @@ -181,6 +203,8 @@ namespace GameCult.Networking pending.Completion.TrySetException(new ObjectDisposedException(nameof(CultNetDatabaseSubscriptionClient))); _initialSnapshots.Clear(); _deliveryModes.Clear(); + lock (_commits) + _commits.Clear(); _client.Dispose(); } @@ -211,48 +235,111 @@ namespace GameCult.Networking private void HandleChange(CultNetDatabaseChangeRawMessage message) { - Enqueue(async () => + Enqueue(() => { var deliveryMode = _deliveryModes.TryGetValue(message.SubscriptionId, out var configured) ? configured : CultNetDatabaseSubscriptionDeliveryMode.ReplicateToCache; - if (string.Equals(message.ChangeKind, "removed", StringComparison.Ordinal)) + // Live delivery applies nothing, so there is no commit to judge; each change is reported as it arrives. + if (deliveryMode != CultNetDatabaseSubscriptionDeliveryMode.ReplicateToCache || message.CommitId == null || !(message.CommitCount > 1)) + return Apply(deliveryMode, new[] { message }); + + // The commit's other changes follow on this connection; the replica applies them as one commit. + // A straggler of a subscription that was ended, or never made, is not buffered. + if (!_deliveryModes.ContainsKey(message.SubscriptionId)) + return Task.CompletedTask; + var held = (message.SubscriptionId, message.CommitId); + PendingCommit? pending; + lock (_commits) { - if (deliveryMode == CultNetDatabaseSubscriptionDeliveryMode.ReplicateToCache) - await DeleteAsync(message).ConfigureAwait(false); - Changed?.Invoke(new CultNetReplicatedDocumentChange( - message.SubscriptionId, - "removed", - message.RecordKey ?? string.Empty, - message.SchemaId ?? string.Empty, - null)); - return; + if (Fault != null) + return Task.CompletedTask; + var count = message.CommitCount!.Value; + var refusal = !_commits.TryGetValue(held, out pending) + ? (_commits.Count >= MaxPendingCommits + ? $"more than {MaxPendingCommits} commits are incomplete at once" + : (long)count + _commits.Sum(entry => (long)entry.Value.Announced) > MaxBufferedChanges + ? $"commit '{message.CommitId}' announces {count} changes; the incomplete commits may announce {MaxBufferedChanges} between them" + : null) + : pending.Announced != count + ? $"commit '{message.CommitId}' announces {pending.Announced} and {count} changes" + : pending.Ids.Contains(message.MessageId) + ? $"commit '{message.CommitId}' delivered message '{message.MessageId}' twice" + : null; + if (refusal != null) + { + Break(new InvalidOperationException($"CultNet subscription '{message.SubscriptionId}': {refusal}.")); + return Task.CompletedTask; + } + + if (!_commits.TryGetValue(held, out pending)) + _commits[held] = pending = new PendingCommit(count); + pending.Ids.Add(message.MessageId); + pending.Messages.Add(message); + if (pending.Messages.Count < count) + return Task.CompletedTask; + _commits.Remove(held); } - if (message.Document == null) return; - var document = deliveryMode == CultNetDatabaseSubscriptionDeliveryMode.Live - ? _documents.DeserializeRawDocument(message.Document) - : await _documents.ApplyRawDocumentPutMessageAsync( - _cache, - new CultNetDocumentPutRawMessage - { - MessageId = message.MessageId, - Document = message.Document - }).ConfigureAwait(false); - Changed?.Invoke(new CultNetReplicatedDocumentChange( - message.SubscriptionId, - message.ChangeKind, - message.Document.RecordKey, - message.Document.SchemaId, - document)); + return Apply(deliveryMode, pending.Messages); }); } - private Task DeleteAsync(CultNetDatabaseChangeRawMessage message) + private sealed class PendingCommit { - if (string.IsNullOrWhiteSpace(message.RecordKey) || string.IsNullOrWhiteSpace(message.SchemaId)) return Task.CompletedTask; - _documents.ResolveDescriptorForSchemaId(message.SchemaId!); - _cache.Remove(new CultRecordKey(message.RecordKey!)); + public PendingCommit(int announced) => Announced = announced; + public int Announced { get; } + public List Messages { get; } = new(); + public HashSet Ids { get; } = new(StringComparer.Ordinal); + } + + // The replica no longer matches its source: say so once, drop what was waiting, and apply nothing more. + private void Break(Exception error) + { + lock (_commits) + { + if (Fault != null) + return; + Fault = error; + _commits.Clear(); + } + + Faulted?.Invoke(error); + } + + private Task Apply(CultNetDatabaseSubscriptionDeliveryMode deliveryMode, IReadOnlyList changes) + { + // A replica that refused a commit no longer matches its source. Applying later commits on top would hide that, so + // nothing more is applied: the fault is surfaced and the replica is rebuilt by subscribing again. + if (Fault != null) + return Task.CompletedTask; + IReadOnlyList applied; + try + { + applied = deliveryMode == CultNetDatabaseSubscriptionDeliveryMode.Live + ? changes.Select(change => change.Document == null ? null : _documents.DeserializeRawDocument(change.Document)).ToArray() + : _documents.ApplyRawChangesAsOneCommit(_cache, changes); + } + catch (Exception error) + { + Break(error); + return Task.CompletedTask; + } + + for (var index = 0; index < changes.Count; index++) + { + var change = changes[index]; + var removed = string.Equals(change.ChangeKind, "removed", StringComparison.Ordinal); + if (!removed && change.Document == null) + continue; + Changed?.Invoke(new CultNetReplicatedDocumentChange( + change.SubscriptionId, + change.ChangeKind, + removed ? change.RecordKey ?? string.Empty : change.Document!.RecordKey, + removed ? change.SchemaId ?? string.Empty : change.Document!.SchemaId, + removed ? null : applied[index])); + } + return Task.CompletedTask; } diff --git a/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs b/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs index 8398feba..1325f85e 100644 --- a/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs +++ b/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs @@ -300,50 +300,76 @@ namespace GameCult.Networking SubscriptionKey key) { var legacyRequest = request.ToLegacyShape(subscriptionId); - return _database.WatchAllChanges().Subscribe(change => + return _database.WatchCommits().Subscribe(commit => { lock (_lifecycleGate) { if (_disposed || !_requests.ContainsKey(key) || !_projections.TryGetValue(key, out var projection)) return; - if (_authorizeRequest?.Invoke(legacyRequest, peer) == false) - { - Reconcile(key, request); - return; - } - - // D6: cites/cited are set-dependent - a change to row B can change whether row A - // matches - so a hop-bearing selection runs the full diff-against-delivered - // reconcile on every change instead of the single-row fast path. - if (request.Selection.HasHop) - { - Reconcile(key, request); - return; - } - - var sourceRecordKey = ResolveChangeRecordKey(change); - CultNetRawDocumentRecord? matched; + // What this subscription sends for the commit is held and sent together, tagged with the commit, so the replica + // applies it as one commit and judges the result. + var outbox = _outbox = new CommitOutbox(commit.Id); try { - matched = CreateMatchedRecord(change, request, peer); + ForwardCommit(commit, request, key, projection, legacyRequest, peer, subscriptionId); } - catch (NotSupportedException refusal) + finally { - // 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; + _outbox = null; + outbox.Send(peer); } - - ApplyProjectedChange( - projection, - sourceRecordKey, - matched, - peer, - subscriptionId); } }); } + private void ForwardCommit( + CultNetDatabaseCommit commit, + SubscriptionRequest request, + SubscriptionKey key, + SubscriptionProjectionState projection, + CultNetDatabaseSubscribeMessage legacyRequest, + ICultNetSchemaServerPeer peer, + string subscriptionId) + { + if (_authorizeRequest?.Invoke(legacyRequest, peer) == false) + { + Reconcile(key, request); + return; + } + + // D6: cites/cited are set-dependent - a change to row B can change whether row A + // matches - so a hop-bearing selection runs the full diff-against-delivered + // reconcile once per commit instead of the single-row fast path. + if (request.Selection.HasHop) + { + Reconcile(key, request); + return; + } + + foreach (var change in commit.Changes) + { + var sourceRecordKey = ResolveChangeRecordKey(change); + 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)); + continue; + } + + ApplyProjectedChange( + projection, + sourceRecordKey, + matched, + peer, + subscriptionId); + } + } + private void Reconcile(SubscriptionKey key, SubscriptionRequest request) { if (!_projections.TryGetValue(key, out var projection)) return; @@ -487,12 +513,15 @@ namespace GameCult.Networking string.Equals(left.SourceAgentId, right.SourceAgentId, StringComparison.Ordinal) && string.Equals(left.SourceRole, right.SourceRole, StringComparison.Ordinal); - private static void SendUpsert( + // The commit being forwarded, under _lifecycleGate: the change messages it produces are held here, not sent one by one. + private CommitOutbox? _outbox; + + private void SendUpsert( ICultNetSchemaServerPeer peer, string subscriptionId, CultNetRawDocumentRecord document, bool added) => - peer.SendCultNet(new CultNetDatabaseChangeRawMessage + Deliver(peer, new CultNetDatabaseChangeRawMessage { MessageId = Guid.NewGuid().ToString("N"), SubscriptionId = subscriptionId, @@ -500,11 +529,11 @@ namespace GameCult.Networking Document = document }); - private static void SendRemoval( + private void SendRemoval( ICultNetSchemaServerPeer peer, string subscriptionId, CultNetRawDocumentRecord document) => - peer.SendCultNet(new CultNetDatabaseChangeRawMessage + Deliver(peer, new CultNetDatabaseChangeRawMessage { MessageId = Guid.NewGuid().ToString("N"), SubscriptionId = subscriptionId, @@ -513,6 +542,38 @@ namespace GameCult.Networking SchemaId = document.SchemaId }); + private void Deliver(ICultNetSchemaServerPeer peer, CultNetDatabaseChangeRawMessage message) + { + if (_outbox is { } outbox) + outbox.Messages.Add(message); + else + peer.SendCultNet(message); + } + + private sealed class CommitOutbox + { + private readonly string _commitId; + + public CommitOutbox(string commitId) => _commitId = commitId; + + public List Messages { get; } = new(); + + public void Send(ICultNetSchemaServerPeer peer) + { + if (Messages.Count > 1) + { + foreach (var message in Messages) + { + message.CommitId = _commitId; + message.CommitCount = Messages.Count; + } + } + + foreach (var message in Messages) + peer.SendCultNet(message); + } + } + private void Withdraw(SubscriptionKey key, bool sendRemovals, bool forgetRequest) { if (_subscriptions.TryRemove(key, out var subscription)) diff --git a/src/GameCult.Networking/CultNetDocumentRegistry.cs b/src/GameCult.Networking/CultNetDocumentRegistry.cs index 58b7ca36..19f17c8c 100644 --- a/src/GameCult.Networking/CultNetDocumentRegistry.cs +++ b/src/GameCult.Networking/CultNetDocumentRegistry.cs @@ -722,6 +722,49 @@ namespace GameCult.Networking return document; } + /// + /// Applies change messages to a cache as one commit, so the cache judges the result of the changes together and not the + /// state between them. Returns the applied document of each message, null for a removal. A refusal by the cache throws + /// and nothing is applied. + /// + internal IReadOnlyList ApplyRawChangesAsOneCommit( + CultCache cache, + IReadOnlyList changes) + { + var applied = new object?[changes.Count]; + var staged = new List<(Type? Type, object? Document, CultRecordKey Key)>(changes.Count); + for (var index = 0; index < changes.Count; index++) + { + var change = changes[index]; + if (string.Equals(change.ChangeKind, "removed", StringComparison.Ordinal)) + { + if (string.IsNullOrWhiteSpace(change.RecordKey) || string.IsNullOrWhiteSpace(change.SchemaId)) + continue; + ResolveDescriptorForSchemaId(change.SchemaId!); + staged.Add((null, null, new CultRecordKey(change.RecordKey!))); + continue; + } + + if (change.Document == null) + continue; + var document = DeserializeRawDocument(change.Document); + applied[index] = document; + staged.Add((ResolveDescriptorForRawDocument(change.Document).DocumentType, document, new CultRecordKey(change.Document.RecordKey))); + } + + cache.Commit(batch => + { + foreach (var (type, document, key) in staged) + { + if (type == null) + batch.Remove(key); + else + batch.Upsert(type, document!, key); + } + }); + return applied; + } + /// /// Decodes one validated raw document without materializing it in a CultCache replica. /// Use this for ephemeral reactive subscriptions whose callback is the consumer. @@ -754,25 +797,22 @@ namespace GameCult.Networking } /// - /// Applies all documents from a raw snapshot response. + /// Applies all documents from a raw snapshot response as one commit, so a snapshot that moves a unique-index value + /// between records is judged on its result. /// - public async Task> ApplyRawSnapshotResponseAsync( + public Task> ApplyRawSnapshotResponseAsync( CultCache cache, CultNetSnapshotResponseRawMessage response) { - var applied = new List(response.Documents.Length); - foreach (var document in response.Documents) - { - applied.Add(await ApplyRawDocumentPutMessageAsync( - cache, - new CultNetDocumentPutRawMessage - { - MessageId = response.MessageId, - Document = document - }).ConfigureAwait(false)); - } - - return applied; + var applied = ApplyRawChangesAsOneCommit( + cache, + response.Documents.Select(document => new CultNetDatabaseChangeRawMessage + { + SubscriptionId = response.MessageId, + ChangeKind = "added", + Document = document + }).ToArray()); + return Task.FromResult>(applied.Select(document => document!).ToArray()); } internal CultDocumentDescriptor ResolveDescriptorForRawDocument(CultNetRawDocumentRecord document) diff --git a/src/GameCult.Networking/CultNetSchemaMessages.cs b/src/GameCult.Networking/CultNetSchemaMessages.cs index a2211672..bc59b007 100644 --- a/src/GameCult.Networking/CultNetSchemaMessages.cs +++ b/src/GameCult.Networking/CultNetSchemaMessages.cs @@ -1179,6 +1179,17 @@ namespace GameCult.Networking /// Gets or sets the deleted schema id for removed changes. /// [Key("schemaId")] public string? SchemaId { get; set; } + /// + /// Gets or sets the identity of the cache commit this change belongs to. Optional and additive: a peer that does not + /// know the field ignores it and applies the change alone. Opaque to the receiver; equal ids on one subscription + /// name changes of one commit. + /// + [Key("commitId")] public string? CommitId { get; set; } + /// + /// Gets or sets how many change messages of this commit this subscription carries. Set with + /// when that is more than one; the receiver applies the commit once all have arrived. + /// + [Key("commitCount")] public int? CommitCount { get; set; } } /// diff --git a/tests/GameCult.Caching.Tests/CultDocumentSelectionSurfaceTests.cs b/tests/GameCult.Caching.Tests/CultDocumentSelectionSurfaceTests.cs index daa71c81..ed77bb90 100644 --- a/tests/GameCult.Caching.Tests/CultDocumentSelectionSurfaceTests.cs +++ b/tests/GameCult.Caching.Tests/CultDocumentSelectionSurfaceTests.cs @@ -226,8 +226,8 @@ namespace GameCult.Caching.Tests public void DuplicateIndexAliasOnTwoDifferentlyNamedMembersIsRefusedByName() { var type = Document("DupAlias"); - Field(type, "First", typeof(string), Key(0), IndexAttr("shared")); - Field(type, "Second", typeof(string), Key(1), IndexAttr("shared")); + Field(type, "First", typeof(string), Key(0), RoleAttr("shared")); + Field(type, "Second", typeof(string), Key(1), RoleAttr("shared")); var message = Assert.Throws(() => CultDocumentRegistry.ForTypes(new[] { type.CreateType()! }))!.Message; Assert.That(message, Does.Contain("DupAlias")); @@ -236,7 +236,7 @@ namespace GameCult.Caching.Tests Assert.That(message, Does.Contain("shared")); } - // R-AL/SM-6: a plain data member's bare name (it declares no [CultIndex], so its role in + // R-AL/SM-6: a plain data member's bare name (it declares no [CultRole], so its role in // ReferencesOf/ReferenceMembers falls back to its own MemberName) colliding with a reference // member's explicit alias. D10 above only groups members that declare an alias and misses // this; before the fix this registered without error and then every hop through the @@ -246,7 +246,7 @@ namespace GameCult.Caching.Tests { var type = Document("OwnerCollision"); Field(type, "Owner", typeof(string), Key(0)); - Field(type, "Ref", typeof(CultRecordRef), Key(1), IndexAttr("Owner")); + Field(type, "Ref", typeof(CultRecordRef), Key(1), RoleAttr("Owner")); var message = Assert.Throws(() => CultDocumentRegistry.ForTypes(new[] { type.CreateType()! }))!.Message; Assert.That(message, Does.Contain("OwnerCollision")); @@ -295,8 +295,8 @@ namespace GameCult.Caching.Tests private static CustomAttributeBuilder Key(int slot) => new(typeof(KeyAttribute).GetConstructor(new[] { typeof(int) })!, new object[] { slot }); - private static CustomAttributeBuilder IndexAttr(string alias) => - new(typeof(CultIndexAttribute).GetConstructor(new[] { typeof(string) })!, new object?[] { alias }); + private static CustomAttributeBuilder RoleAttr(string alias) => + new(typeof(CultRoleAttribute).GetConstructor(new[] { typeof(string) })!, new object?[] { alias }); private static CustomAttributeBuilder ReferenceAttr(Type? targetType, bool many) => new(typeof(CultReferenceAttribute).GetConstructor(new[] { typeof(Type), typeof(bool) })!, new object?[] { targetType, many }); @@ -308,7 +308,7 @@ namespace GameCult.Caching.Tests public string Name = string.Empty; [Key(1)] - [CultIndex("mass")] + [CultRole("mass")] public float Mass; } @@ -325,7 +325,7 @@ namespace GameCult.Caching.Tests public sealed class FixtureLeafB : FixtureMiddle { [Key(2)] - [CultIndex("weight")] + [CultRole("weight")] public decimal? Weight; } @@ -341,19 +341,19 @@ namespace GameCult.Caching.Tests public string Name = string.Empty; [Key(1)] - [CultIndex("big_integer")] + [CultRole("big_integer")] public long BigInteger; [Key(2)] - [CultIndex("trimmed_decimal")] + [CultRole("trimmed_decimal")] public decimal TrimmedDecimal; [Key(3)] - [CultIndex("wide_double")] + [CultRole("wide_double")] public double WideDouble; [Key(4)] - [CultIndex("tiny_double")] + [CultRole("tiny_double")] public double TinyDouble; } diff --git a/tests/GameCult.Caching.Tests/DocumentShapeTests.cs b/tests/GameCult.Caching.Tests/DocumentShapeTests.cs index c34834ca..0d473a5a 100644 --- a/tests/GameCult.Caching.Tests/DocumentShapeTests.cs +++ b/tests/GameCult.Caching.Tests/DocumentShapeTests.cs @@ -394,7 +394,7 @@ namespace GameCult.Caching.Tests public class Gear { [Key(0)] [CultName] public string Name { get; set; } = ""; - [Key(1)] [CultIndex("slot")] public string Slot { get; set; } = ""; + [Key(1)] [CultRole("slot")] public string Slot { get; set; } = ""; } [CultDocument("shape.weapon", "shape.weapon.v1")] diff --git a/tests/GameCult.Caching.Tests/EmittedDocumentTypes.cs b/tests/GameCult.Caching.Tests/EmittedDocumentTypes.cs index ac6a6dad..5ca50c69 100644 --- a/tests/GameCult.Caching.Tests/EmittedDocumentTypes.cs +++ b/tests/GameCult.Caching.Tests/EmittedDocumentTypes.cs @@ -12,7 +12,7 @@ namespace GameCult.Caching.Tests // two ways), an alias whose resolved type would depend on scan order, or a type a test needs to itself. internal static class EmittedDocumentTypes { - internal readonly record struct Field(string Name, Type Type, int Key, bool IsName = false); + internal readonly record struct Field(string Name, Type Type, int Key, bool IsName = false, string? Index = null, string? Role = null); internal static Type Emit(string typeName, string schemaName, string schemaVersion, Field[]? fields = null, string[]? compatibleSchemaIds = null) { @@ -37,6 +37,10 @@ namespace GameCult.Caching.Tests builder.SetCustomAttribute(new CustomAttributeBuilder(typeof(KeyAttribute).GetConstructor(new[] { typeof(int) })!, new object[] { field.Key })); if (field.IsName) builder.SetCustomAttribute(new CustomAttributeBuilder(typeof(CultNameAttribute).GetConstructor(Type.EmptyTypes)!, Array.Empty())); + if (field.Index != null) + builder.SetCustomAttribute(new CustomAttributeBuilder(typeof(CultIndexAttribute).GetConstructor(new[] { typeof(string) })!, new object[] { field.Index })); + if (field.Role != null) + builder.SetCustomAttribute(new CustomAttributeBuilder(typeof(CultRoleAttribute).GetConstructor(new[] { typeof(string) })!, new object[] { field.Role })); } return type.CreateType()!; diff --git a/tests/GameCult.Caching.Tests/FirstCallTests.cs b/tests/GameCult.Caching.Tests/FirstCallTests.cs index 8540d9d0..8fad1229 100644 --- a/tests/GameCult.Caching.Tests/FirstCallTests.cs +++ b/tests/GameCult.Caching.Tests/FirstCallTests.cs @@ -14,7 +14,7 @@ using MessagePack; public sealed class Item { [Key(0)] [CultName] public string Name = string.Empty; - [Key(1)] [CultIndex] public string Category = string.Empty; + [Key(1)] public string Category = string.Empty; [Key(2)] public int Value; } diff --git a/tests/GameCult.Caching.Tests/UniqueIndexTests.cs b/tests/GameCult.Caching.Tests/UniqueIndexTests.cs new file mode 100644 index 00000000..45817296 --- /dev/null +++ b/tests/GameCult.Caching.Tests/UniqueIndexTests.cs @@ -0,0 +1,459 @@ +#nullable enable +using System; +using System.IO; +using System.Globalization; +using System.Linq; +using System.Threading.Tasks; +using GameCult.Caching; +using GameCult.Caching.MessagePack; +using MessagePack; +using NUnit.Framework; + +namespace GameCult.Caching.Tests +{ + // variants:ruling:unique-indexes. Within one document type and one [CultIndex] alias, no two held records hold one + // non-blank value; CultCache.Resolve judges every commit and every load, and a refusal never echoes the value. + public class UniqueIndexTests + { + private const string Canary = "CANARY-7f3a"; + + private static readonly CultDocumentRegistry Registry = CultDocumentRegistry.ForTypes(new[] + { + typeof(UqItem), typeof(UqTwin), typeof(UqTagged), typeof(UqDecimal), typeof(UqDouble), typeof(UqStamp) + }); + + // Emitted, so CultDocumentRegistry.Shared never scans them (one schema declared two ways, and a member Shared would refuse). + // The same schema name, version and shape, once as a role and once as a unique index: a role store holds the pair, the index type refuses it. + private static readonly Type UqLoadRole = EmittedDocumentTypes.Emit( + "UqLoadRole", "tests.uq_load", "tests.uq_load.v1", new[] { new EmittedDocumentTypes.Field("Code", typeof(string), 0, Role: "code") }); + private static readonly Type UqLoadIndex = EmittedDocumentTypes.Emit( + "UqLoadIndex", "tests.uq_load", "tests.uq_load.v1", new[] { new EmittedDocumentTypes.Field("Code", typeof(string), 0, Index: "code") }); + private static readonly Type UqBoth = EmittedDocumentTypes.Emit( + "UqBoth", "tests.uq_both", "tests.uq_both.v1", new[] { new EmittedDocumentTypes.Field("Both", typeof(string), 0, Index: "x", Role: "x") }); + + private static object Load(Type type, string code) => EmittedDocumentTypes.New(type, ("Code", code)); + + private string _directory = string.Empty; + + [SetUp] + public void CreateDirectory() + { + _directory = Path.Combine(Path.GetTempPath(), $"cultlib-uidx-{Guid.NewGuid():N}"); + Directory.CreateDirectory(_directory); + } + + [TearDown] + public void DeleteDirectory() + { + if (Directory.Exists(_directory)) + Directory.Delete(_directory, recursive: true); + } + + private string PathOf(string name) => Path.Combine(_directory, name); + + private static CultCache Open(string path, CultDocumentRegistry? registry = null) => + CultCacheMessagePack.Create(path, new CultCacheOpenOptions { Registry = registry ?? Registry }); + + private static CultRecordKey Key(string name) => new(name); + + private static UqItem Item(string? sku) => new() { Sku = sku }; + + private static InvalidOperationException Refused(TestDelegate write) => Assert.Throws(write)!; + + [Test] + public void ABatchLeavingTwoRecordsWithOneValueIsRefusedNamingBothKeysAndTheIndex() + { + var path = PathOf("items.cc"); + using var cache = Open(path); + cache.Commit(batch => batch.Upsert(typeof(UqItem), Item("keep"), Key("seed"))); + var before = File.ReadAllBytes(path); + + var refused = Refused(() => cache.Commit(batch => + { + batch.Upsert(typeof(UqItem), Item(Canary), Key("zeta")); + batch.Upsert(typeof(UqItem), Item(Canary), Key("alpha")); + })); + + Assert.That(refused.Message, Does.Contain("alpha").And.Contain("zeta").And.Contain("sku").And.Contain("tests.uq_item")); + Assert.That(refused.Message.IndexOf("alpha", StringComparison.Ordinal), Is.LessThan(refused.Message.IndexOf("zeta", StringComparison.Ordinal)), "keys in ordinal order"); + Assert.That(refused.Message, Does.Not.Contain(Canary), "the value is never echoed"); + Assert.That(cache.Get(Key("alpha")), Is.Null); + Assert.That(cache.Get(Key("zeta")), Is.Null); + Assert.That(File.ReadAllBytes(path), Is.EqualTo(before), "the store is byte-identical"); + Assert.That(cache.GetByIndex("sku", Canary), Is.Null); + } + + [Test] + public async Task AWriteOfAHeldValueAtAnotherKeyIsRefused() + { + using var cache = Open(PathOf("items.cc")); + await cache.UpsertAsync(typeof(UqItem), Item(Canary), Key("first")); + + var single = Assert.ThrowsAsync(() => cache.UpsertAsync(typeof(UqItem), Item(Canary), Key("second")))!; + Assert.That(single.Message, Does.Contain("first").And.Contain("second").And.Contain("sku").And.Not.Contain(Canary)); + var batched = Refused(() => cache.Commit(batch => batch.Upsert(typeof(UqItem), Item(Canary), Key("third")))); + Assert.That(batched.Message, Does.Contain("first").And.Contain("third")); + + Assert.That(cache.Get(Key("second")), Is.Null); + Assert.That(cache.Get(Key("third")), Is.Null); + Assert.That(cache.GetByIndex("sku", Canary), Is.Not.Null); + + await cache.UpsertAsync(typeof(UqItem), Item(Canary), Key("first")); + } + + [Test] + public void SwappingTwoValuesInOneCommitLandsAndBothAreFound() + { + var path = PathOf("items.cc"); + using (var cache = Open(path)) + { + cache.Commit(batch => + { + batch.Upsert(typeof(UqItem), Item("one"), Key("a")); + batch.Upsert(typeof(UqItem), Item("two"), Key("b")); + }); + cache.Commit(batch => + { + batch.Upsert(typeof(UqItem), Item("two"), Key("a")); + batch.Upsert(typeof(UqItem), Item("one"), Key("b")); + }); + + Assert.That(cache.GetStored(Key("a"))!.Document, Is.SameAs(cache.GetByIndex("sku", "two"))); + Assert.That(cache.GetStored(Key("b"))!.Document, Is.SameAs(cache.GetByIndex("sku", "one"))); + } + + using var reopened = Open(path); + Assert.That(reopened.GetStored(Key("a"))!.Document, Is.SameAs(reopened.GetByIndex("sku", "two"))); + Assert.That(reopened.GetStored(Key("b"))!.Document, Is.SameAs(reopened.GetByIndex("sku", "one"))); + } + + [Test] + public void AValueMovesWhenItsHolderIsRemovedOrChangedInTheSameCommit() + { + using var cache = Open(PathOf("items.cc")); + cache.Commit(batch => + { + batch.Upsert(typeof(UqItem), Item("one"), Key("a")); + batch.Upsert(typeof(UqItem), Item("two"), Key("b")); + }); + + cache.Commit(batch => + { + batch.Remove(Key("a")); + batch.Upsert(typeof(UqItem), Item("one"), Key("c")); + }); + Assert.That(cache.GetStored(Key("c"))!.Document, Is.SameAs(cache.GetByIndex("sku", "one")), "removed holder released its value"); + + cache.Commit(batch => + { + batch.Upsert(typeof(UqItem), Item("three"), Key("b")); + batch.Upsert(typeof(UqItem), Item("two"), Key("d")); + }); + Assert.That(cache.GetStored(Key("d"))!.Document, Is.SameAs(cache.GetByIndex("sku", "two")), "changed holder released its value"); + Assert.That(cache.GetStored(Key("b"))!.Document, Is.SameAs(cache.GetByIndex("sku", "three"))); + Assert.That(cache.GetByIndex("sku", "one")!.Sku, Is.EqualTo("one")); + } + + [Test] + public void BlankValuesNeverConflict() + { + var path = PathOf("items.cc"); + using (var cache = Open(path)) + { + cache.Commit(batch => + { + batch.Upsert(typeof(UqItem), Item(null), Key("a")); + batch.Upsert(typeof(UqItem), Item(""), Key("b")); + batch.Upsert(typeof(UqItem), Item(" "), Key("c")); + batch.Upsert(typeof(UqItem), Item(null), Key("d")); + }); + Assert.That(cache.GetByIndex("sku", ""), Is.Null); + Assert.That(cache.GetByIndex("sku", " "), Is.Null); + } + + using var reopened = Open(path); + Assert.That(reopened.Get(Key("d")), Is.Not.Null, "a store of blank values reopens"); + } + + [Test] + public void TwoTypesMayShareAnAliasAndAValue() + { + using var cache = Open(PathOf("items.cc")); + cache.Commit(batch => + { + batch.Upsert(typeof(UqItem), Item("same"), Key("item")); + batch.Upsert(typeof(UqTwin), new UqTwin { Sku = "same" }, Key("twin")); + }); + + Assert.That(cache.GetByIndex("sku", "same"), Is.Not.Null); + Assert.That(cache.GetByIndex("sku", "same"), Is.Not.Null); + Assert.That(Registry.GetRequired().DeclaredMembers.Single().IndexAlias, Is.EqualTo("sku"), "a unique index addresses selection as a role does"); + } + + [Test] + public void RoleMembersMayShareValuesAndStaySelectable() + { + var path = PathOf("tagged.cc"); + using (var cache = Open(path)) + { + cache.Commit(batch => + { + batch.Upsert(typeof(UqTagged), new UqTagged { Kind = "weapon" }, Key("a")); + batch.Upsert(typeof(UqTagged), new UqTagged { Kind = "weapon" }, Key("b")); + }); + Assert.That(cache.GetByIndex("kind", "weapon"), Is.Null, "a role is not a lookup"); + } + + using var reopened = Open(path); + Assert.That(reopened.Get(Key("b")), Is.Not.Null, "a role pair reopens"); + var member = Registry.GetRequired().DeclaredMembers.Single(candidate => candidate.MemberName == nameof(UqTagged.Kind)); + Assert.That(member.IndexAlias, Is.EqualTo("kind"), "the role still supplies the selection address"); + } + + [Test] + public void AStoreHoldingDuplicatesRefusesToOpenNamingBothKeysAndTheIndex() + { + var path = PathOf("load.cc"); + var roleRegistry = CultDocumentRegistry.ForTypes(new[] { UqLoadRole }); + using (var writer = Open(path, roleRegistry)) + { + writer.Commit(batch => + { + batch.Upsert(UqLoadRole, Load(UqLoadRole, Canary), Key("m")); + batch.Upsert(UqLoadRole, Load(UqLoadRole, Canary), Key("k")); + }); + } + + var before = File.ReadAllBytes(path); + var indexRegistry = CultDocumentRegistry.ForTypes(new[] { UqLoadIndex }); + var refused = Refused(() => Open(path, indexRegistry).Dispose()); + + Assert.That(refused.Message, Does.Contain("k").And.Contain("m").And.Contain("code").And.Not.Contain(Canary)); + Assert.That(File.ReadAllBytes(path), Is.EqualTo(before), "the file is untouched"); + + using var cache = new CultCache(indexRegistry); + var store = new SingleFileMessagePackBackingStore(path); + Refused(() => cache.AddBackingStore(store)); + Assert.That(cache.BackingStores, Is.Empty, "the refused store is not attached"); + Assert.That(cache.Get(Key("k")), Is.Null); + } + + [Test] + public void AWriteOntoAMovedFileIsAConflictAndTheRetryIsRefused() + { + var path = PathOf("items.cc"); + using var seed = Open(path); + seed.Commit(batch => batch.Upsert(typeof(UqItem), Item("base"), Key("a"))); + using var second = Open(path); + using (var first = Open(path)) + first.Commit(batch => batch.Upsert(typeof(UqItem), Item(Canary), Key("b"))); + + var conflict = Assert.Throws( + () => second.Commit(batch => batch.Upsert(typeof(UqItem), Item(Canary), Key("c"))))!; + Assert.That(conflict.Message, Does.Contain("unique index")); + Assert.That(second.Get(Key("b")), Is.Not.Null, "the moved record is loaded"); + + var retry = Refused(() => second.Commit(batch => batch.Upsert(typeof(UqItem), Item(Canary), Key("c")))); + Assert.That(retry.Message, Does.Contain("b").And.Contain("c").And.Not.Contain(Canary)); + + using var reopened = Open(path); + Assert.That(reopened.Get(Key("c")), Is.Null, "the file never holds the pair"); + Assert.That(reopened.Get(Key("b")), Is.Not.Null); + } + + [Test] + public void CachelessWritesOfAUniqueIndexedTypeAreRefused() + { + var path = PathOf("items.cc"); + using (var cache = Open(path)) + cache.Commit(batch => batch.Upsert(typeof(UqItem), Item("a"), Key("a"))); + var before = File.ReadAllBytes(path); + var store = new SingleFileMessagePackBackingStore(path); + + Refused(() => store.WriteDocument(Key("x"), Item(Canary), registry: Registry)); + var entry = Registry.GetRequired().ToCatalogEntry(); + Refused(() => store.WriteRecord( + new CultPersistedRecord { Key = "y", SchemaId = entry.SchemaId, StoredAt = DateTimeOffset.UtcNow.ToString("O"), Payload = new byte[] { 0x90 } }, + entry, + registry: Registry)); + + Refused(() => store.WriteRecord( + new CultPersistedRecord { Key = "y", SchemaId = entry.SchemaId, StoredAt = DateTimeOffset.UtcNow.ToString("O"), Payload = new byte[] { 0x90 } }, + entry)); + Refused(() => store.WriteRecord( + new CultPersistedRecord { Key = "y", SchemaId = entry.SchemaId, StoredAt = DateTimeOffset.UtcNow.ToString("O"), Payload = new byte[] { 0x90 } }, + entry, + registry: CultDocumentRegistry.ForTypes(new[] { typeof(UqTagged) }))); + Assert.That(File.ReadAllBytes(path), Is.EqualTo(before), "the file is byte-identical, whichever registry (none, another) was lent"); + store.WriteDocument(Key("t"), new UqTagged { Kind = "k" }, registry: Registry); + using var reopened = Open(path); + Assert.That(reopened.Get(Key("t")), Is.Not.Null, "a role-only type still writes cacheless"); + } + + + // The rule is a function of the whole commit: it must not stop at a prefix, a mixed batch or a small load. + [TestCase(0, 1)] + [TestCase(0, 6)] + [TestCase(2, 5)] + [TestCase(5, 6)] + public void ADuplicateAnywhereInALargeMixedBatchIsRefused(int first, int second) + { + using var cache = Open(PathOf("items.cc")); + + var refused = Refused(() => cache.Commit(batch => + { + batch.Upsert(typeof(UqTagged), new UqTagged { Kind = "same" }, Key("tag-a")); + for (var i = 0; i < 7; i++) + batch.Upsert(typeof(UqItem), Item(i == first || i == second ? "dup" : "v" + i), Key("item-" + i)); + batch.Upsert(typeof(UqTagged), new UqTagged { Kind = "same" }, Key("tag-b")); + })); + + Assert.That(refused.Message, Does.Contain("item-" + first).And.Contain("item-" + second).And.Not.Contain("dup")); + Assert.That(cache.Get(Key("tag-a")), Is.Null, "nothing landed"); + } + + [Test] + public void ADuplicateBesideARecordOfAnotherTypeSharingTheAliasIsStillRefused() + { + using var cache = Open(PathOf("items.cc")); + var refused = Refused(() => cache.Commit(batch => + { + batch.Upsert(typeof(UqTwin), new UqTwin { Sku = "same" }, Key("twin")); + batch.Upsert(typeof(UqItem), Item("same"), Key("item-1")); + batch.Upsert(typeof(UqItem), Item("same"), Key("item-2")); + })); + Assert.That(refused.Message, Does.Contain("item-1").And.Contain("item-2").And.Not.Contain("twin")); + } + + [Test] + public void ValuesDifferingOnlyByCaseAreDifferentValues() + { + using var cache = Open(PathOf("items.cc")); + cache.Commit(batch => + { + batch.Upsert(typeof(UqItem), Item("Sku"), Key("a")); + batch.Upsert(typeof(UqItem), Item("sku"), Key("b")); + }); + Assert.That(cache.GetStored(Key("a"))!.Document, Is.SameAs(cache.GetByIndex("sku", "Sku"))); + Assert.That(cache.GetStored(Key("b"))!.Document, Is.SameAs(cache.GetByIndex("sku", "sku"))); + } + + [Test] + public void ALoadOfManyRecordsRefusesAPairWhereverItSits([Values(0, 3, 6)] int pairAt) + { + var path = PathOf("load.cc"); + var roleRegistry = CultDocumentRegistry.ForTypes(new[] { UqLoadRole }); + using (var writer = Open(path, roleRegistry)) + { + writer.Commit(batch => + { + for (var i = 0; i < 8; i++) + batch.Upsert(UqLoadRole, Load(UqLoadRole, i == pairAt || i == pairAt + 1 ? "dup" : "v" + i), Key("rec-" + i)); + }); + } + + var refused = Refused(() => Open(path, CultDocumentRegistry.ForTypes(new[] { UqLoadIndex })).Dispose()); + Assert.That(refused.Message, Does.Contain("rec-" + pairAt).And.Contain("rec-" + (pairAt + 1)).And.Not.Contain("dup")); + } + + [Test] + public void AMixedBatchOntoAMovedFileIsAConflictWhateverItsOrder() + { + var path = PathOf("items.cc"); + using var seed = Open(path); + seed.Commit(batch => batch.Upsert(typeof(UqItem), Item("base"), Key("a"))); + using var second = Open(path); + using (var first = Open(path)) + first.Commit(batch => batch.Upsert(typeof(UqItem), Item("moved"), Key("b"))); + + Assert.Throws(() => second.Commit(batch => + { + batch.Upsert(typeof(UqTagged), new UqTagged { Kind = "k" }, Key("tag")); + batch.Upsert(typeof(UqItem), Item("fresh"), Key("c")); + })); + } + + [Test] + public void ANumericIndexValueIsOneCanonicalValueInEveryCulture() + { + var culture = (CultureInfo)CultureInfo.InvariantCulture.Clone(); + culture.NumberFormat.NumberDecimalSeparator = ","; + var previous = CultureInfo.CurrentCulture; + CultureInfo.CurrentCulture = culture; + try + { + using var cache = Open(PathOf("numbers.cc")); + cache.Commit(batch => batch.Upsert(typeof(UqDecimal), new UqDecimal { Price = 1.0m }, Key("d1"))); + Assert.That(cache.GetByIndex("price", "1"), Is.Not.Null, "1.0m is the canonical value 1"); + var sameDecimal = Refused(() => cache.Commit(batch => batch.Upsert(typeof(UqDecimal), new UqDecimal { Price = 1.00m }, Key("d2")))); + Assert.That(sameDecimal.Message, Does.Contain("d1").And.Contain("d2")); + + cache.Commit(batch => batch.Upsert(typeof(UqDecimal), new UqDecimal { Price = 1.5m }, Key("d3"))); + Assert.That(cache.GetByIndex("price", "1.5"), Is.Not.Null, "lookups name the invariant spelling, not the current culture's"); + Assert.That(cache.GetByIndex("price", "1,5"), Is.Null); + + culture.DateTimeFormat.ShortDatePattern = "yyyy.dd.MM"; + culture.DateTimeFormat.DateSeparator = "."; + cache.Commit(batch => batch.Upsert(typeof(UqStamp), new UqStamp { When = new DateTime(2026, 1, 2, 3, 4, 5) }, Key("t1"))); + Assert.That(cache.GetByIndex("when", "01/02/2026 03:04:05"), Is.Not.Null, "a date is its invariant spelling"); + + cache.Commit(batch => batch.Upsert(typeof(UqDouble), new UqDouble { Level = 0.0 }, Key("z1"))); + var signedZero = Refused(() => cache.Commit(batch => batch.Upsert(typeof(UqDouble), new UqDouble { Level = -0.0 }, Key("z2")))); + Assert.That(signedZero.Message, Does.Contain("z1").And.Contain("z2")); + } + finally + { + CultureInfo.CurrentCulture = previous; + } + } + + [Test] + public void AMemberWithBothAttributesIsRefusedAtRegistration() + { + var refused = Refused(() => CultDocumentRegistry.ForTypes(new[] { UqBoth })); + Assert.That(refused.Message, Does.Contain("Both").And.Contain("[CultIndex]").And.Contain("[CultRole]")); + } + + [CultDocument("tests.uq_item", "tests.uq_item.v1")] + [MessagePackObject(AllowPrivate = true)] + public sealed class UqItem + { + [Key(0)] [CultIndex("sku")] public string? Sku { get; set; } + } + + [CultDocument("tests.uq_twin", "tests.uq_twin.v1")] + [MessagePackObject(AllowPrivate = true)] + public sealed class UqTwin + { + [Key(0)] [CultIndex("sku")] public string? Sku { get; set; } + } + + [CultDocument("tests.uq_decimal", "tests.uq_decimal.v1")] + [MessagePackObject(AllowPrivate = true)] + public sealed class UqDecimal + { + [Key(0)] [CultIndex("price")] public decimal Price { get; set; } + } + + [CultDocument("tests.uq_stamp", "tests.uq_stamp.v1")] + [MessagePackObject(AllowPrivate = true)] + public sealed class UqStamp + { + [Key(0)] [CultIndex("when")] public DateTime When { get; set; } + } + + [CultDocument("tests.uq_double", "tests.uq_double.v1")] + [MessagePackObject(AllowPrivate = true)] + public sealed class UqDouble + { + [Key(0)] [CultIndex("level")] public double Level { get; set; } + } + + [CultDocument("tests.uq_tagged", "tests.uq_tagged.v1")] + [MessagePackObject(AllowPrivate = true)] + public sealed class UqTagged + { + [Key(0)] [CultRole("kind")] public string? Kind { get; set; } + } + } +} diff --git a/tests/GameCult.Caching.Tests/VariantTests.cs b/tests/GameCult.Caching.Tests/VariantTests.cs index 8f574cd1..17f9ba26 100644 --- a/tests/GameCult.Caching.Tests/VariantTests.cs +++ b/tests/GameCult.Caching.Tests/VariantTests.cs @@ -580,7 +580,7 @@ namespace GameCult.Caching.Tests { cache.Override(nameof(VariantGear.Name), "laser big") }))); - Assert.That(inherited.Message, Does.Contain(BigKey.Value).And.Contain(BaseKey.Value).And.Contain("code").And.Contain("l1")); + Assert.That(inherited.Message, Does.Contain(BigKey.Value).And.Contain(BaseKey.Value).And.Contain("code").And.Not.Contain("l1"), "the value is not echoed"); var explicitly = Refused(() => cache.Commit(batch => batch.UpsertVariant(BigKey, BaseKey, new[] { @@ -631,7 +631,7 @@ namespace GameCult.Caching.Tests SeedBig(cache); var refused = Refused(() => cache.Commit(batch => batch.Upsert(typeof(VariantGear), Other("laser big"), other))); - Assert.That(refused.Message, Does.Contain(other.Value).And.Contain(BigKey.Value).And.Contain("code").And.Contain("laser big")); + Assert.That(refused.Message, Does.Contain(other.Value).And.Contain(BigKey.Value).And.Contain("code").And.Not.Contain("laser big")); var single = Assert.ThrowsAsync( () => cache.UpsertAsync(Other("laser big"), new CultRecordHandle(other)))!; @@ -660,7 +660,7 @@ namespace GameCult.Caching.Tests cache.Commit(batch => batch.Upsert(typeof(VariantGear), Other("laser big"), other)); var refused = Refused(() => SeedBig(cache)); - Assert.That(refused.Message, Does.Contain(BigKey.Value).And.Contain(other.Value).And.Contain("code").And.Contain("laser big")); + Assert.That(refused.Message, Does.Contain(BigKey.Value).And.Contain(other.Value).And.Contain("code").And.Not.Contain("laser big")); Assert.That(cache.Get(BigKey), Is.Null, "nothing landed"); // The same batch, either staging order: the result is what is judged. @@ -685,8 +685,9 @@ namespace GameCult.Caching.Tests })); Assert.That(sameBatchVariantFirst.Message, Does.Contain(BigKey.Value).And.Contain("taker")); - // Plain-vs-plain sharing is not this rule's business. - Assert.DoesNotThrow(() => cache.Commit(batch => batch.Upsert(typeof(VariantGear), Other("l1"), new CultRecordKey("twin")))); + // Plain-vs-plain sharing is the same rule. + var plain = Refused(() => cache.Commit(batch => batch.Upsert(typeof(VariantGear), Other("l1"), new CultRecordKey("twin")))); + Assert.That(plain.Message, Does.Contain("twin").And.Contain(BaseKey.Value)); } using var reopened = Open(path); @@ -819,7 +820,7 @@ namespace GameCult.Caching.Tests using var a = Open(path); using (var b = Open(path)) b.Commit(batch => batch.Upsert(typeof(VariantGear), Other("zc"), z)); - a.Commit(batch => batch.Upsert(typeof(VariantGear), Other("yc"), new CultRecordKey("y"))); + a.Commit(batch => batch.Upsert(typeof(VariantOther), new VariantOther { Name = "yc" }, new CultRecordKey("y"))); var conflict = Assert.Throws(() => SeedBig(a))!; @@ -1100,22 +1101,8 @@ namespace GameCult.Caching.Tests Assert.That(reloaded.Get(BigKey)!.Tags, Is.EqualTo(new[] { "beam", "pierce" })); } - // ---- the held index is truthful: every holder of a value, so no rule judges a variant against a lossy view ---- - private static VariantGear Named(string name, string code) => new() { Name = name, Power = 1, Code = code }; - private static CultRecordKey[] SharedCodePair(CultCache cache) - { - var p1 = new CultRecordKey("p1"); - var p2 = new CultRecordKey("p2"); - cache.Commit(batch => - { - batch.Upsert(typeof(VariantGear), Named("p1", "x"), p1); - batch.Upsert(typeof(VariantGear), Named("p2", "x"), p2); - }); - return new[] { p1, p2 }; - } - private static void VariantOnCode(CultCache cache, string code) => cache.Commit(batch => batch.UpsertVariant(BigKey, BaseKey, new[] { @@ -1123,73 +1110,6 @@ namespace GameCult.Caching.Tests cache.Override(nameof(VariantGear.Code), code) })); - [Test] - public void RemovingTheHolderADuplicateValueFindsLeavesTheOtherFindable() - { - var path = PathOf("gear.cc"); - using (var cache = Open(path)) - { - var pair = SharedCodePair(cache); - Assert.That(cache.GetByIndex("code", "x")!.Name, Is.EqualTo("p1"), "the smallest key"); - - cache.Remove(pair[0]); - Assert.That(cache.GetByIndex("code", "x")!.Name, Is.EqualTo("p2"), "removing the holder finds the other one"); - - cache.Commit(batch => batch.Upsert(typeof(VariantGear), Named("p1", "x"), pair[0])); - Assert.That(cache.GetByIndex("code", "x")!.Name, Is.EqualTo("p1")); - cache.Remove(pair[1]); - Assert.That(cache.GetByIndex("code", "x")!.Name, Is.EqualTo("p1"), "removing the other changes nothing"); - cache.Remove(pair[0]); - Assert.That(cache.GetByIndex("code", "x"), Is.Null); - cache.BackingStores[0].PushAll(); - } - - using var reopened = Open(path); - Assert.That(reopened.GetByIndex("code", "x"), Is.Null); - } - - [Test] - public void AVariantOnAValueAPlainRecordStillHoldsAfterItsTwinWasRemovedIsRefusedAndTheStoreReopens() - { - var path = PathOf("gear.cc"); - using (var cache = Open(path)) - { - SeedBase(cache); - var pair = SharedCodePair(cache); - cache.Remove(pair[1]); - - var refused = Refused(() => VariantOnCode(cache, "x")); - Assert.That(refused.Message, Does.Contain(BigKey.Value).And.Contain("p1").And.Contain("code").And.Contain("'x'")); - Assert.That(cache.Get(BigKey), Is.Null, "nothing landed"); - cache.BackingStores[0].PushAll(); - } - - using var reopened = Open(path); - Assert.That(reopened.Get(BigKey), Is.Null); - Assert.That(reopened.GetByIndex("code", "x")!.Name, Is.EqualTo("p1")); - } - - [Test] - public void AVariantOnAValueAPlainRecordStillHoldsAfterItsTwinChangedItsValueIsRefusedAndTheStoreReopens() - { - var path = PathOf("gear.cc"); - using (var cache = Open(path)) - { - SeedBase(cache); - var pair = SharedCodePair(cache); - cache.Commit(batch => batch.Upsert(typeof(VariantGear), Named("p2", "y"), pair[1])); - - var refused = Refused(() => VariantOnCode(cache, "x")); - Assert.That(refused.Message, Does.Contain(BigKey.Value).And.Contain("p1").And.Contain("'x'")); - Assert.DoesNotThrow(() => VariantOnCode(cache, "z"), "an unshared value is fine"); - cache.BackingStores[0].PushAll(); - } - - using var reopened = Open(path); - Assert.That(reopened.GetByIndex("code", "z")!.Name, Is.EqualTo("laser big")); - Assert.That(reopened.GetByIndex("code", "x")!.Name, Is.EqualTo("p1")); - } - // ---- a merge onto the file is judged on the set the file will hold ---- [Test] @@ -1218,7 +1138,7 @@ namespace GameCult.Caching.Tests var conflict = Assert.Throws(Write)!; Assert.That(conflict.ChangedKeys, Does.Contain(BigKey.Value)); var refused = Refused(Write); - Assert.That(refused.Message, Does.Contain(other.Value).And.Contain(BigKey.Value).And.Contain("code").And.Contain("shared"), "the retry is judged by Resolve"); + Assert.That(refused.Message, Does.Contain(other.Value).And.Contain(BigKey.Value).And.Contain("code").And.Not.Contain("shared"), "the retry is judged by Resolve"); Assert.That(b.Get(other), Is.Null, "nothing landed in memory"); } @@ -1408,24 +1328,24 @@ namespace GameCult.Caching.Tests { var path = PathOf("gear.cc"); var other = new CultRecordKey("other"); - var narrow = CultDocumentRegistry.ForTypes(new[] { typeof(VariantGear) }); + var narrow = CultDocumentRegistry.ForTypes(new[] { typeof(VariantOther) }); using (var seed = Open(path)) SeedBase(seed); using (var a = Open(path)) using (var b = Open(path, registry: narrow)) { - a.Commit(batch => batch.Upsert(typeof(VariantOther), new VariantOther { Name = "foreign" }, new CultRecordKey("foreign"))); + a.Commit(batch => batch.Upsert(typeof(VariantGear), Named("foreign", "foreign-code"), new CultRecordKey("foreign"))); Assert.That(b.Commit(batch => { batch.Expect(other, null); - batch.Upsert(typeof(VariantGear), Named("other", "fine"), other); + batch.Upsert(typeof(VariantOther), new VariantOther { Name = "fine" }, other); }), Is.True); } using var reopened = Open(path); - Assert.That(reopened.Get(other), Is.Not.Null); - Assert.That(reopened.Get(new CultRecordKey("foreign")), Is.Not.Null, "the other writer's record survives the merge"); + Assert.That(reopened.Get(other), Is.Not.Null); + Assert.That(reopened.Get(new CultRecordKey("foreign")), Is.Not.Null, "the other writer's record survives the merge"); } [Test] @@ -1448,28 +1368,6 @@ namespace GameCult.Caching.Tests } } - [Test] - public void PlainDuplicatesFindTheSmallestKeyLiveAfterAFlushAndAfterAReopen() - { - var path = PathOf("gear.cc"); - using (var cache = Open(path)) - { - cache.Commit(batch => batch.Upsert(typeof(VariantGear), Named("nb", "x"), new CultRecordKey("b"))); - cache.Commit(batch => batch.Upsert(typeof(VariantGear), Named("na", "x"), new CultRecordKey("a"))); - Assert.That(cache.GetByIndex("code", "x")!.Name, Is.EqualTo("na"), "live"); - cache.BackingStores[0].PushAll(); - Assert.That(cache.GetByIndex("code", "x")!.Name, Is.EqualTo("na"), "after a flush"); - } - - using (var reopened = Open(path)) - { - Assert.That(reopened.GetByIndex("code", "x")!.Name, Is.EqualTo("na"), "after a reopen"); - reopened.Remove(new CultRecordKey("a")); - reopened.Commit(batch => batch.Upsert(typeof(VariantGear), Named("na", "x"), new CultRecordKey("a"))); - Assert.That(reopened.GetByIndex("code", "x")!.Name, Is.EqualTo("na"), "rewriting the smaller key does not change the choice"); - } - } - [Test] public void ADroppedSlotIsReportedOncePerRecord() { diff --git a/tests/GameCult.Networking.Tests/CultNetReplicaCommitTests.cs b/tests/GameCult.Networking.Tests/CultNetReplicaCommitTests.cs new file mode 100644 index 00000000..0e2f2031 --- /dev/null +++ b/tests/GameCult.Networking.Tests/CultNetReplicaCommitTests.cs @@ -0,0 +1,621 @@ +#nullable enable +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using GameCult.Caching; +using R3; +using MessagePack; +using NUnit.Framework; + +namespace GameCult.Networking.Tests +{ + // variants:ruling:replica-commit-grouped-changes. The authoritative cache judges a commit on its result; a replica applying the + // changes one record at a time would judge each intermediate state, where a swap of two [CultIndex] values holds two holders. + // The subscription server tags the changes of one cache commit, the client applies them as one commit, and a replica that + // refuses a commit says so. No transport: the server and client are joined in memory, and every message crosses MessagePack. + public sealed class CultNetReplicaCommitTests + { + [CultDocument("tests.replica_commit_item", "tests.replica_commit_item.v1")] + [MessagePackObject(AllowPrivate = true)] + public sealed class ReplicaItem + { + [Key(0)] + [CultIndex("sku")] + public string Sku = string.Empty; + + [Key(1)] + public string Note = string.Empty; + } + + // The change message as a peer from before commit identity knew it. + [MessagePackObject] + public sealed class LegacyChange + { + [Key("schemaVersion")] public string SchemaVersion { get; set; } = string.Empty; + [Key("messageId")] public string MessageId { get; set; } = string.Empty; + [Key("subscriptionId")] public string SubscriptionId { get; set; } = string.Empty; + [Key("changeKind")] public string ChangeKind { get; set; } = string.Empty; + [Key("document")] public CultNetRawDocumentRecord? Document { get; set; } + [Key("recordKey")] public string? RecordKey { get; set; } + [Key("schemaId")] public string? SchemaId { get; set; } + } + + private sealed class LoopbackServer : ICultNetSchemaServer + { + private readonly Dictionary _handlers = new(); + public void OnCultNet(Func callback) where T : ICultNetSchemaMessage => _handlers[typeof(T)] = callback; + public void RemoveCultNetMessageListener(Delegate callback) where T : ICultNetSchemaMessage => _handlers.Remove(typeof(T)); + public Task Dispatch(T message, ICultNetSchemaServerPeer peer) where T : ICultNetSchemaMessage => + ((Func)_handlers[typeof(T)])(message, peer); + } + + private sealed class Client : ICultNetSchemaClient + { + private readonly LoopbackServer _server; + private readonly Dictionary> _handlers = new(); + private readonly Peer _peer; + public Client(LoopbackServer server, Peer peer) { _server = server; _peer = peer; peer.Client = this; } + public bool Connected => true; + public void Connect(string host, int port) { } + public void SendCultNet(T message) where T : ICultNetSchemaMessage + { + if (message is CultNetDatabaseSubscribeMessage subscribe) + _server.Dispatch(subscribe, _peer).GetAwaiter().GetResult(); + else if (message is CultNetDatabaseUnsubscribeMessage unsubscribe) + _server.Dispatch(unsubscribe, _peer).GetAwaiter().GetResult(); + } + public void OnCultNet(Action callback) where T : ICultNetSchemaMessage + { + if (!_handlers.TryGetValue(typeof(T), out var list)) _handlers[typeof(T)] = list = new List(); + list.Add(callback); + } + public void Emit(T message) where T : ICultNetSchemaMessage + { + if (_handlers.TryGetValue(typeof(T), out var list)) + foreach (var handler in list.ToArray()) ((Action)handler)(message); + } + public void Dispose() { } + } + + private sealed class Peer : ICultNetSchemaServerPeer + { + public Client? Client; + public ConcurrentQueue Sent { get; } = new(); + public void SendCultNet(T message) where T : ICultNetSchemaMessage + { + // The wire: serialized, then read back as the receiver would. + var bytes = CultNetSchemaMessageSerialization.Serialize(message); + Sent.Enqueue(bytes); + Client!.Emit(CultNetSchemaMessageSerialization.Deserialize(bytes)); + } + public CultNetDatabaseChangeRawMessage[] Changes => Sent + .Select(bytes => CultNetSchemaMessageSerialization.Deserialize(bytes)) + .OfType().ToArray(); + } + + private sealed class Replication : IDisposable + { + public CultCache Source { get; } = new(); + public CultCache Replica { get; } = new(); + public Peer Peer { get; } = new(); + public CultNetDatabaseSubscriptionClient Client { get; } + public List Faults { get; } = new(); + public CultNetDatabase Database { get; } + private readonly CultNetDatabaseSubscriptionServer _server; + + public Replication(Action? seedSource = null, Action? seedReplica = null) + { + seedSource?.Invoke(Source); + seedReplica?.Invoke(Replica); + Database = new CultNetDatabase(Source); + var wire = new LoopbackServer(); + _server = new CultNetDatabaseSubscriptionServer(wire, Database); + var documents = new CultNetDocumentRegistry(Replica.Registry) + .Register(CultNetDocumentBinding.ForDocument(Replica.Registry)); + Client = new CultNetDatabaseSubscriptionClient(new Client(wire, Peer), Replica, documents); + Client.Faulted += Faults.Add; + Client.SubscribeAsync("items", schemaIds: new[] { Source.Registry.GetRequired().SchemaId }) + .GetAwaiter().GetResult(); + } + + public ReplicaItem? ReplicaItem(string key) => Replica.Get(new CultRecordKey(key)); + + public void Dispose() + { + Client.Dispose(); + _server.Dispose(); + Database.Dispose(); + } + } + + private static ReplicaItem Item(string sku) => new() { Sku = sku }; + + private static void Stage(CultCache cache, params (string Key, string Sku)[] items) => + Assert.That(cache.Commit(batch => + { + foreach (var (key, sku) in items) + batch.Upsert(typeof(ReplicaItem), Item(sku), new CultRecordKey(key)); + }), Is.True); + + private static async Task Until(Func condition, int seconds = 5) + { + var deadline = DateTime.UtcNow.AddSeconds(seconds); + while (!condition()) + { + if (DateTime.UtcNow > deadline) Assert.Fail("The replica did not reach the expected state."); + await Task.Delay(5); + } + } + + [Test] + public async Task ASwapOfTwoIndexValuesReachesTheReplicaIntact() + { + using var run = new Replication(source => Stage(source, ("a", "one"), ("b", "two"))); + Assert.That(run.ReplicaItem("a")!.Sku, Is.EqualTo("one")); + + Stage(run.Source, ("a", "two"), ("b", "one")); + await Until(() => run.ReplicaItem("a")!.Sku == "two"); + + Assert.That(run.ReplicaItem("b")!.Sku, Is.EqualTo("one")); + Assert.That(run.Client.Fault, Is.Null); + Assert.That(run.Faults, Is.Empty); + Assert.That(run.Replica.GetByIndex("sku", "two"), Is.SameAs(run.ReplicaItem("a"))); + var swap = run.Peer.Changes; + Assert.That(swap.Select(change => change.CommitId).Distinct().Single(), Is.Not.Null, "the commit is named on the wire"); + Assert.That(swap.Select(change => change.CommitCount), Is.All.EqualTo(2)); + } + + [Test] + public async Task AValueMovedToANewRecordWhileItsOldHolderIsRemovedReachesTheReplica() + { + using var run = new Replication(source => Stage(source, ("a", "one"))); + + Assert.That(run.Source.Commit(batch => + { + batch.Remove(new CultRecordKey("a")); + batch.Upsert(typeof(ReplicaItem), Item("one"), new CultRecordKey("c")); + }), Is.True); + await Until(() => run.ReplicaItem("c") != null); + + Assert.That(run.ReplicaItem("a"), Is.Null); + Assert.That(run.Client.Fault, Is.Null); + Assert.That(run.Replica.GetByIndex("sku", "one"), Is.SameAs(run.ReplicaItem("c"))); + } + + [Test] + public async Task ACommitOfOneChangeCarriesNoCommitFieldsAndIsApplied() + { + using var run = new Replication(); + + Stage(run.Source, ("a", "one")); + await Until(() => run.ReplicaItem("a") != null); + + var change = run.Peer.Changes.Single(); + Assert.That(change.CommitId, Is.Null); + Assert.That(change.CommitCount, Is.Null); + } + + [Test] + public async Task AReplicaThatRefusesACommitFaultsLoudlyAndAppliesNothingMore() + { + // The replica holds a record the source does not, under the value the source now gives to another record: a conflict + // the replica cannot resolve, not an intermediate state. + using var run = new Replication(seedReplica: replica => Stage(replica, ("local", "taken"))); + + Stage(run.Source, ("a", "taken"), ("b", "free")); + await Until(() => run.Client.Fault != null); + + Assert.That(run.Faults, Has.Count.EqualTo(1)); + Assert.That(run.Faults[0], Is.SameAs(run.Client.Fault)); + Assert.That(run.Faults[0].Message, Does.Contain("sku")); + Assert.That(run.ReplicaItem("a"), Is.Null, "the refused commit is not partly applied"); + Assert.That(run.ReplicaItem("b"), Is.Null, "the refused commit is not partly applied"); + + Stage(run.Source, ("c", "later")); + await Task.Delay(100); + Assert.That(run.ReplicaItem("c"), Is.Null, "a replica known to differ takes nothing on top"); + Assert.That(run.Faults, Has.Count.EqualTo(1)); + } + + [Test] + public void AnOlderPeerReadsTheChangeAndIgnoresTheCommitFields() + { + var tagged = new CultNetDatabaseChangeRawMessage + { + MessageId = "m", + SubscriptionId = "items", + ChangeKind = "removed", + RecordKey = "a", + SchemaId = "tests.replica_commit_item.v1", + CommitId = "7", + CommitCount = 2 + }; + + var legacy = MessagePackSerializer.Deserialize( + CultNetSchemaMessageSerialization.Serialize(tagged), CultNetSchemaMessageSerialization.Options); + + Assert.That(legacy.ChangeKind, Is.EqualTo("removed")); + Assert.That(legacy.RecordKey, Is.EqualTo("a")); + Assert.That(legacy.SubscriptionId, Is.EqualTo("items")); + } + + [Test] + public async Task ASwapFromAServerThatDoesNotTagCommitsStillFaultsInsteadOfDiverging() + { + // An older server sends the two halves of a swap untagged: each is judged alone and refused. The refusal is surfaced. + using var run = new Replication(source => Stage(source, ("a", "one"), ("b", "two"))); + var raw = (string key, string sku) => run.Database.Documents.ToRawRecord( + run.Source.Registry.GetRequired(), new CultRecordKey(key), Item(sku), "2026-01-01T00:00:00Z"); + + run.Peer.Client!.Emit(new CultNetDatabaseChangeRawMessage + { + MessageId = "x", SubscriptionId = "items", ChangeKind = "updated", Document = raw("a", "two") + }); + await Until(() => run.Client.Fault != null); + + Assert.That(run.ReplicaItem("a")!.Sku, Is.EqualTo("one")); + } + + [Test] + public async Task ARotationOfThreeValuesReachesTheReplicaIntact() + { + using var run = new Replication(source => Stage(source, ("a", "one"), ("b", "two"), ("c", "three"))); + + Stage(run.Source, ("a", "two"), ("b", "three"), ("c", "one")); + await Until(() => run.ReplicaItem("a")!.Sku == "two"); + + Assert.That(new[] { "a", "b", "c" }.Select(key => run.ReplicaItem(key)!.Sku), Is.EqualTo(new[] { "two", "three", "one" })); + Assert.That(run.Client.Fault, Is.Null); + Assert.That(run.Peer.Changes.Select(change => change.CommitCount), Is.All.EqualTo(3)); + } + + // The database server is the other producer of change messages; it names a commit the same way. + [Test] + public void TheDatabaseServerTagsTheChangesOfOneCommitAndNotALoneChange() + { + var cache = new CultCache(); + Stage(cache, ("a", "one"), ("b", "two")); + using var database = new CultNetDatabase(cache); + using var server = new Server(cache, ServerSecurityOptions.Development()); + using var databaseServer = new CultNetDatabaseServer(server, database); + var commits = new List(); + using var watch = database.WatchCommits().Subscribe(commits.Add); + var selection = new CultNetSelection + { + Schemas = new[] { cache.Registry.GetRequired().SchemaId }, + Projection = CultNetSelectionProjections.Document + }; + IReadOnlyList Deliver(CultNetDatabaseCommit commit) => + (IReadOnlyList)typeof(CultNetDatabaseServer) + .GetMethod("DeliverCommit", System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.NonPublic)! + .Invoke(databaseServer, new object[] { commit, "sub", selection, new HashSet(), (Action)(_ => { }) })!; + + Stage(cache, ("a", "two"), ("b", "one")); + Stage(cache, ("c", "three")); + + var swap = Deliver(commits[0]); + Assert.That(swap, Has.Count.EqualTo(2)); + Assert.That(swap.Select(change => change.CommitId).Distinct().Single(), Is.EqualTo(commits[0].Id)); + Assert.That(swap.Select(change => change.CommitCount), Is.All.EqualTo(2)); + var lone = Deliver(commits[1]).Single(); + Assert.That(lone.CommitId, Is.Null); + Assert.That(lone.CommitCount, Is.Null); + Assert.That(commits[0].Id, Is.Not.EqualTo(commits[1].Id)); + } + + [Test] + public void ASnapshotThatSwapsIndexValuesAgainstWhatTheReplicaHoldsIsApplied() + { + using var run = new Replication( + source => Stage(source, ("a", "two"), ("b", "one")), + replica => Stage(replica, ("a", "one"), ("b", "two"))); + + Assert.That(run.ReplicaItem("a")!.Sku, Is.EqualTo("two")); + Assert.That(run.ReplicaItem("b")!.Sku, Is.EqualTo("one")); + } + + // Hand-made wire messages: what a peer sends that the real server would not. + private static CultNetDatabaseChangeRawMessage Put(Replication run, string subscription, string key, string sku, string? commit, int count, string? id = null) => + new() + { + MessageId = id ?? Guid.NewGuid().ToString("N"), + SubscriptionId = subscription, + ChangeKind = "updated", + Document = run.Database.Documents.ToRawRecord( + run.Source.Registry.GetRequired(), new CultRecordKey(key), Item(sku), "2026-01-01T00:00:00Z"), + CommitId = commit, + CommitCount = commit == null ? null : count + }; + + private static CultNetDatabaseChangeRawMessage Removal(string subscription, string key, string commit, int count) => + new() + { + MessageId = Guid.NewGuid().ToString("N"), + SubscriptionId = subscription, + ChangeKind = "removed", + RecordKey = key, + SchemaId = "tests.replica_commit_item.v1", + CommitId = commit, + CommitCount = count + }; + + [Test] + public async Task TwoSubscriptionsMayUseOneCommitIdWithoutJoiningTheirCommits() + { + using var run = new Replication(source => Stage(source, ("a", "one"), ("b", "two"))); + run.Client.SubscribeAsync("other", recordKeys: new[] { "nothing-matches" }).GetAwaiter().GetResult(); + var emit = run.Peer.Client!; + + emit.Emit(Put(run, "items", "a", "two", "9", 2)); + emit.Emit(Put(run, "other", "x", "ex", "9", 2)); + await Task.Delay(100); + Assert.That(run.ReplicaItem("a")!.Sku, Is.EqualTo("one"), "one change of each of two commits is not a commit"); + Assert.That(run.ReplicaItem("x"), Is.Null); + + emit.Emit(Put(run, "items", "b", "one", "9", 2)); + await Until(() => run.ReplicaItem("a")!.Sku == "two"); + Assert.That(run.ReplicaItem("b")!.Sku, Is.EqualTo("one")); + Assert.That(run.ReplicaItem("x"), Is.Null); + Assert.That(run.Client.Fault, Is.Null); + } + + [Test] + public async Task ARemovalAfterTheFirstChangeOfACommitIsApplied() + { + using var run = new Replication(source => Stage(source, ("a", "one"))); + var emit = run.Peer.Client!; + + emit.Emit(Put(run, "items", "c", "one", "5", 2)); + emit.Emit(Removal("items", "a", "5", 2)); + await Until(() => run.ReplicaItem("c") != null); + + Assert.That(run.ReplicaItem("a"), Is.Null); + Assert.That(run.Client.Fault, Is.Null); + } + + [Test] + public void ASnapshotOfManyDocumentsAppliesEveryOne() + { + using var run = new Replication(source => Stage(source, ("a", "one"), ("b", "two"), ("c", "three"), ("d", "four"))); + + Assert.That(new[] { "a", "b", "c", "d" }.Select(key => run.ReplicaItem(key)?.Sku), Is.EqualTo(new[] { "one", "two", "three", "four" })); + } + + [Test] + public async Task ManyCommitsInARowAreNotMistakenForAnUnboundedBuffer() + { + using var run = new Replication(source => Stage(source, ("a", "s0"), ("b", "s1"))); + + for (var round = 1; round <= 80; round++) + { + var a = "s" + (round % 2); + var b = "s" + ((round + 1) % 2); + Stage(run.Source, ("a", a), ("b", b)); + } + + Stage(run.Source, ("a", "last-a"), ("b", "last-b")); + await Until(() => run.Client.Fault != null || run.ReplicaItem("a")!.Sku == "last-a"); + Assert.That(run.Client.Fault, Is.Null, "completed commits leave the buffer"); + Assert.That(run.ReplicaItem("b")!.Sku, Is.EqualTo("last-b")); + } + + [Test] + public async Task AMessageDeliveredTwiceInOneCommitFaults() + { + using var run = new Replication(); + + run.Peer.Client!.Emit(Put(run, "items", "a", "one", "3", 3, id: "same")); + run.Peer.Client!.Emit(Put(run, "items", "b", "two", "3", 3, id: "same")); + await Until(() => run.Client.Fault != null); + + Assert.That(run.Faults, Has.Count.EqualTo(1)); + Assert.That(run.ReplicaItem("a"), Is.Null); + } + + [Test] + public async Task ACommitThatChangesItsAnnouncedSizeFaults() + { + using var run = new Replication(); + + run.Peer.Client!.Emit(Put(run, "items", "a", "one", "3", 2)); + run.Peer.Client!.Emit(Put(run, "items", "b", "two", "3", 3)); + await Until(() => run.Client.Fault != null); + + Assert.That(run.ReplicaItem("a"), Is.Null); + } + + [Test] + public async Task ACommitAnnouncingMoreChangesThanAReplicaHoldsFaults() + { + using var run = new Replication(); + + run.Peer.Client!.Emit(Put(run, "items", "a", "one", "3", 1_000_001)); + await Until(() => run.Client.Fault != null); + } + + [Test] + public async Task MoreIncompleteCommitsThanTheBoundFault() + { + using var run = new Replication(); + + for (var commit = 0; commit < 64; commit++) + run.Peer.Client!.Emit(Put(run, "items", "k" + commit, "v" + commit, "c" + commit, 2)); + await Task.Delay(200); + Assert.That(run.Client.Fault, Is.Null, "64 incomplete commits are within the bound"); + Assert.That(Pending(run.Client), Is.EqualTo(64)); + + run.Peer.Client!.Emit(Put(run, "items", "k64", "v64", "c64", 2)); + await Until(() => run.Client.Fault != null); + Assert.That(Pending(run.Client), Is.Zero, "a fault drops what was waiting"); + } + + [Test] + public async Task UnsubscribingDropsTheCommitsStillWaiting() + { + using var run = new Replication(); + var emit = run.Peer.Client!; + + emit.Emit(Put(run, "items", "a", "one", "4", 2)); + await Task.Delay(50); + run.Client.Unsubscribe("items"); + emit.Emit(Put(run, "items", "b", "two", "4", 2)); + await Task.Delay(100); + + Assert.That(run.ReplicaItem("a"), Is.Null); + Assert.That(run.ReplicaItem("b"), Is.Null, "the second half of a dropped commit is not a commit"); + } + + [Test] + public async Task AFaultedClientRefusesANewSubscription() + { + using var run = new Replication(seedReplica: replica => Stage(replica, ("local", "taken"))); + Stage(run.Source, ("a", "taken")); + await Until(() => run.Client.Fault != null); + + var refusal = Assert.Throws(() => run.Client.SubscribeAsync("again")); + Assert.That(refusal!.InnerException, Is.SameAs(run.Client.Fault)); + } + + private static int Pending(CultNetDatabaseSubscriptionClient client) + { + var commits = (System.Collections.IDictionary)typeof(CultNetDatabaseSubscriptionClient) + .GetField("_commits", System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.NonPublic)!.GetValue(client)!; + lock (commits) + return commits.Count; + } + + [Test] + public async Task TheIncompleteCommitsMayAnnounceOneMillionChangesBetweenThem() + { + using var run = new Replication(); + + run.Peer.Client!.Emit(Put(run, "items", "a", "one", "1", 500_000)); + run.Peer.Client!.Emit(Put(run, "items", "b", "two", "2", 500_000)); + await Task.Delay(200); + Assert.That(run.Client.Fault, Is.Null, "500k and 500k fit"); + + run.Peer.Client!.Emit(Put(run, "items", "c", "three", "3", 2)); + await Until(() => run.Client.Fault != null); + } + + [Test] + public async Task ACommitOfOneMillionChangesIsHeldAndOneMillionAndOneIsNot() + { + using var held = new Replication(); + held.Peer.Client!.Emit(Put(held, "items", "a", "one", "1", 1_000_000)); + await Task.Delay(200); + Assert.That(held.Client.Fault, Is.Null); + Assert.That(Pending(held.Client), Is.EqualTo(1)); + + using var over = new Replication(); + over.Peer.Client!.Emit(Put(over, "items", "a", "one", "1", 1_000_001)); + await Until(() => over.Client.Fault != null); + } + + [Test] + public async Task ACommitThatShrinksItsAnnouncedSizeFaults() + { + using var run = new Replication(); + + run.Peer.Client!.Emit(Put(run, "items", "a", "one", "3", 3)); + run.Peer.Client!.Emit(Put(run, "items", "b", "two", "3", 2)); + await Until(() => run.Client.Fault != null); + } + + [Test] + public async Task UnsubscribingOneSubscriptionKeepsTheCommitsOfAnother() + { + using var run = new Replication(); + run.Client.SubscribeAsync("items2", recordKeys: new[] { "nothing-matches" }).GetAwaiter().GetResult(); + var emit = run.Peer.Client!; + + emit.Emit(Put(run, "items", "a", "one", "4", 2)); + emit.Emit(Put(run, "items2", "b", "two", "4", 2)); + await Task.Delay(100); + Assert.That(Pending(run.Client), Is.EqualTo(2)); + + run.Client.Unsubscribe("items"); + Assert.That(Pending(run.Client), Is.EqualTo(1), "only the unsubscribed subscription's commits leave"); + emit.Emit(Put(run, "items2", "c", "three", "4", 2)); + await Until(() => run.ReplicaItem("b") != null); + Assert.That(run.ReplicaItem("c"), Is.Not.Null); + Assert.That(run.ReplicaItem("a"), Is.Null); + } + + [Test] + public async Task AStragglerAfterUnsubscribeOrFromAnUnknownSubscriptionIsNotBuffered() + { + using var run = new Replication(); + run.Client.Unsubscribe("items"); + + run.Peer.Client!.Emit(Put(run, "items", "a", "one", "4", 2)); + run.Peer.Client!.Emit(Put(run, "never-subscribed", "b", "two", "4", 2)); + await Task.Delay(150); + + Assert.That(Pending(run.Client), Is.Zero); + Assert.That(run.Client.Fault, Is.Null); + } + + [Test] + public void DisposingTheClientDropsTheCommitsStillWaiting() + { + var run = new Replication(); + run.Peer.Client!.Emit(Put(run, "items", "a", "one", "4", 2)); + SpinWait.SpinUntil(() => Pending(run.Client) == 1, 2000); + Assert.That(Pending(run.Client), Is.EqualTo(1)); + + run.Dispose(); + + Assert.That(Pending(run.Client), Is.Zero); + } + + [Test] + public async Task AFaultedClientBuffersNothingMore() + { + using var run = new Replication(); + run.Peer.Client!.Emit(Put(run, "items", "a", "one", "3", 3, id: "same")); + run.Peer.Client!.Emit(Put(run, "items", "b", "two", "3", 3, id: "same")); + await Until(() => run.Client.Fault != null); + + run.Peer.Client!.Emit(Put(run, "items", "c", "three", "8", 2)); + run.Peer.Client!.Emit(Put(run, "items", "d", "four", "9", 2)); + await Task.Delay(150); + + Assert.That(Pending(run.Client), Is.Zero); + Assert.That(run.Faults, Has.Count.EqualTo(1), "a fault is reported once"); + } + + [Test] + public void AFaultIsReportedOnceHoweverOftenItIsRaised() + { + using var run = new Replication(); + var raise = typeof(CultNetDatabaseSubscriptionClient) + .GetMethod("Break", System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.NonPublic)!; + + raise.Invoke(run.Client, new object[] { new InvalidOperationException("first") }); + raise.Invoke(run.Client, new object[] { new InvalidOperationException("second") }); + + Assert.That(run.Faults, Has.Count.EqualTo(1)); + Assert.That(run.Client.Fault!.Message, Is.EqualTo("first")); + } + + [TestCase(5000)] + [TestCase(100_000)] + public async Task ACommitOfManyChangesThroughTheRealSubscriptionServerAppliesIntact(int size) + { + using var run = new Replication(); + var keys = Enumerable.Range(0, size).Select(i => ("k" + i, "s" + i)).ToArray(); + + Stage(run.Source, keys); + await Until(() => run.ReplicaItem("k" + (size - 1)) != null || run.Client.Fault != null, seconds: 180); + + Assert.That(run.Client.Fault, Is.Null); + Assert.That(run.Replica.GetAll().Count(), Is.EqualTo(size)); + Assert.That(run.Peer.Changes.Select(change => change.CommitCount).Distinct().Single(), Is.EqualTo(size)); + Assert.That(run.ReplicaItem("k" + (size / 2))!.Sku, Is.EqualTo("s" + (size / 2))); + } + } +} diff --git a/tests/GameCult.Networking.Tests/CultNetSelectionEvaluatorTests.cs b/tests/GameCult.Networking.Tests/CultNetSelectionEvaluatorTests.cs index b2dfcf3c..b700ec19 100644 --- a/tests/GameCult.Networking.Tests/CultNetSelectionEvaluatorTests.cs +++ b/tests/GameCult.Networking.Tests/CultNetSelectionEvaluatorTests.cs @@ -486,6 +486,39 @@ namespace GameCult.Networking.Tests Assert.That(evaluation.Rows.Select(r => r.Key.Value), Is.EqualTo(new[] { "a" })); } + // [CultIndex] means unique and is also a selection address: a unique member filters by equality and by number like a role. + [Test] + public void AUniqueIndexMemberIsAlsoSelectable() + { + var registry = CultDocumentRegistry.ForTypes(new[] { typeof(SelUnique) }); + var rows = new[] + { + Row(registry, new SelUnique { Sku = "s1", Rank = 1 }, "a", 1), + Row(registry, new SelUnique { Sku = "s2", Rank = 7 }, "b", 2) + }; + + var byValue = CultNetSelectionEvaluator.Select(registry, rows, + new CultNetSelection { Fields = new[] { new CultNetFieldPredicate { Index = "sku", Op = "any_of", Values = new[] { "s2" } } } }, asOf: 2); + var byNumber = CultNetSelectionEvaluator.Select(registry, rows, + new CultNetSelection { Fields = new[] { new CultNetFieldPredicate { Index = "rank", Op = "gt", Number = "3" } } }, asOf: 2); + + Assert.That(byValue.Rows.Select(r => r.Key.Value), Is.EqualTo(new[] { "b" })); + Assert.That(byNumber.Rows.Select(r => r.Key.Value), Is.EqualTo(new[] { "b" })); + } + + [CultDocument("cultnet.selection-tests.unique", "cultnet.selection-tests.unique.v1")] + [MessagePackObject(AllowPrivate = true)] + public sealed class SelUnique + { + [Key(0)] + [CultIndex("sku")] + public string Sku = string.Empty; + + [Key(1)] + [CultIndex("rank")] + public int Rank; + } + public abstract class SelFixtureMiddle { [Key(0)] @@ -493,11 +526,11 @@ namespace GameCult.Networking.Tests public string Name = string.Empty; [Key(1)] - [CultIndex("mass")] + [CultRole("mass")] public float Mass; [Key(2)] - [CultIndex("kind")] + [CultRole("kind")] public string Kind = string.Empty; } diff --git a/tests/GameCult.Networking.Tests/CultNetSelectionFixBatchTests.cs b/tests/GameCult.Networking.Tests/CultNetSelectionFixBatchTests.cs index 9ecc4556..030ae100 100644 --- a/tests/GameCult.Networking.Tests/CultNetSelectionFixBatchTests.cs +++ b/tests/GameCult.Networking.Tests/CultNetSelectionFixBatchTests.cs @@ -581,7 +581,7 @@ namespace GameCult.Networking.Tests return instance; } - // A plain data member "Owner" with no [CultIndex] (its role falls back to its bare name) + // A plain data member "Owner" with no [CultRole] (its role falls back to its bare name) // and a reference member whose alias is explicitly "Owner" - the exact shape Soul probed. var ownerMember = MakeMember(typeof(GuardProbeShape).GetField(nameof(GuardProbeShape.Owner))!, typeof(string), 0, indexAlias: null, isReference: false, targetType: null); diff --git a/tests/GameCult.Networking.Tests/CultNetSelectionPageTests.cs b/tests/GameCult.Networking.Tests/CultNetSelectionPageTests.cs index 0f1f20a9..845f9468 100644 --- a/tests/GameCult.Networking.Tests/CultNetSelectionPageTests.cs +++ b/tests/GameCult.Networking.Tests/CultNetSelectionPageTests.cs @@ -130,7 +130,7 @@ namespace GameCult.Networking.Tests public string Name = string.Empty; [Key(1)] - [CultIndex("kind")] + [CultRole("kind")] public string Kind = string.Empty; } diff --git a/tests/GameCult.Networking.Tests/CultNetSelectionSurvivorTests.cs b/tests/GameCult.Networking.Tests/CultNetSelectionSurvivorTests.cs index a03b62d2..e625c134 100644 --- a/tests/GameCult.Networking.Tests/CultNetSelectionSurvivorTests.cs +++ b/tests/GameCult.Networking.Tests/CultNetSelectionSurvivorTests.cs @@ -62,7 +62,7 @@ namespace GameCult.Networking.Tests public sealed class SurvivorNumericLeaf { [Key(0)] [CultName] public string Name = string.Empty; - [Key(1)] [CultIndex("level")] public int Level; + [Key(1)] [CultRole("level")] public int Level; } [CultDocument("cultnet.selection-tests.survivor_string_leaf", "cultnet.selection-tests.survivor_string_leaf.v1")] @@ -70,7 +70,7 @@ namespace GameCult.Networking.Tests public sealed class SurvivorStringLeaf { [Key(0)] [CultName] public string Name = string.Empty; - [Key(1)] [CultIndex("level")] public string Level = string.Empty; + [Key(1)] [CultRole("level")] public string Level = string.Empty; } // --------------------------------------------------------------------------------------------- @@ -173,7 +173,7 @@ namespace GameCult.Networking.Tests // The declared index alias ("linkRole") differs from the member name ("Link") - the exact // shape section 2's role-naming rule exists for, and no prior fixture had one. [Key(1)] - [CultIndex("linkRole")] + [CultRole("linkRole")] [CultReference(typeof(SurvivorAliasTarget))] public CultRecordRef Link; } @@ -408,7 +408,7 @@ namespace GameCult.Networking.Tests [Key(0)] [CultName] public string Name = string.Empty; [Key(1)] - [CultIndex("hopRole")] + [CultRole("hopRole")] [CultReference(typeof(SurvivorAliasTarget))] public CultRecordRef Link; } @@ -431,12 +431,12 @@ namespace GameCult.Networking.Tests [Key(0)] [CultName] public string Name = string.Empty; [Key(1)] - [CultIndex("zLastRole")] + [CultRole("zLastRole")] [CultReference(typeof(SurvivorAliasTarget))] public CultRecordRef RefZ; [Key(2)] - [CultIndex("aFirstRole")] + [CultRole("aFirstRole")] [CultReference(typeof(SurvivorAliasTarget))] public CultRecordRef RefA; } diff --git a/tests/GameCult.Networking.Tests/SelectionParityVectorTests.cs b/tests/GameCult.Networking.Tests/SelectionParityVectorTests.cs index e9a070fc..3a076d16 100644 --- a/tests/GameCult.Networking.Tests/SelectionParityVectorTests.cs +++ b/tests/GameCult.Networking.Tests/SelectionParityVectorTests.cs @@ -586,11 +586,11 @@ namespace GameCult.Networking.Tests public string Name = string.Empty; [Key(1)] - [CultIndex("kind")] + [CultRole("kind")] public string Kind = string.Empty; [Key(2)] - [CultIndex("mass")] + [CultRole("mass")] public float Mass; } @@ -631,7 +631,7 @@ namespace GameCult.Networking.Tests public string Name = string.Empty; [Key(1)] - [CultIndex("narrow_ref")] + [CultRole("narrow_ref")] [CultReference(typeof(ParityLeafA))] public CultRecordRef NarrowRef; }