using System; using System.Collections.Concurrent; using System.Collections.Generic; using System.Diagnostics; using System.Globalization; using System.IO; using System.Linq; using System.Numerics; using System.Reflection; using System.Runtime.CompilerServices; using System.Runtime.ExceptionServices; using System.Security.Cryptography; using System.Text; using System.Threading; using System.Threading.Tasks; using MessagePack; using R3; namespace GameCult.Caching { public sealed class CultSchemaCatalogEntry { public string SchemaId { get; set; } = string.Empty; public string SchemaName { get; set; } = string.Empty; public string SchemaVersion { get; set; } = string.Empty; public string ContentHash { get; set; } = string.Empty; public string CanonicalSchemaJson { get; set; } = string.Empty; public string[] CompatibleSchemaIds { get; set; } = Array.Empty(); public CultSchemaMemberCatalogEntry[] Members { get; set; } = Array.Empty(); // The catalog a write leaves, derived from the records the written store holds and nothing else. For each schema id a record // carries, one entry that publishes it: for an id the write stages a record under, the entry the writer registered (or passed) // for it, which owns the id; for every other id the arrived (durable) entry that owns it, else the first that lists it as // compatible (the store that read it carries its bytes as read). A registered entry never publishes an id nothing staged sits under, so // the caller passes registered entries only for the ids it writes. Entries are written as chosen, never merged; a registered // descriptor wins over an arrived entry with the same own id. Two arrived entries with the same own id and different schema // names, or a record no chosen entry publishes, refuse the write. // Entries that tie are taken in one fixed order, so the catalog does not depend on the order they arrive in. private static IEnumerable Canonical(IEnumerable entries) => entries .OrderBy(entry => entry.SchemaName, StringComparer.Ordinal) .ThenBy(entry => entry.ContentHash, StringComparer.Ordinal) .ThenBy(entry => string.Join(",", entry.CompatibleSchemaIds), StringComparer.Ordinal); internal static CultSchemaCatalogEntry[] Derive( IEnumerable records, IReadOnlyCollection registered, IReadOnlyCollection arrived) { var chosen = new Dictionary(StringComparer.Ordinal); var carried = records.GroupBy(record => record.SchemaId, StringComparer.Ordinal).OrderBy(group => group.Key, StringComparer.Ordinal).ToArray(); foreach (var group in carried) { var schemaId = group.Key; var recordKey = group.Min(record => record.Key)!; var ownRegisteredAll = Canonical(registered.Where(entry => string.Equals(entry.SchemaId, schemaId, StringComparison.Ordinal))).ToArray(); var ownRegistered = ownRegisteredAll.FirstOrDefault(); var ownArrived = Canonical(arrived.Where(entry => string.Equals(entry.SchemaId, schemaId, StringComparison.Ordinal))).ToArray(); // Entries of one tier that share an own id must be one schema; a registered entry settles a disagreement among arrived ones. var disagreeing = ownRegisteredAll.Length > 0 ? ownRegisteredAll : ownArrived; if (disagreeing.Select(entry => entry.SchemaName).Distinct(StringComparer.Ordinal).Count() > 1) { var names = disagreeing.Select(entry => entry.SchemaName).Distinct(StringComparer.Ordinal).ToArray(); throw new CultSchemaConflictException( $"Schema id '{schemaId}' is claimed by schemas {string.Join(" and ", names.Select(name => $"'{name}'"))}; record '{recordKey}' cannot be written under it.", schemaId, names, recordKey); } CultSchemaCatalogEntry? pick; bool isRegistered; if (ownRegistered != null) { pick = ownRegistered; isRegistered = true; } else if (ownArrived.Length > 0) { pick = ownArrived[0]; isRegistered = false; } else { pick = Canonical(arrived.Where(entry => entry.CompatibleSchemaIds.Contains(schemaId, StringComparer.Ordinal))).FirstOrDefault(); isRegistered = false; } if (pick == null) continue; if (chosen.TryGetValue(pick.SchemaId, out var prior) && prior.Registered && !isRegistered) continue; chosen[pick.SchemaId] = (pick, isRegistered); } foreach (var group in carried) { if (!chosen.Values.Any(pair => string.Equals(pair.Entry.SchemaId, group.Key, StringComparison.Ordinal) || pair.Entry.CompatibleSchemaIds.Contains(group.Key, StringComparer.Ordinal))) { var recordKey = group.Min(record => record.Key)!; throw new CultSchemaConflictException( $"No schema entry publishes schema id '{group.Key}' of record '{recordKey}'.", group.Key, chosen.Values.Select(pair => pair.Entry.SchemaName).ToArray(), recordKey); } } return chosen.Values.Select(pair => pair.Entry).ToArray(); } } public sealed class CultSchemaMemberCatalogEntry { public int Slot { get; set; } public string MemberName { get; set; } = string.Empty; public string TypeName { get; set; } = string.Empty; public bool IsReference { get; set; } public bool IsMany { get; set; } public string? TargetSchemaName { get; set; } public bool IsName { get; set; } public string? IndexAlias { get; set; } } public sealed class CultPersistedRecord { public string Key { get; set; } = string.Empty; public string SchemaId { get; set; } = string.Empty; public string StoredAt { get; set; } = string.Empty; public byte[] Payload { get; set; } = Array.Empty(); // Non-null for a variant record (slot 4): its base and overrides. Its payload is empty. public CultVariantDelta? Variant { get; set; } } // One record a write puts in a single-file store: encoded, with the catalog entry that publishes the id it is stored under and // whether its document holds element ids (null when the writer cannot see, as for a raw payload). internal sealed class CultStagedRecord { internal CultStagedRecord(CultPersistedRecord record, CultSchemaCatalogEntry entry, bool? holdsIds) { Record = record; Entry = entry; HoldsIds = holdsIds; } internal CultPersistedRecord Record { get; } internal CultSchemaCatalogEntry Entry { get; } internal bool? HoldsIds { get; } } /// /// A record a store carries untouched because no registered type can read it, for example a record of a type this build does /// not have, or one renamed under a new schema name without declaring its old id. The store keeps its bytes as the file holds /// them, a write of another record keeps it, and a write that would replace or remove it is refused with a /// . The cache never holds it. Declaring its schema id on a type claims it at the next load. /// A record a type can read (through the catalog's schema name) but no type owns or declares is not foreign: the cache holds it /// read-only, and the same refusal applies to a write or removal of it. /// public sealed class CultForeignRecord { internal CultForeignRecord(string key, string schemaId, string schemaName, string storedAt, string storedVersion) { Key = key; SchemaId = schemaId; SchemaName = schemaName; StoredAt = storedAt; StoredVersion = storedVersion; } public string Key { get; } public string SchemaId { get; } // The name the store's catalog gives the schema. public string SchemaName { get; } public string StoredAt { get; } // The version of the record's stored bytes when its store read it, in that store's spelling, given by the store through // CacheBackingStore.Foreign. ExpectUnchanged compares it with the version on disk at the commit. internal string StoredVersion { get; } } public sealed class CultPersistedStoreSnapshot { public const string FormatV1 = "cultcache.store.v1"; public const string FormatV2 = "cultcache.store.v2"; public const string FormatV3 = "cultcache.store.v3"; public string FormatVersion { get; set; } = FormatV1; public CultSchemaCatalogEntry[] SchemaCatalog { get; set; } = Array.Empty(); public CultPersistedRecord[] Records { get; set; } = Array.Empty(); } public enum CultSchemaMigrationKind { Exact, CompatibleDrift } public sealed class CultSchemaMigrationWarning { public string Code { get; set; } = string.Empty; public string Message { get; set; } = string.Empty; } public sealed class CultSchemaMigrationReport { public string PersistedSchemaId { get; set; } = string.Empty; public string LocalSchemaId { get; set; } = string.Empty; public string PersistedSchemaName { get; set; } = string.Empty; public string LocalSchemaName { get; set; } = string.Empty; public CultSchemaMigrationKind Kind { get; set; } public int[] IgnoredExtraSlots { get; set; } = Array.Empty(); public int[] DefaultedMissingSlots { get; set; } = Array.Empty(); public CultSchemaMigrationWarning[] Warnings { get; set; } = Array.Empty(); } internal sealed class CultSchemaResolutionResult { public CultSchemaResolutionResult(CultDocumentDescriptor descriptor, CultSchemaMigrationReport report) { Descriptor = descriptor; Report = report; } public CultDocumentDescriptor Descriptor { get; } public CultSchemaMigrationReport Report { get; } } // A declared member's addressing shape, read-only: what a caller outside GameCult.Caching // (the CultNet selection evaluator, chiefly) is allowed to know about a member without // reflecting over the document itself. Built from the same declarations CultIndexAttribute // and CultReferenceAttribute already recorded; carries no value. public readonly struct CultDocumentMemberView { internal CultDocumentMemberView( string memberName, int slot, string? indexAlias, bool isName, bool isReference, bool isMany, bool isNumeric, Type? targetType) { MemberName = memberName; Slot = slot; IndexAlias = indexAlias; IsName = isName; IsReference = isReference; IsMany = isMany; IsNumeric = isNumeric; TargetType = targetType; } public string MemberName { get; } public int Slot { get; } public string? IndexAlias { get; } public bool IsName { get; } public bool IsReference { get; } public bool IsMany { get; } public bool IsNumeric { get; } // The reference's declared target: the type CultReferenceAttribute names, or CultRecordRef's T. // Null when the member is not a reference. May be abstract (D9): the leaf set is resolved at // evaluation by CultDocumentRegistry.ResolveTargetLeaves, never persisted here or in the catalog. public Type? TargetType { get; } } public sealed class CultDocumentDescriptor { internal CultDocumentDescriptor( Type documentType, string schemaName, string schemaVersion, string schemaId, IReadOnlyList compatibleSchemaIds, string contentHash, string canonicalSchemaJson, bool isGlobal, string? nameMember, Func? nameAccessor, IReadOnlyDictionary> indexAccessors, IReadOnlyList members, IReadOnlyList richMembers) { DocumentType = documentType; SchemaName = schemaName; SchemaVersion = schemaVersion; SchemaId = schemaId; CompatibleSchemaIds = compatibleSchemaIds; ContentHash = contentHash; CanonicalSchemaJson = canonicalSchemaJson; IsGlobal = isGlobal; NameMember = nameMember; NameAccessor = nameAccessor; IndexAccessors = indexAccessors; Members = members; RichMembers = richMembers; DeclaredMembers = richMembers .Select(member => new CultDocumentMemberView( member.Member.Name, member.Slot, member.IndexAlias, member.IsName, member.IsReference, member.IsMany, member.IsNumeric, member.TargetType)) .ToArray(); } public Type DocumentType { get; } public string SchemaName { get; } public string SchemaVersion { get; } public string SchemaId { get; } /// The ids besides this type declared through . public IReadOnlyList CompatibleSchemaIds { get; } public string ContentHash { get; } public string CanonicalSchemaJson { get; } public bool IsGlobal { get; } public string? NameMember { get; } internal Func? NameAccessor { get; } internal IReadOnlyDictionary> IndexAccessors { get; } internal IReadOnlyList Members { get; } internal IReadOnlyList RichMembers { get; } // The public read surface (CultNet selection cut 1, section 6): declared addressing, and the // three ways to read a value through it. GameCult.Networking has no InternalsVisibleTo here on // purpose - this is the only door, so nothing outside the cache decides what a member means. public IReadOnlyList DeclaredMembers { get; } public bool TryGetIndexValue(object document, string alias, out string? value) { if (document == null) throw new ArgumentNullException(nameof(document)); if (string.IsNullOrEmpty(alias)) { value = null; return false; } foreach (var member in RichMembers) { if (string.Equals(member.IndexAlias, alias, StringComparison.Ordinal)) { value = member.GetterNullable(document); return true; } } value = null; return false; } public bool TryGetIndexNumber(object document, string alias, out string? value) { if (document == null) throw new ArgumentNullException(nameof(document)); if (!string.IsNullOrEmpty(alias)) { foreach (var member in RichMembers) { if (string.Equals(member.IndexAlias, alias, StringComparison.Ordinal) && member.IsNumeric) { var number = member.NumberOf(document); if (number != null) { value = number; return true; } break; } } } value = null; return false; } // The edges a declared reference member carries: one per CultRecordRef, one per element of // an IEnumerable>, one per key of an IDictionary, V> with the // value as the payload (D11). role is the member's index alias, or its name when it has none. public IEnumerable<(CultRecordKey Target, object? Payload)> ReferencesOf(object document, string role) { if (document == null) throw new ArgumentNullException(nameof(document)); if (string.IsNullOrEmpty(role)) return Array.Empty<(CultRecordKey, object?)>(); foreach (var member in RichMembers) { if (!member.IsReference) continue; var memberRole = member.IndexAlias ?? member.Member.Name; if (string.Equals(memberRole, role, StringComparison.Ordinal)) { return member.ReferenceEnumerator!(document); } } return Array.Empty<(CultRecordKey, object?)>(); } public CultSchemaCatalogEntry ToCatalogEntry() { return new CultSchemaCatalogEntry { SchemaId = SchemaId, SchemaName = SchemaName, SchemaVersion = SchemaVersion, ContentHash = ContentHash, CanonicalSchemaJson = CanonicalSchemaJson, CompatibleSchemaIds = [SchemaId, .. CompatibleSchemaIds], Members = Members .OrderBy(member => member.Slot) .Select(member => new CultSchemaMemberCatalogEntry { Slot = member.Slot, MemberName = member.MemberName, TypeName = member.TypeName, IsReference = member.IsReference, IsMany = member.IsMany, TargetSchemaName = member.TargetSchemaName, IsName = member.IsName, IndexAlias = member.IndexAlias }) .ToArray() }; } internal string GetPreferredFileStem(object document) { var preferred = NameAccessor?.Invoke(document); return string.IsNullOrWhiteSpace(preferred) ? SchemaName : preferred!; } } internal sealed class CultDocumentMemberDescriptor { public string MemberName { get; set; } = string.Empty; public int Slot { get; set; } public string TypeName { get; set; } = string.Empty; public bool IsReference { get; set; } public bool IsMany { get; set; } public string? TargetSchemaName { get; set; } public bool IsName { get; set; } public string? IndexAlias { get; set; } } public sealed class CultStoredDocument { public CultStoredDocument( CultRecordKey key, string storedAt, CultDocumentDescriptor descriptor, object document) { Key = key; StoredAt = storedAt; Descriptor = descriptor; StoredSchemaId = descriptor.SchemaId; StoredVersion = UnstoredVersion(); _document = document; } // A variant: the delta is what is stored; the cache resolves it against its base in the hold that admits it. internal CultStoredDocument( CultRecordKey key, string storedAt, CultDocumentDescriptor descriptor, CultVariantDelta variant, object? resolved, CultCodec? codec) { Key = key; StoredAt = storedAt; Descriptor = descriptor; StoredSchemaId = descriptor.SchemaId; StoredVersion = UnstoredVersion(); Variant = variant; _document = resolved; _codec = codec; } private readonly object? _document; // What reads a variant's override values; null for a plain record. private readonly CultCodec? _codec; public CultRecordKey Key { get; } // When the record was stored: metadata. A write that stores the record mints it a storedAt later than the one it replaces; // no condition reads it. public string StoredAt { get; internal set; } public CultDocumentDescriptor Descriptor { get; } // The schema id the record carries in its store: the id it was loaded under, which may be an older or foreign id that // resolved to Descriptor, until a write of this record puts it under Descriptor's id. public string StoredSchemaId { get; internal set; } // Names the bytes this record was read from or written as: the SHA-256 of its persisted encoding, set by the store that // read or wrote them. A record no store has read or written has a version of its own that equals nothing else. internal string StoredVersion { get; set; } private static string UnstoredVersion() => "unstored:" + Guid.NewGuid().ToString("N"); // The complete document. For a variant, the resolved view: derived, never written back. public object Document => _document ?? throw new InvalidOperationException($"Variant {Key.Value} has not been resolved; only the cache resolves a variant."); // The base and overrides a variant record stores; null for a plain record. public CultVariantDelta? Variant { get; } // A plain record staged by Flatten: the one way a plain record may replace a variant at its key. internal bool Flattens { get; set; } // True while this record holds element ids minted when it loaded, that no write has persisted (see MintElementIds). internal bool IdsInMemoryOnly { get; set; } // True when some element this record persists holds a non-empty id, and the only thing a store reads to mark its header. // It is read off the stored form each time it is asked, so a store asks it as it writes: a document changed after it was // admitted, or loaded and changed in place, writes the header its content needs. A plain record persists its document; a // variant persists its overrides, so it holds an id when an override value does (its base's ids are the base's). A // variant built without a codec (a store loaded it with no cache attached) cannot see into its values and counts as holding // one, so a store never writes an unmarked header over ids it could not see. public bool HoldsIds { get { if (Variant == null) return CultElementIds.Holds(Descriptor.DocumentType, _document); if (_codec == null) return true; return Variant.Overrides.Any(entry => { var memberType = Descriptor.RichMembers.First(member => member.Slot == entry.Path[0].Slot).MemberType; return CultElementIds.Holds(memberType, _codec.Deserialize(memberType, Descriptor.DocumentType, entry.Value)); }); } } internal CultStoredDocument Resolved(object document, CultVariantDelta delta, bool idsInMemoryOnly, CultCodec codec) => new(Key, StoredAt, Descriptor, delta, document, codec) { IdsInMemoryOnly = idsInMemoryOnly, StoredSchemaId = StoredSchemaId, StoredVersion = StoredVersion }; } public sealed class CultDocumentRegistry { private static readonly Lazy SharedRegistry = new(() => new CultDocumentRegistry()); private volatile RegistryIndexes _indexes = new(); private readonly object _registrationGate = new(); public static CultDocumentRegistry Shared => SharedRegistry.Value; public CultDocumentRegistry() { Refresh(); } private CultDocumentRegistry(bool discoverLoadedDocuments) { if (discoverLoadedDocuments) Refresh(); } public static CultDocumentRegistry ForTypes(IEnumerable documentTypes) { if (documentTypes == null) throw new ArgumentNullException(nameof(documentTypes)); var registry = new CultDocumentRegistry(discoverLoadedDocuments: false); foreach (var type in documentTypes.Distinct()) { registry.GetRequired(type ?? throw new ArgumentException( "Document type collections cannot contain null entries.", nameof(documentTypes))); } return registry; } public IEnumerable AllDescriptors => _indexes.ByType.Values.OrderBy(d => d.SchemaName, StringComparer.Ordinal); // D9: a reference's target set is every registered leaf assignable to its declared target type, // computed against the registry as it stands right now - never cached, never persisted. A leaf // registered after some other descriptor was built must still show up here. public IReadOnlyList ResolveTargetLeaves(Type targetType) { if (targetType == null) throw new ArgumentNullException(nameof(targetType)); return AllDescriptors.Where(d => targetType.IsAssignableFrom(d.DocumentType)).ToArray(); } public void Refresh() { lock (_registrationGate) { var rebuilt = new RegistryIndexes(); foreach (var type in ReflectionExtensions.GetAttributedDocumentTypes()) { RegisterDescriptor(rebuilt, BuildDescriptor(type)); } _indexes = rebuilt; } } public CultDocumentDescriptor GetRequired(Type type) { if (_indexes.ByType.TryGetValue(type, out var descriptor)) { return descriptor; } var attribute = type.GetCustomAttribute(); if (attribute == null) { throw new InvalidOperationException( $"Type {type.FullName} is not marked with {nameof(CultDocumentAttribute)}."); } descriptor = BuildDescriptor(type); return RegisterDescriptor(descriptor); } public CultDocumentDescriptor GetRequired() where T : class { return GetRequired(typeof(T)); } public CultDocumentDescriptor GetRequiredBySchemaId(string schemaId) { if (_indexes.BySchemaId.TryGetValue(schemaId, out var descriptor)) { return descriptor; } throw new InvalidOperationException($"Unknown CultCache schema id '{schemaId}'."); } public CultDocumentDescriptor ResolvePersistedSchema(string schemaId, IReadOnlyCollection catalog) { return ResolvePersistedSchemaDetailed(schemaId, catalog).Descriptor; } public CultSchemaMigrationReport ResolvePersistedSchemaReport(string schemaId, IReadOnlyCollection catalog) { return ResolvePersistedSchemaDetailed(schemaId, catalog).Report; } internal CultSchemaResolutionResult ResolvePersistedSchemaDetailed(string schemaId, IReadOnlyCollection catalog) { var local = Claimant(schemaId, catalog, out var persisted); if (local == null) throw new InvalidOperationException( $"No local CultCache schema matches persisted schema '{persisted!.SchemaName}' ({schemaId})."); if (persisted == null) { return new CultSchemaResolutionResult( local, new CultSchemaMigrationReport { PersistedSchemaId = schemaId, LocalSchemaId = local.SchemaId, PersistedSchemaName = local.SchemaName, LocalSchemaName = local.SchemaName, Kind = CultSchemaMigrationKind.Exact }); } return BuildCompatibleResolutionResult(persisted, local); } // The registered type a record under this schema id belongs to, or null when no registered type claims it: a foreign record, // which a store carries untouched. Deciding this never decodes the record or compares shapes, so a store can classify what a // file holds without reading a payload. persisted is the catalog entry the record resolves through, and null when a // registered type owns the id. Claimants that cannot be told apart refuse. Branches past the listers (a local id the persisted // entry lists, then the schema name) pick a type to read with. They confer no ownership: Declares decides what a cache may write. internal CultDocumentDescriptor? Claimant( string schemaId, IReadOnlyCollection catalog, out CultSchemaCatalogEntry? persisted) { var indexes = _indexes; persisted = null; if (indexes.BySchemaId.TryGetValue(schemaId, out var exact)) return exact; persisted = PublishedBy(schemaId, catalog); if (persisted == null) { throw new InvalidOperationException($"Persisted schema '{schemaId}' is not present in the embedded catalog."); } // No type owns the id, so the one type that lists it is the record's type. Two listers and no owner cannot say which. if (indexes.ByCompatibleSchemaId.TryGetValue(schemaId, out var listers)) { if (listers.Length > 1) throw new CultSchemaConflictException( $"CultCache schema id '{schemaId}' is owned by no registered type and declared compatible by several " + $"({string.Join(", ", listers.Select(lister => lister.DocumentType.FullName))}), so a record under it resolves to none. " + "Register the type that owns the id, or declare it on one type.", schemaId, listers.Select(lister => lister.SchemaName).ToArray(), string.Empty); return listers[0]; } foreach (var compatibleSchemaId in persisted.CompatibleSchemaIds.Where(candidate => !string.IsNullOrWhiteSpace(candidate))) { if (indexes.BySchemaId.TryGetValue(compatibleSchemaId, out var compatibleLocal)) return compatibleLocal; } if (indexes.BySchemaName.TryGetValue(persisted.SchemaName, out var localVersions)) { if (localVersions.Length == 1) return localVersions[0]; throw new InvalidOperationException( $"CultCache schema name '{persisted.SchemaName}' is ambiguous across local versions " + $"[{string.Join(", ", localVersions.Select(candidate => candidate.SchemaVersion).OrderBy(version => version, StringComparer.Ordinal))}]. " + "Persisted schema compatibility must identify an explicit schema id."); } return null; } // The catalog entry a record under this id is published by: the entry that owns the id, else the first that lists it compatible. internal static CultSchemaCatalogEntry? PublishedBy(string schemaId, IReadOnlyCollection catalog) => catalog.FirstOrDefault(entry => string.Equals(entry.SchemaId, schemaId, StringComparison.Ordinal)) ?? catalog.FirstOrDefault(entry => entry.CompatibleSchemaIds.Contains(schemaId, StringComparer.Ordinal)); // The one answer to whether a cache may write, relabel or remove a record stored under this id: a registered type owns it or // declares it compatible. Read live, so a type registered later that declares the id claims its records at their next write. internal bool Declares(string schemaId) { var indexes = _indexes; return indexes.BySchemaId.ContainsKey(schemaId) || indexes.ByCompatibleSchemaId.ContainsKey(schemaId); } private CultDocumentDescriptor RegisterDescriptor(CultDocumentDescriptor descriptor) { lock (_registrationGate) { var current = _indexes; if (current.ByType.TryGetValue(descriptor.DocumentType, out var registered)) return registered; var updated = new RegistryIndexes(current); registered = RegisterDescriptor(updated, descriptor); _indexes = updated; return registered; } } private static CultDocumentDescriptor RegisterDescriptor( RegistryIndexes indexes, CultDocumentDescriptor descriptor) { if (indexes.ByType.TryGetValue(descriptor.DocumentType, out var registered)) return registered; // One type owns a schema, whatever its declaration. Its id hashes the schema's name, version and members, so a second // type of the same name and version is refused here whether or not it hashes alike. Other types may list the id as // compatible beside its owner, as a v2 class does that keeps v1's class registered: a record under the id resolves to // its owner, and to a lister only when nothing owns it. Neither depends on the order types were registered in. indexes.BySchemaName.TryGetValue(descriptor.SchemaName, out var schemaNameVersions); var schemaVersionOwner = schemaNameVersions?.FirstOrDefault(candidate => string.Equals(candidate.SchemaVersion, descriptor.SchemaVersion, StringComparison.Ordinal)); if (schemaVersionOwner != null) throw SchemaClaimedTwice(schemaVersionOwner, descriptor); indexes.ByType[descriptor.DocumentType] = descriptor; indexes.BySchemaId[descriptor.SchemaId] = descriptor; foreach (var compatibleSchemaId in descriptor.CompatibleSchemaIds) indexes.ByCompatibleSchemaId[compatibleSchemaId] = indexes.ByCompatibleSchemaId.TryGetValue(compatibleSchemaId, out var listers) ? [.. listers, descriptor] : [descriptor]; indexes.BySchemaName[descriptor.SchemaName] = schemaNameVersions == null ? [descriptor] : [.. schemaNameVersions, descriptor]; return descriptor; } private static CultSchemaConflictException SchemaClaimedTwice( CultDocumentDescriptor existing, CultDocumentDescriptor claimed) => new( $"CultCache schema '{claimed.SchemaName}' version '{claimed.SchemaVersion}' is already registered to CLR type " + $"'{existing.DocumentType.FullName}' and cannot also be claimed by '{claimed.DocumentType.FullName}'. " + "A schema is owned by one registered type; others may only declare its id compatible.", claimed.SchemaId, new[] { existing.SchemaName, claimed.SchemaName }, string.Empty); private sealed class RegistryIndexes { public RegistryIndexes() { ByType = new Dictionary(); BySchemaId = new Dictionary(StringComparer.Ordinal); ByCompatibleSchemaId = new Dictionary(StringComparer.Ordinal); BySchemaName = new Dictionary(StringComparer.Ordinal); } public RegistryIndexes(RegistryIndexes source) { ByType = new Dictionary(source.ByType); BySchemaId = new Dictionary(source.BySchemaId, StringComparer.Ordinal); ByCompatibleSchemaId = new Dictionary(source.ByCompatibleSchemaId, StringComparer.Ordinal); BySchemaName = new Dictionary(source.BySchemaName, StringComparer.Ordinal); } public Dictionary ByType { get; } public Dictionary BySchemaId { get; } // Every type declaring the id compatible, in registration order; resolution never depends on that order. public Dictionary ByCompatibleSchemaId { get; } public Dictionary BySchemaName { get; } } private static CultDocumentDescriptor BuildDescriptor(Type type) { var attribute = type.GetCustomAttribute() ?? throw new InvalidOperationException( $"Type {type.FullName} is not marked with {nameof(CultDocumentAttribute)}."); var members = DiscoverMembers(type); var ids = CultElementIds.Inspect(type); if (ids.Problems.Count > 0) throw new InvalidOperationException(ids.Problems[0]); var nameMember = members.FirstOrDefault(member => member.IsName); var indexAccessors = members .Where(member => member.IndexAlias != null) .ToDictionary( member => member.IndexAlias!, member => member.Getter, StringComparer.Ordinal); var descriptorMembers = members .Select(member => new CultDocumentMemberDescriptor { MemberName = member.Member.Name, Slot = member.Slot, TypeName = CultSchemaTypeNames.FromType(member.MemberType), IsReference = member.IsReference, IsMany = member.IsMany, TargetSchemaName = member.TargetSchemaName, IsName = member.IsName, IndexAlias = member.IndexAlias }) .ToArray(); var schemaJson = BuildCanonicalSchemaJson(attribute.SchemaName, attribute.SchemaVersion, descriptorMembers); var contentHash = Sha256(schemaJson); var semanticFingerprint = BuildSemanticFingerprint(attribute.SchemaName, attribute.SchemaVersion, descriptorMembers); var schemaId = Sha256(semanticFingerprint); var compatibleSchemaIds = attribute.CompatibleSchemaIds .Where(id => !string.IsNullOrWhiteSpace(id) && !string.Equals(id, schemaId, StringComparison.Ordinal)) .Distinct(StringComparer.Ordinal) .ToArray(); return new CultDocumentDescriptor( type, attribute.SchemaName, attribute.SchemaVersion, schemaId, compatibleSchemaIds, contentHash, schemaJson, type.GetCustomAttribute() != null, nameMember?.Member.Name, nameMember?.GetterNullable, indexAccessors, descriptorMembers, members); } private static string BuildSemanticFingerprint( string schemaName, string schemaVersion, IReadOnlyList members) { var builder = new StringBuilder(); builder.Append(schemaName) .Append('|') .Append(schemaVersion); foreach (var member in members.OrderBy(member => member.Slot)) { builder.Append('|') .Append(member.Slot) .Append(':') .Append(member.MemberName) .Append(':') .Append(member.TypeName) .Append(':') .Append(member.IsReference ? "ref" : "value") .Append(':') .Append(member.TargetSchemaName ?? string.Empty) .Append(':') .Append(member.IsMany ? "many" : "one"); } return builder.ToString(); } private static string BuildCanonicalSchemaJson( string schemaName, string schemaVersion, IReadOnlyList members) { var builder = new StringBuilder(); builder.Append("{\"schemaName\":\"") .Append(CultSchemaTypeNames.EscapeForLiteral(schemaName)) .Append("\",\"schemaVersion\":\"") .Append(CultSchemaTypeNames.EscapeForLiteral(schemaVersion)) .Append("\",\"members\":["); for (var index = 0; index < members.Count; index++) { var member = members[index]; if (index > 0) { builder.Append(','); } builder.Append("{\"slot\":") .Append(member.Slot) .Append(",\"name\":\"") .Append(CultSchemaTypeNames.EscapeForLiteral(member.MemberName)) .Append("\",\"type\":\"") .Append(CultSchemaTypeNames.EscapeForLiteral(member.TypeName)) .Append("\",\"isReference\":") .Append(member.IsReference ? "true" : "false") .Append(",\"many\":") .Append(member.IsMany ? "true" : "false") .Append(",\"targetSchemaName\":"); if (member.TargetSchemaName == null) { builder.Append("null"); } else { builder.Append('"').Append(CultSchemaTypeNames.EscapeForLiteral(member.TargetSchemaName)).Append('"'); } builder.Append(",\"indexAlias\":"); if (member.IndexAlias == null) { builder.Append("null"); } else { builder.Append('"').Append(CultSchemaTypeNames.EscapeForLiteral(member.IndexAlias)).Append('"'); } builder.Append(",\"isName\":") .Append(member.IsName ? "true" : "false") .Append('}'); } builder.Append("]}"); return builder.ToString(); } private static CultSchemaResolutionResult BuildCompatibleResolutionResult( CultSchemaCatalogEntry persisted, CultDocumentDescriptor local) { if (string.Equals(persisted.ContentHash, local.ContentHash, StringComparison.Ordinal)) { return new CultSchemaResolutionResult( local, new CultSchemaMigrationReport { PersistedSchemaId = persisted.SchemaId, LocalSchemaId = local.SchemaId, PersistedSchemaName = persisted.SchemaName, LocalSchemaName = local.SchemaName, Kind = CultSchemaMigrationKind.Exact }); } var persistedSchema = new CanonicalSchemaShape( persisted.SchemaName, persisted.SchemaVersion, persisted.Members.Select(member => new CanonicalSchemaMember( member.Slot, member.MemberName, member.TypeName, member.IsReference, member.IsMany, member.TargetSchemaName, member.IsName, member.IndexAlias)).ToArray()); var localSchema = new CanonicalSchemaShape( local.SchemaName, local.SchemaVersion, local.Members.Select(member => new CanonicalSchemaMember( member.Slot, member.MemberName, member.TypeName, member.IsReference, member.IsMany, member.TargetSchemaName, member.IsName, member.IndexAlias)).ToArray()); var comparison = CompareSchemaShapes(persistedSchema, localSchema); if (!comparison.IsCompatible) { throw new InvalidOperationException( $"Persisted schema '{persisted.SchemaName}' ({persisted.SchemaId}) is incompatible with local schema '{local.SchemaName}' ({local.SchemaId}): {string.Join("; ", comparison.Errors)}"); } var warnings = new List(); if (!string.Equals(persisted.SchemaId, local.SchemaId, StringComparison.Ordinal)) { warnings.Add(new CultSchemaMigrationWarning { Code = "compatible_schema_id_drift", Message = $"Persisted schema '{persisted.SchemaId}' resolved to local schema '{local.SchemaId}'." }); } if (!string.Equals(persisted.ContentHash, local.ContentHash, StringComparison.Ordinal)) { warnings.Add(new CultSchemaMigrationWarning { Code = "content_hash_drift", Message = $"Persisted schema content hash '{persisted.ContentHash}' differs from local hash '{local.ContentHash}'." }); } warnings.AddRange(comparison.Warnings); return new CultSchemaResolutionResult( local, new CultSchemaMigrationReport { PersistedSchemaId = persisted.SchemaId, LocalSchemaId = local.SchemaId, PersistedSchemaName = persisted.SchemaName, LocalSchemaName = local.SchemaName, Kind = CultSchemaMigrationKind.CompatibleDrift, IgnoredExtraSlots = comparison.IgnoredExtraSlots.OrderBy(slot => slot).ToArray(), DefaultedMissingSlots = comparison.DefaultedMissingSlots.OrderBy(slot => slot).ToArray(), Warnings = warnings.ToArray() }); } private static SchemaShapeComparison CompareSchemaShapes(CanonicalSchemaShape persisted, CanonicalSchemaShape local) { var errors = new List(); var warnings = new List(); var ignoredExtraSlots = new List(); var defaultedMissingSlots = new List(); var persistedBySlot = persisted.Members.ToDictionary(member => member.Slot); var localBySlot = local.Members.ToDictionary(member => member.Slot); foreach (var localMember in local.Members.OrderBy(member => member.Slot)) { if (!persistedBySlot.TryGetValue(localMember.Slot, out var persistedMember)) { defaultedMissingSlots.Add(localMember.Slot); warnings.Add(new CultSchemaMigrationWarning { Code = "defaulted_missing_slot", Message = $"Local slot {localMember.Slot} ('{localMember.Name}') is missing from the persisted schema and will use the local default value." }); continue; } if (!string.Equals(localMember.TypeName, persistedMember.TypeName, StringComparison.Ordinal)) { errors.Add($"slot {localMember.Slot} changed type '{persistedMember.TypeName}' -> '{localMember.TypeName}'"); } // 1.0.x persisted a CultRecordRef collection as a plain value slot (not a reference, no target); the // payload is the same array of keys. Gaining the reference classification on an otherwise identical slot // is compatible drift. Losing it, or any other change of reference or target, is not. var gainedReference = !persistedMember.IsReference && persistedMember.TargetSchemaName == null && localMember.IsReference && localMember.TargetSchemaName != null; if (gainedReference) { warnings.Add(new CultSchemaMigrationWarning { Code = "gained_reference_semantics", Message = $"Local slot {localMember.Slot} ('{localMember.Name}') is a reference to '{localMember.TargetSchemaName}'; the persisted schema stored it as a plain value with the same payload." }); } else if (localMember.IsReference != persistedMember.IsReference) { errors.Add($"slot {localMember.Slot} changed reference semantics"); } if (localMember.IsMany != persistedMember.IsMany) { errors.Add($"slot {localMember.Slot} changed cardinality"); } if (!gainedReference && !string.Equals(localMember.TargetSchemaName, persistedMember.TargetSchemaName, StringComparison.Ordinal)) { errors.Add($"slot {localMember.Slot} changed target schema '{persistedMember.TargetSchemaName ?? ""}' -> '{localMember.TargetSchemaName ?? ""}'"); } if (localMember.IsName != persistedMember.IsName) { errors.Add($"slot {localMember.Slot} changed name-lookup semantics"); } if (!string.Equals(localMember.IndexAlias, persistedMember.IndexAlias, StringComparison.Ordinal)) { errors.Add($"slot {localMember.Slot} changed index alias '{persistedMember.IndexAlias ?? ""}' -> '{localMember.IndexAlias ?? ""}'"); } } foreach (var persistedMember in persisted.Members.OrderBy(member => member.Slot)) { if (localBySlot.ContainsKey(persistedMember.Slot)) { continue; } ignoredExtraSlots.Add(persistedMember.Slot); warnings.Add(new CultSchemaMigrationWarning { Code = "ignored_extra_slot", Message = $"Persisted slot {persistedMember.Slot} ('{persistedMember.Name}') is not present locally and will be ignored." }); } return new SchemaShapeComparison( errors.Count == 0, errors.ToArray(), warnings.ToArray(), ignoredExtraSlots.ToArray(), defaultedMissingSlots.ToArray()); } private sealed class CanonicalSchemaShape { public CanonicalSchemaShape(string schemaName, string schemaVersion, CanonicalSchemaMember[] members) { SchemaName = schemaName; SchemaVersion = schemaVersion; Members = members; } public string SchemaName { get; } public string SchemaVersion { get; } public CanonicalSchemaMember[] Members { get; } } private sealed class CanonicalSchemaMember { public CanonicalSchemaMember( int slot, string name, string typeName, bool isReference, bool isMany, string? targetSchemaName, bool isName, string? indexAlias) { Slot = slot; Name = name; TypeName = typeName; IsReference = isReference; IsMany = isMany; TargetSchemaName = targetSchemaName; IsName = isName; IndexAlias = indexAlias; } public int Slot { get; } public string Name { get; } public string TypeName { get; } public bool IsReference { get; } public bool IsMany { get; } public string? TargetSchemaName { get; } public bool IsName { get; } public string? IndexAlias { get; } } private sealed class SchemaShapeComparison { public SchemaShapeComparison( bool isCompatible, string[] errors, CultSchemaMigrationWarning[] warnings, int[] ignoredExtraSlots, int[] defaultedMissingSlots) { IsCompatible = isCompatible; Errors = errors; Warnings = warnings; IgnoredExtraSlots = ignoredExtraSlots; DefaultedMissingSlots = defaultedMissingSlots; } public bool IsCompatible { get; } public string[] Errors { get; } public CultSchemaMigrationWarning[] Warnings { get; } public int[] IgnoredExtraSlots { get; } public int[] DefaultedMissingSlots { get; } } // Public instance fields and properties (get-only included: MessagePack writes every keyed public member), most-derived // declaration first; GetCustomAttributes inherits a property's attributes up its override chain, as MessagePack's resolver reads them. private static IReadOnlyList DiscoverMembers(Type type) { const BindingFlags declared = BindingFlags.DeclaredOnly | BindingFlags.Instance | BindingFlags.Public; var messagePackObject = type.GetCustomAttribute(true); var allowPrivate = messagePackObject?.AllowPrivate == true; // Writable: MessagePack can assign the member after construction. A member that is not must be filled by the constructor. var candidates = new List<(MemberInfo Member, Type MemberType, int Depth, bool Writable, PropertyInfo? DivergentRoot)>(); var seenRoots = new HashSet<(Type, string)>(); var depth = 0; for (var current = type; current != null && current != typeof(object); current = current.BaseType, depth++) { foreach (var field in current.GetFields(declared)) { // MessagePack's rule: a readonly field is writable under AllowPrivate only. if (!IsIgnored(field)) candidates.Add((field, field.FieldType, depth, !field.IsInitOnly || allowPrivate, null)); } foreach (var property in current.GetProperties(declared)) { if (property.GetIndexParameters().Length > 0 || !seenRoots.Add(RootOf(property))) continue; var chain = OverrideChain(property).ToArray(); var setter = chain.Select(declaration => declaration.SetMethod).FirstOrDefault(method => method != null); if (chain.All(declaration => declaration.GetMethod == null)) continue; // MessagePack's resolver reads the base declaration's [Key]/[IgnoreMember]; an override that changes them is refused. var root = chain[chain.Length - 1]; var divergentRoot = chain.Any(declaration => !Equals(GetKeyValue(declaration), GetKeyValue(root)) || IsIgnored(declaration) != IsIgnored(root)) ? root : null; if (divergentRoot == null && IsIgnored(property)) continue; // MessagePack's own rule: it calls a setter that is public, or any setter (init-only included) under AllowPrivate. candidates.Add((property, property.PropertyType, depth, setter != null && (setter.IsPublic || allowPrivate), divergentRoot)); } } var rejections = new List(); if (messagePackObject == null) rejections.Add(NotMessagePackObjectMessage(type.Name)); else if (!type.IsVisible && !allowPrivate) rejections.Add(NotVisibleMessage(type.Name)); else { // AllowPrivate makes MessagePack read non-public members, which the registry does not persist; without it MessagePack // silently skips a keyed non-public member. const BindingFlags nonPublic = BindingFlags.DeclaredOnly | BindingFlags.Instance | BindingFlags.NonPublic; for (var current = type; current != null && current != typeof(object); current = current.BaseType) { foreach (var member in current.GetFields(nonPublic).Cast().Concat(current.GetProperties(nonPublic))) { if (allowPrivate && !member.IsDefined(typeof(CompilerGeneratedAttribute)) && !IsIgnored(member)) rejections.Add(UnmarkedNonPublicMemberMessage(type.Name, Qualified(member))); else if (!allowPrivate && GetKeyValue(member) != null && !IsIgnored(member)) rejections.Add(KeyedNonPublicMemberMessage(type.Name, Qualified(member))); } } } var keyed = new List<(MemberInfo Member, Type MemberType, int Slot, bool Writable)>(); foreach (var candidate in candidates.OrderBy(candidate => candidate.Depth).ThenBy(candidate => candidate.Member.Name, StringComparer.Ordinal)) { var key = GetKeyValue(candidate.Member); if (candidate.DivergentRoot != null) rejections.Add(DivergentOverrideMessage(type.Name, Qualified(candidate.Member), Qualified(candidate.DivergentRoot))); else if (key is string) rejections.Add(StringKeyMessage(type.Name, candidate.Member.Name)); else if (key is not int slot) rejections.Add(UnkeyedMemberMessage(type.Name, candidate.Member.Name)); else keyed.Add((candidate.Member, candidate.MemberType, slot, candidate.Writable)); } if (messagePackObject != null) { // A member MessagePack cannot assign is read back only through the picked constructor's parameter at its slot. var arity = ConstructorArity(type, allowPrivate, keyed.Select(entry => (entry.MemberType, entry.Slot))); if (arity == null) rejections.Add(NoConstructorMessage(type.Name)); else { foreach (var entry in keyed.Where(entry => !entry.Writable && entry.Slot >= arity)) { rejections.Add(entry.Member switch { FieldInfo => ReadonlyUnfilledMessage(type.Name, entry.Member.Name, entry.Slot), PropertyInfo property when property.SetMethod == null && OverrideChain(property).All(declaration => declaration.SetMethod == null) => GetOnlyUnfilledMessage(type.Name, entry.Member.Name, entry.Slot), _ => NonPublicSetterMessage(type.Name, entry.Member.Name) }); } } } foreach (var group in keyed.GroupBy(entry => entry.Slot).Where(group => group.Count() > 1).OrderBy(group => group.Key)) { var pair = group.Take(2).ToArray(); rejections.Add(DuplicateSlotMessage(type.Name, Qualified(pair[0].Member), Qualified(pair[1].Member), group.Key)); } foreach (var group in keyed.GroupBy(entry => entry.Member.Name) .Where(group => group.Select(entry => entry.Slot).Distinct().Count() > 1) .OrderBy(group => group.Key, StringComparer.Ordinal)) { var pair = group.Take(2).ToArray(); rejections.Add(HiddenMemberMessage(type.Name, Qualified(pair[0].Member), Qualified(pair[1].Member))); } // 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. foreach (var group in keyed .Select(entry => (Entry: entry, Alias: ResolveIndexAlias(entry.Member))) .Where(pair => pair.Alias != null) .GroupBy(pair => pair.Alias, StringComparer.Ordinal) .Where(group => group.Select(pair => pair.Entry.Member.Name).Distinct().Count() > 1) .OrderBy(group => group.Key, StringComparer.Ordinal)) { var pair = group.Take(2).ToArray(); rejections.Add(DuplicateIndexAliasMessage(type.Name, Qualified(pair[0].Entry.Member), Qualified(pair[1].Entry.Member), group.Key!)); } // D10b: the reference-role namespace is (IndexAlias ?? MemberName) for every member, not // only the aliased ones D10 groups above. A member with no [CultIndex] still contributes // its bare name to that namespace, and a reference member declaring the same string as its // alias collides with it - the collision D10 cannot see because an unaliased member's Alias // is null. foreach (var group in keyed .Select(entry => ( Entry: entry, Role: ResolveIndexAlias(entry.Member) ?? entry.Member.Name, IsReference: PersistedMember.IsReferenceCandidate(entry.Member, entry.MemberType))) .GroupBy(pair => pair.Role, StringComparer.Ordinal) .Where(group => group.Select(pair => pair.Entry.Member.Name).Distinct().Count() > 1 && group.Any(pair => pair.IsReference)) .OrderBy(group => group.Key, StringComparer.Ordinal)) { var pair = group.Take(2).ToArray(); rejections.Add(DuplicateReferenceRoleMessage(type.Name, Qualified(pair[0].Entry.Member), Qualified(pair[1].Entry.Member), group.Key)); } // D11: a member declared as a reference (by [CultReference] or a bare CultRecordRef type) // whose CLR shape the cache cannot walk. many:false needs CultRecordRef itself; many:true // needs IEnumerable> or IDictionary, V>. A declaration the // cache cannot read is refused here, not answered as an unreadable no-op reference later. foreach (var candidate in keyed) { var referenceAttribute = candidate.Member.GetCustomAttribute(); var isBareRecordRef = IsCultRecordRefType(candidate.MemberType); if (referenceAttribute == null && !isBareRecordRef) continue; var many = referenceAttribute?.Many ?? false; var supported = many ? IsEnumerableOfRecordRef(candidate.MemberType) || TryGetDictionaryRefTypes(candidate.MemberType, out _, out _) : isBareRecordRef; if (!supported) rejections.Add(UnsupportedReferenceShapeMessage(type.Name, Qualified(candidate.Member))); } if (rejections.Count > 0) throw new InvalidOperationException(rejections[0]); return keyed .Select(entry => PersistedMember.FromMember( entry.Member, entry.MemberType, entry.Member is FieldInfo field ? field.GetValue : ((PropertyInfo)entry.Member).GetValue, entry.Slot)) .OrderBy(member => member.Slot) .ToArray(); } private static (Type, string) RootOf(PropertyInfo property) => ((property.GetMethod ?? property.SetMethod)!.GetBaseDefinition().DeclaringType!, property.Name); private static IEnumerable OverrideChain(PropertyInfo property) { var root = RootOf(property); for (var current = property.DeclaringType; current != null; current = current.BaseType) { var declaration = current.GetProperty(property.Name, BindingFlags.DeclaredOnly | BindingFlags.Instance | BindingFlags.Public | BindingFlags.NonPublic); if (declaration != null && declaration.GetIndexParameters().Length == 0 && RootOf(declaration) == root) yield return declaration; } } // MessagePack's DynamicObjectResolver picks the [SerializationConstructor] if one is marked, otherwise the longest public constructor // (any constructor under AllowPrivate) whose every parameter takes the keyed member at its position (slot == parameter index). // Returns that constructor's parameter count, 0 for a struct with none, or null when MessagePack finds no constructor. private static int? ConstructorArity(Type type, bool allowPrivate, IEnumerable<(Type MemberType, int Slot)> members) { var slots = new Dictionary(); foreach (var member in members) slots.TryAdd(member.Slot, member.MemberType); var constructors = type.GetConstructors(BindingFlags.Instance | BindingFlags.Public | BindingFlags.NonPublic); var marked = constructors.Where(constructor => constructor.IsDefined(typeof(SerializationConstructorAttribute))).ToArray(); var arities = (marked.Length > 0 ? marked : constructors.Where(constructor => allowPrivate || constructor.IsPublic)) .Select(constructor => constructor.GetParameters()) .Where(parameters => parameters.All(parameter => slots.TryGetValue(parameter.Position, out var memberType) && parameter.ParameterType.IsAssignableFrom(memberType))) .Select(parameters => (int?)parameters.Length) .ToArray(); return arities.Length > 0 ? arities.Max() : type.IsValueType && marked.Length == 0 ? 0 : null; } private static string Qualified(MemberInfo member) => member.DeclaringType!.Name.Split('`')[0] + "." + member.Name; // 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) return null; return string.IsNullOrWhiteSpace(indexAttribute.Alias) ? member.Name : indexAttribute.Alias; } private static bool IsCultRecordRefType(Type type) => type.IsGenericType && type.GetGenericTypeDefinition() == typeof(CultRecordRef<>); private static bool IsEnumerableOfRecordRef(Type type) { foreach (var candidate in type.GetInterfaces().Append(type)) { if (candidate.IsGenericType && candidate.GetGenericTypeDefinition() == typeof(IEnumerable<>) && IsCultRecordRefType(candidate.GetGenericArguments()[0])) return true; } return false; } private static bool TryGetDictionaryRefTypes(Type type, out PropertyInfo keyProperty, out PropertyInfo valueProperty) { foreach (var candidate in type.GetInterfaces().Append(type)) { if (!candidate.IsGenericType || candidate.GetGenericTypeDefinition() != typeof(IDictionary<,>)) continue; var arguments = candidate.GetGenericArguments(); if (!IsCultRecordRefType(arguments[0])) continue; var pairType = typeof(KeyValuePair<,>).MakeGenericType(arguments[0], arguments[1]); keyProperty = pairType.GetProperty("Key")!; valueProperty = pairType.GetProperty("Value")!; return true; } keyProperty = null!; valueProperty = null!; return false; } private static readonly HashSet NumericClrTypes = new() { typeof(sbyte), typeof(byte), typeof(short), typeof(ushort), typeof(int), typeof(uint), typeof(long), typeof(ulong), typeof(float), typeof(double), typeof(decimal) }; // D8's closed CLR numeric set: the sole authority for whether a comparison predicate may reach // a member. The TypeName string persisted in the catalog is generated from this set (via // CultSchemaTypeNames.FromType) and must never be hand-checked against separately - S22 pins it. private static bool IsNumericClrType(Type type) { var underlying = Nullable.GetUnderlyingType(type) ?? type; return NumericClrTypes.Contains(underlying); } // Q-J (docs/cultnet-selection-cut.md, section 2 "Numbers"): a row's exact canonical decimal, // never a double - a long past 2^53 must not lose precision passing through this path. Boxing a // Nullable with a value produces a boxed T (never a boxed Nullable), so raw is always one // of the closed CLR numeric set's non-nullable types here. NaN and infinity render as null: // "a NaN or infinite member matches no comparison." private static string? RenderCanonicalNumber(object raw) => raw switch { float f => float.IsNaN(f) || float.IsInfinity(f) ? null : RenderExactFloat(f), double d => double.IsNaN(d) || double.IsInfinity(d) ? null : RenderExactDouble(d), decimal m => CanonicalizeDecimalDigits(m.ToString(CultureInfo.InvariantCulture)), _ => CanonicalizeDecimalDigits(Convert.ToString(raw, CultureInfo.InvariantCulture)!) }; // R-D (docs/cultnet-selection-cut.md): the wire's canonical decimal is the value's exact decimal // expansion, not a shortest round-trip form. .NET's default float/double ToString has used the // shortest round-trip form since .NET Core 3.0, which rounds a tie between two decimals to // whichever is "shorter" - Rust's shortest form breaks the same tie the other way at some // values, so the two runtimes rendered different strings for the same bit pattern (394/200k f32, // 48/200k f64 in the parity vectors). Every finite float and double is an exact dyadic rational // (significand * 2^exponent for an integer significand and exponent), so it has a finite exact // decimal expansion computed with BigInteger: multiply the significand by 5^-exponent when the // exponent is negative (1/2^k == 5^k/10^k) to get the exact digits, and place the decimal point. // No CLR floating-point arithmetic and no string round-tripping is on this path. private static string RenderExactFloat(float value) { var bits = BitConverter.SingleToInt32Bits(value); var negative = bits < 0; var biasedExponent = (bits >> 23) & 0xFF; var mantissaBits = bits & 0x7FFFFF; if (biasedExponent == 0 && mantissaBits == 0) return "0"; // canonical zero: a single "0", never "-0" (Q-J). BigInteger significand; int exponent; if (biasedExponent == 0) { significand = mantissaBits; exponent = -126 - 23; // subnormal: no implicit leading 1, minimum exponent. } else { significand = mantissaBits | (1 << 23); exponent = biasedExponent - 127 - 23; } return RenderExactDecimal(negative, significand, exponent); } private static string RenderExactDouble(double value) { var bits = BitConverter.DoubleToInt64Bits(value); var negative = bits < 0; var biasedExponent = (int)((bits >> 52) & 0x7FF); var mantissaBits = bits & 0xFFFFFFFFFFFFFL; if (biasedExponent == 0 && mantissaBits == 0) return "0"; // canonical zero: a single "0", never "-0" (Q-J). BigInteger significand; int exponent; if (biasedExponent == 0) { significand = mantissaBits; exponent = -1022 - 52; // subnormal: no implicit leading 1, minimum exponent. } else { significand = mantissaBits | (1L << 52); exponent = biasedExponent - 1023 - 52; } return RenderExactDecimal(negative, significand, exponent); } // value = significand * 2^exponent, significand > 0. exponent >= 0 is an exact integer shift; // exponent < 0 rewrites 2^exponent as 5^-exponent / 10^-exponent, so the digits are exact and // the decimal point lands -exponent places from the right - no precision limit, no rounding. private static string RenderExactDecimal(bool negative, BigInteger significand, int exponent) { string digits; int fractionDigits; if (exponent >= 0) { digits = (significand << exponent).ToString(CultureInfo.InvariantCulture); fractionDigits = 0; } else { fractionDigits = -exponent; digits = (significand * BigInteger.Pow(5, fractionDigits)).ToString(CultureInfo.InvariantCulture); } string positional; if (fractionDigits == 0) { positional = digits; } else { if (digits.Length <= fractionDigits) digits = digits.PadLeft(fractionDigits + 1, '0'); var splitAt = digits.Length - fractionDigits; positional = digits.Substring(0, splitAt) + "." + digits.Substring(splitAt); } return CanonicalizeDecimalDigits(negative ? "-" + positional : positional); } // The one canonicalizer every rendered number passes through: strips leading integer zeros and // trailing fractional zeros, and collapses a negative value that reduces to zero into "0" - the // door's regex forbids all three spellings on the way in (Q-J), and this is why the row side // never produces one on the way out. private static string CanonicalizeDecimalDigits(string signedPositional) { var negative = signedPositional.Length > 0 && signedPositional[0] == '-'; var unsigned = negative ? signedPositional.Substring(1) : signedPositional; var dot = unsigned.IndexOf('.'); var integerPart = dot < 0 ? unsigned : unsigned.Substring(0, dot); var fractionPart = dot < 0 ? string.Empty : unsigned.Substring(dot + 1); integerPart = integerPart.TrimStart('0'); if (integerPart.Length == 0) integerPart = "0"; fractionPart = fractionPart.TrimEnd('0'); var isZero = integerPart == "0" && fractionPart.Length == 0; var result = fractionPart.Length == 0 ? integerPart : integerPart + "." + fractionPart; return negative && !isZero ? "-" + result : result; } private static string DuplicateIndexAliasMessage(string documentTypeName, string first, string second, string alias) => $"Cult document {documentTypeName} members {first} and {second} both declare index alias \"{alias}\"; an index alias must name one member."; // D10b (R-AL): a plain member's bare name colliding with a reference member's index alias. // CultDocumentRegistry.ReferencesOf and CultNetSelectionEvaluator.ReferenceMembers both key // reference lookups off (IndexAlias ?? MemberName); D10 above only groups members that // declare an alias, so an unaliased member's bare name sharing that string with a reference // member's alias registered silently and faulted every hop through it at resolution time. // // Breaking registration change, noted per the fix batch 7 ruling: a shape that used to // register and work - a data member aliased "Owner" beside a reference member named // "Owner" - is now refused. There is no compatibility path; the collision was always a // silent resolution fault waiting to happen, so the fix is a door, not a warning. private static string DuplicateReferenceRoleMessage(string documentTypeName, string first, string second, string role) => $"Cult document {documentTypeName} members {first} and {second} both resolve to reference role \"{role}\" (index alias, or bare member name when unaliased); a reference role must name one member. " + $"This is a breaking registration change from before D10b existed: a shape where a plain member's bare name collided with a reference member's index alias used to register silently and fault at resolution time instead."; private static string UnsupportedReferenceShapeMessage(string documentTypeName, string member) => $"Cult document {documentTypeName} member {member} declares a reference the cache cannot walk; many:false needs CultRecordRef, many:true needs IEnumerable> or IDictionary, V>."; private static string NotMessagePackObjectMessage(string documentTypeName) => $"Cult document {documentTypeName} is not a MessagePack object; add [MessagePackObject] so MessagePack serializes its [Key(n)] members."; private static string NotVisibleMessage(string documentTypeName) => $"Cult document {documentTypeName} is not public; MessagePack serializes a non-public type only with [MessagePackObject(AllowPrivate = true)], so make it public or set AllowPrivate."; private static string UnmarkedNonPublicMemberMessage(string documentTypeName, string memberName) => $"Cult document {documentTypeName} member {memberName} is non-public; [MessagePackObject(AllowPrivate = true)] makes MessagePack read it, so mark it [IgnoreMember]."; internal static string UnkeyedMemberMessage(string documentTypeName, string memberName) => $"Cult document {documentTypeName} member {memberName} has no [Key]; MessagePack requires [Key(n)] or [IgnoreMember] on every public member, so mark it one or the other."; private static string StringKeyMessage(string documentTypeName, string memberName) => $"Cult document {documentTypeName} member {memberName} has a string [Key]; string keys are not supported; use integer [Key(n)]."; private static string ReadonlyUnfilledMessage(string documentTypeName, string memberName, int slot) => $"Cult document {documentTypeName} member {memberName} is a readonly field that no constructor fills; MessagePack writes it but never reads it back, so give the constructor MessagePack calls a parameter at position {slot}, drop readonly, or add [MessagePackObject(AllowPrivate = true)]."; private static string GetOnlyUnfilledMessage(string documentTypeName, string memberName, int slot) => $"Cult document {documentTypeName} member {memberName} is a get-only property that no constructor fills; MessagePack writes it but never reads it back, so give the constructor MessagePack calls a parameter at position {slot}, add a setter, or mark it [IgnoreMember]."; private static string NonPublicSetterMessage(string documentTypeName, string memberName) => $"Cult document {documentTypeName} member {memberName} has a non-public set accessor; MessagePack writes it but never reads it back without AllowPrivate, so make the setter public or add [MessagePackObject(AllowPrivate = true)]."; private static string KeyedNonPublicMemberMessage(string documentTypeName, string memberName) => $"Cult document {documentTypeName} member {memberName} is non-public; MessagePack skips it silently, so [Key] on a non-public member requires [MessagePackObject(AllowPrivate = true)]."; private static string NoConstructorMessage(string documentTypeName) => $"Cult document {documentTypeName} has no constructor MessagePack can call; add a parameterless constructor (public unless AllowPrivate), or one whose parameters take the [Key(0)], [Key(1)], ... members in order."; private static string DuplicateSlotMessage(string documentTypeName, string first, string second, int slot) => $"Cult document {documentTypeName} members {first} and {second} share [Key({slot})]; every persisted member needs a distinct [Key(n)]."; private static string DivergentOverrideMessage(string documentTypeName, string member, string root) => $"Cult document {documentTypeName} member {member} overrides {root} with a different [Key] or [IgnoreMember]; MessagePack reads the base declaration's, so an override must repeat or omit them."; private static string HiddenMemberMessage(string documentTypeName, string first, string second) => $"Cult document {documentTypeName} member {first} hides persisted member {second}; persisted member names must be unique."; private static bool IsIgnored(MemberInfo member) => member.GetCustomAttribute() != null; private static object? GetKeyValue(MemberInfo member) { var key = member.GetCustomAttribute(); return key?.IntKey is int slot ? slot : key?.StringKey; } private static string Sha256(string input) { using var sha = SHA256.Create(); var bytes = sha.ComputeHash(Encoding.UTF8.GetBytes(input)); return "sha256:" + BitConverter.ToString(bytes).Replace("-", string.Empty).ToLowerInvariant(); } // internal, not private: CultDocumentDescriptor (a separate top-level class) reads RichMembers // through this type to build the public DeclaredMembers view and the three value accessors. internal sealed class PersistedMember { public MemberInfo Member { get; set; } = default!; public Type MemberType { get; set; } = default!; public int Slot { get; set; } public bool IsName { get; set; } public string? IndexAlias { get; set; } public bool IsReference { get; set; } public bool IsMany { get; set; } public string? TargetSchemaName { get; set; } // The reference's declared target Type, held alongside TargetSchemaName (which is null for // an abstract target - D9). CultDocumentRegistry.ResolveTargetLeaves needs the Type, not the // name, to compute the leaf set by assignability. public Type? TargetType { get; set; } public bool IsNumeric { get; set; } public Func Getter { get; set; } = default!; public Func GetterNullable { 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 // section 2 "Numbers". public Func NumberOf { get; set; } = default!; // D11: null for a non-reference member; otherwise yields (target key, payload) per declared // shape - one entry for a bare reference, one per element for a many-enumerable, one per key // with its value as payload for a many-dictionary. Registration already refused every other // shape (DiscoverMembers), so this is never called against something it cannot walk. public Func>? ReferenceEnumerator { get; set; } public static PersistedMember FromMember( MemberInfo member, Type memberType, Func getValue, int slot) { var referenceAttribute = member.GetCustomAttribute(); var targetType = ResolveReferenceTarget(memberType, referenceAttribute?.TargetType); var targetSchemaName = targetType?.GetCustomAttribute()?.SchemaName; var isReference = targetType != null || referenceAttribute != null; var many = referenceAttribute?.Many ?? false; return new PersistedMember { Member = member, MemberType = memberType, Slot = slot, IsName = member.GetCustomAttribute() != null, IndexAlias = ResolveIndexAlias(member), IsReference = isReference, IsMany = many, TargetSchemaName = targetSchemaName, TargetType = targetType, IsNumeric = IsNumericClrType(memberType), Getter = document => getValue(document)?.ToString() ?? string.Empty, GetterNullable = document => getValue(document)?.ToString(), NumberOf = document => { var raw = getValue(document); return raw == null ? null : RenderCanonicalNumber(raw); }, ReferenceEnumerator = isReference ? BuildReferenceEnumerator(memberType, many, getValue) : null }; } // Registration (DiscoverMembers' D11 check) already refused every shape but the three this // builds an enumerator for; this does not re-refuse, it only classifies which of the three. private static Func> BuildReferenceEnumerator( Type memberType, bool many, Func getValue) { if (!many) { return document => getValue(document) is ICultRecordRef reference ? new[] { (reference.Key, (object?)null) } : Array.Empty<(CultRecordKey, object?)>(); } if (TryGetDictionaryRefTypes(memberType, out var keyProperty, out var valueProperty)) { return document => { if (getValue(document) is not System.Collections.IEnumerable sequence) return Array.Empty<(CultRecordKey, object?)>(); var results = new List<(CultRecordKey, object?)>(); foreach (var entry in sequence) { if (keyProperty.GetValue(entry) is ICultRecordRef reference) results.Add((reference.Key, valueProperty.GetValue(entry))); } return results; }; } return document => { if (getValue(document) is not System.Collections.IEnumerable sequence) return Array.Empty<(CultRecordKey, object?)>(); var results = new List<(CultRecordKey, object?)>(); foreach (var entry in sequence) { if (entry is ICultRecordRef reference) results.Add((reference.Key, (object?)null)); } return results; }; } // Mirrors the isReference computation FromMember uses below, callable from DiscoverMembers // (D10b) before a slot is turned into a full PersistedMember. internal static bool IsReferenceCandidate(MemberInfo member, Type memberType) { var referenceAttribute = member.GetCustomAttribute(); var targetType = ResolveReferenceTarget(memberType, referenceAttribute?.TargetType); return targetType != null || referenceAttribute != null; } private static Type? ResolveReferenceTarget(Type memberType, Type? explicitTarget) { if (explicitTarget != null) { return explicitTarget; } if (memberType.IsGenericType && memberType.GetGenericTypeDefinition() == typeof(CultRecordRef<>)) { return memberType.GetGenericArguments()[0]; } // R-T(b): a many reference's implicit target is its CultRecordRef element's T - the // same narrowing a scalar CultRecordRef gets above - not left null just because the // member's own declared type is List<> or IDictionary<,>, not CultRecordRef<> itself. // Before this, an implicitly-typed many reference reached CultNetSelectionEvaluator's // EnsureWithinDeclaredTarget with a null TargetType and its out-of-target check was // silently skipped for every edge through it (R-T(b)/S18). if (IsEnumerableOfRecordRef(memberType)) { foreach (var candidate in memberType.GetInterfaces().Append(memberType)) { if (candidate.IsGenericType && candidate.GetGenericTypeDefinition() == typeof(IEnumerable<>) && IsCultRecordRefType(candidate.GetGenericArguments()[0])) { return candidate.GetGenericArguments()[0].GetGenericArguments()[0]; } } } if (TryGetDictionaryRefTypes(memberType, out var keyProperty, out _)) { return keyProperty.PropertyType.GetGenericArguments()[0]; } return null; } } } public enum CultCommitOutcome { Committed, Mismatch, Contended } public sealed class CultCommitRequest { internal CultCommitRequest( IReadOnlyList upserts, IReadOnlyList deletes, IReadOnlyList checks, IReadOnlyList<(CultRecordKey Key, string? Version)> expected, bool expectUnchanged) { Upserts = upserts; Deletes = deletes; Checks = checks; Expected = expected; ExpectUnchanged = expectUnchanged; } public IReadOnlyList Upserts { get; } public IReadOnlyList Deletes { get; } /// /// Keys the batch removes that its cache does not hold. Such a removal removes nothing, so it is never in /// ; but a store that holds a record at one of these keys under an id no registered type declares /// refuses the commit, as it refuses a write of that record. The home's judges /// these keys under its lock, with the keys it writes; every other attached store judges them first through /// , each under its own lock, and a refusal anywhere writes nothing. A judged /// key is never removed: a record a declared type reads stays where it is. /// public IReadOnlyList Checks { get; } public IReadOnlyList<(CultRecordKey Key, string? Version)> Expected { get; } public bool ExpectUnchanged { get; } public bool HasConditions => Expected.Count > 0 || ExpectUnchanged; /// /// Whether the batch's conditions hold. is the version of each record the store holds on disk now, /// by key; is what the store's cache last loaded or committed, and the /// foreign records the store carries as it last read them (; empty for a home /// with none). Under each foreign record compares against the version it had when the store /// read it, as a record the cache holds compares against its own. Identity is the record's version, the SHA-256 of its stored /// bytes, which names its schema id and storedAt too. /// public bool ConditionsHold(IReadOnlyDictionary durable, IEnumerable observed, IEnumerable foreign) { foreach (var (key, version) in Expected) { var holds = durable.TryGetValue(key.Value, out var stored) ? version != null && string.Equals(stored, version, StringComparison.Ordinal) : version == null; if (!holds) return false; } return !ExpectUnchanged || durable.Select(pair => (pair.Key, Version: pair.Value)) .OrderBy(pair => pair.Key, StringComparer.Ordinal) .SequenceEqual(observed .Select(stored => (Key: stored.Key.Value, Version: stored.StoredVersion)) .Concat(foreign.Select(record => (record.Key, Version: record.StoredVersion))) .OrderBy(pair => pair.Key, StringComparer.Ordinal)); } } // One staged write: a plain record (Stored), a variant (Variant), a flatten, or, with none of those, a removal. internal sealed class StagedOperation { public StagedOperation(CultRecordKey key, CultStoredDocument? stored, CultVariantDelta? variant, bool flatten) { Key = key; Stored = stored; Variant = variant; Flatten = flatten; } public CultRecordKey Key { get; } public CultStoredDocument? Stored { get; } public CultVariantDelta? Variant { get; } public bool Flatten { get; } public bool IsRemoval => Stored == null && Variant == null && !Flatten; } // An explicit value: nothing staged here is visible to anyone until Commit returns. public sealed class CultCacheBatch { private readonly CultCache _cache; internal readonly Dictionary Operations = new(StringComparer.Ordinal); internal readonly List<(CultRecordKey Key, string? Version)> Expected = new(); internal bool ExpectsUnchanged; internal bool Sealed; internal CultCacheBatch(CultCache cache) { _cache = cache; } public CultRecordHandle Upsert(T document, CultRecordHandle? handle = null) { if (document == null) throw new ArgumentNullException(nameof(document)); return new CultRecordHandle(Stage(document, handle?.Key)); } public CultRecordKey Upsert(Type type, object document, CultRecordKey? key = null) { CultCache.RequireInstanceOf(type, document); return Stage(document, key); } public void Remove(CultRecordKey key) { ThrowIfSealed(); Operations[key.Value] = new StagedOperation(key, null, null, false); } // Stages a variant record: its base and a sparse set of overrides, resolved by the cache when the batch commits. // A variant changes only through here; rebasing is this call with another base. public CultRecordKey UpsertVariant(CultRecordKey key, CultRecordKey baseKey, IEnumerable? overrides = null) { ThrowIfSealed(); Operations[key.Value] = new StagedOperation( key, null, new CultVariantDelta(baseKey.Value, (overrides ?? Array.Empty()).ToArray()), false); return key; } // Replaces a variant with a plain record holding its resolved document: the explicit, separate operation (Q1a). public void Flatten(CultRecordKey key) { ThrowIfSealed(); Operations[key.Value] = new StagedOperation(key, null, null, true); } public void Expect(CultRecordKey key, object? current) { ThrowIfSealed(); var observed = _cache.Observe(key, current); Expected.Add((key, observed?.StoredVersion)); } public void ExpectUnchanged() { ThrowIfSealed(); ExpectsUnchanged = true; } private CultRecordKey Stage(object document, CultRecordKey? key) { ThrowIfSealed(); var stored = _cache.CreateStoredDocument(document, key); Operations[stored.Key.Value] = new StagedOperation(stored.Key, stored, null, false); return stored.Key; } private void ThrowIfSealed() { if (Sealed) throw new InvalidOperationException("This batch has already been committed or abandoned."); } } public sealed class CultCache : IDisposable { private readonly CultDocumentRegistry _registry; 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(); private readonly Dictionary _globals = new(); private readonly CultCodec? _codec; // Keys holding a variant: derived from the entries by Apply, never decided anywhere else. private readonly HashSet _variantKeys = new(StringComparer.Ordinal); private readonly ConditionalWeakTable _handles = new(); // The cache's own observers, copy-on-write under _observerGate: publication reads one snapshot per change and // catches per observer, so no observer decides whether another receives a change. private Observer[] _observers = Array.Empty>(); private bool _observersCompleted; private readonly object _observerGate = new(); private readonly object _gate = new(); private long _sequence; // What this thread's outermost hold admitted, published by that hold when it exits. [ThreadStatic] private static Hold? _held; // Journals run under the gate, in registration order; both fields are read and written only under it. private readonly List>>> _journals = new(); private bool _journaling; private bool _dirtyInMemory; // The codec is the store's encoding (CultCacheMessagePack.Create supplies it). A cache without one holds no variants: // resolving a variant needs to clone documents and decode override values. public CultCache(CultDocumentRegistry? registry = null, CultCodec? codec = null) { _registry = registry ?? CultDocumentRegistry.Shared; _codec = codec; } public CultDocumentRegistry Registry => _registry; // A store wrote: every record it wrote is stored under its type's own id at the storedAt the write gave it, with the // version of the bytes it wrote, and a plain record carries the ids it held, so it is no longer in-memory only. A variant's ids are not cleared here: the // store wrote the delta it was handed, and only a write of the variant changes that. internal void Wrote(IEnumerable<(CultPersistedRecord Record, string Version)> written) => Held(() => { foreach (var (record, version) in written) { if (!_entries.TryGetValue(record.Key, out var entry)) continue; entry.StoredSchemaId = record.SchemaId; entry.StoredAt = record.StoredAt; entry.StoredVersion = version; if (entry.Variant == null) entry.IdsInMemoryOnly = false; } return true; }); public CultCodec? Codec => _codec; // The whole override of one member, encoded with this cache's codec, ready for UpsertVariant. public CultVariantOverride Override(Type documentType, string member, object? value) { if (_codec == null) throw new InvalidOperationException("This cache has no codec; create it with CultCacheMessagePack.Create."); var found = _registry.GetRequired(documentType).RichMembers.FirstOrDefault(entry => entry.Member.Name == member) ?? throw new ArgumentException($"{documentType.Name} has no persisted member '{member}'.", nameof(member)); return CultVariantOverride.Set( found.Slot, _codec.Serialize(value!, found.MemberType, documentType)); } public CultVariantOverride Override(string member, object? value) where T : class => Override(typeof(T), member, value); // The one-shot rewrite for stores written before element ids existed: upserts every record whose ids exist only in // memory (minted when it loaded), through the normal write path, one commit per home store. Returns how many records // it rewrote; a second call finds none. A selected record stored under an id no registered type declares is read-only, // so its home store's commit is refused with CultSchemaConflictException; commits to earlier stores stand. public int MintElementIds() { var groups = Held(() => _entries.Values .Where(entry => entry.IdsInMemoryOnly) .GroupBy(entry => Home(entry.Descriptor.DocumentType)) .Select(group => group.ToArray()) .ToArray()); foreach (var group in groups) { Commit(batch => { foreach (var entry in group) { if (entry.Variant is { } delta) batch.UpsertVariant(entry.Key, new CultRecordKey(delta.BaseKey), delta.Overrides); else batch.Upsert(entry.Descriptor.DocumentType, entry.Document, entry.Key); } }); } return groups.Sum(group => group.Length); } public bool IsDirty { get { lock (_gate) return _stores.Count == 0 ? _dirtyInMemory : _stores.Any(entry => entry.Store.IsDirty); } } public bool FlushAttachedStoresOnDispose { get; set; } public event Action? OnUpdate; public IReadOnlyList BackingStores { get { lock (_gate) return _stores.Select(entry => entry.Store).ToArray(); } } public IEnumerable AllEntries { get { lock (_gate) return _entries.Values.Select(entry => entry.Document).ToArray(); } } public IEnumerable AllStoredDocuments { get { lock (_gate) return _entries.Values .OrderBy(entry => entry.Descriptor.SchemaName, StringComparer.Ordinal) .ThenBy(entry => entry.Key.Value, StringComparer.Ordinal) .ToArray(); } } public Observable> Watch() where T : class { return Observable.Create(Register) .Where(change => typeof(T).IsAssignableFrom(change.Stored.Descriptor.DocumentType)) .Select(change => new CultCacheDocumentChange( change.Kind, change.Stored.Key, change.Document as T, change.Previous as T, change.Sequence)); } // Disposing removes the journal. A dispose the cache refuses (from inside a journal, or inside another cache's hold) // throws and leaves the registration live and disposable; it counts as disposed only once the removal succeeded. private sealed class JournalRegistration : IDisposable { private readonly CultCache _cache; private readonly Action>> _journal; private bool _removed; public JournalRegistration(CultCache cache, Action>> journal) { _cache = cache; _journal = journal; } public void Dispose() { if (_removed) return; _cache.Held(() => _cache._journals.Remove(_journal)); _removed = true; } } private IDisposable Register(Observer observer) { lock (_observerGate) { if (!_observersCompleted) { _observers = _observers.Append(observer).ToArray(); return Disposable.Create(() => { lock (_observerGate) _observers = _observers.Where(held => !ReferenceEquals(held, observer)).ToArray(); }); } } observer.OnCompleted(); return Disposable.Empty; } public Observable> WatchRecord(CultRecordKey key) where T : class { return Watch().Where(change => change.Key.Equals(key)); } // An ordered admission journal: called under the gate, right after an admission is applied, once per admission // that changed something, with that admission's changes in Sequence order. Journals of one cache run in // registration order. A journal sees the new state (reads work) and is the one place cache order is a fact: it // must not write to the cache or wait for anything that needs it, and any hold on this cache while a journal runs // throws. A journal's exception does not undo the admission: it is published, then the exception is rethrown to // the writer, as an observer's is. public IDisposable AddJournal(Action>> journal) { if (journal == null) throw new ArgumentNullException(nameof(journal)); Held(() => { _journals.Add(journal); return true; }); return new JournalRegistration(this, journal); } // Attaching reads the store; there is no interval in which it is attached but unread. public void AddBackingStore(CacheBackingStore store, params Type[] homes) { if (store == null) throw new ArgumentNullException(nameof(store)); homes ??= Array.Empty(); Held(() => { if (store.Loaded != null) throw new InvalidOperationException($"Backing store {store} is already attached to a cache."); if (store.IsDirty) throw new InvalidOperationException($"Backing store {store} has staged writes; attach it clean."); if (homes.Length == 0 && _stores.Any(entry => entry.Homes.Length == 0)) throw new InvalidOperationException( $"Backing store {store} would be a second untyped store; name the types it is home to."); foreach (var home in homes) { var owner = _stores.FirstOrDefault(entry => entry.Homes.Contains(home)).Store; if (owner != null) throw new InvalidOperationException($"{home.FullName} is already routed to {owner}; it cannot also route to {store}."); } var candidate = (store, homes); foreach (var type in _entries.Values.Select(entry => entry.Descriptor).Distinct()) { var before = Home(type.DocumentType); _stores.Add(candidate); var after = Home(type.DocumentType); _stores.RemoveAt(_stores.Count - 1); if (before != after) throw new InvalidOperationException( $"Attaching {store} would move {type.SchemaName} from {before?.ToString() ?? "memory"} to {after}; " + "attach routed stores before the untyped store."); } store.AttachRegistry(_registry); store.Cache = this; store.Loaded = (loaded, dropped) => Admit(loaded, dropped, store, (_, _) => CultCommitOutcome.Committed); _stores.Add(candidate); try { store.PullAll(); } catch { _stores.Remove(candidate); store.Loaded = null; store.Cache = null; throw; } return true; }); } public Task PullAllBackingStoresAsync() { // Our stores take the gate themselves; a third-party store might not, so the cache takes it here. // One store's throwing OnUpdate handler does not stop the others from loading. var failures = new List(); foreach (var store in BackingStores) { try { Held(() => { store.PullAll(); return true; }); } catch (Exception exception) { failures.Add(exception); } } if (failures.Count == 1) ExceptionDispatchInfo.Capture(failures[0]).Throw(); if (failures.Count > 1) throw new AggregateException(failures); return Task.CompletedTask; } public void FlushAllBackingStores() => Held(() => { foreach (var (store, _) in _stores) { if (!store.IsReadOnly && (store.IsDirty || store.HasStagedChecks)) store.PushAll(); } return true; }); public Task FlushAsync() { FlushAllBackingStores(); return Task.CompletedTask; } public Task> AddAsync(T document, CultRecordHandle? handle = null) { return UpsertAsync(document, handle); } // Refused with CultSchemaConflictException when the key holds a record stored under an id no registered type declares. public Task> UpsertAsync(T document, CultRecordHandle? handle = null) { if (document == null) throw new ArgumentNullException(nameof(document)); return Task.FromResult(new CultRecordHandle(Write(document, handle?.Key))); } // Refused with CultSchemaConflictException when the key holds a record stored under an id no registered type declares. public Task UpsertAsync(Type documentType, object document, CultRecordKey? key = null) { RequireInstanceOf(documentType, document); return Task.FromResult(Write(document, key)); } // A single-call variant write; a batch stages the same thing with CultCacheBatch.UpsertVariant. // Refused with CultSchemaConflictException when the key holds a record stored under an id no registered type declares. public Task UpsertVariantAsync(CultRecordKey key, CultRecordKey baseKey, IEnumerable? overrides = null) { var delta = new CultVariantDelta(baseKey.Value, (overrides ?? Array.Empty()).ToArray()); return Task.FromResult(Held(() => { var stored = StampVariant(key, DescriptorOfBase(key, delta.BaseKey, null), delta); Admit(new[] { stored }, Array.Empty(), null, (home, docs) => { home?.Push(docs[0]); return CultCommitOutcome.Committed; }); return key; })); } // Replaces a variant with a plain record holding its resolved document (Q1a's explicit operation). // Refused with CultSchemaConflictException when the key holds a record stored under an id no registered type declares. public Task FlattenAsync(CultRecordKey key) => Task.FromResult(Held(() => { var stored = StampFlatten(key); Admit(new[] { stored }, Array.Empty(), null, (home, docs) => { home?.Push(docs[0]); return CultCommitOutcome.Committed; }); return key; })); /// /// Removes the record at ; false when the cache holds nothing there. /// /// /// A record stored under an id no registered type declares, read-only or foreign, is never removed: removing one the cache /// holds is refused with now. A key the cache does not hold costs nothing here and /// returns false; each writable store judges it at its next flush or commit, under its lock, and refuses that flush or /// commit if it holds a record there under such an id (a read-only store answers now, from what it last loaded). /// public bool Remove(CultRecordKey key) => Held(() => { if (!_entries.TryGetValue(key.Value, out var existing)) { RefuseRemovalOfUndeclared(new[] { key.Value }); return false; } Admit(Array.Empty(), new[] { existing }, null, (home, _) => { home?.Delete(existing); return CultCommitOutcome.Committed; }); return true; }); public void Remove(CultRecordHandle handle) { Remove(handle.Key); } /// /// Commits the staged batch. Returns false when a condition failed: nothing was written, changed in memory, or published. /// /// /// A commit that succeeds and then fails in a journal or observer still stands and is published, and the writer /// receives the failure. One failure arrives as itself, several as an . A writer that /// wants a particular failure (a CultNetShardLogException, when a database's log refused the commit) catches that type, /// or unwraps and looks for it. /// public bool Commit(Action stage) { return Land(stage, wait: true) == CultCommitOutcome.Committed; } public CultCommitOutcome TryCommit(Action stage) { return Land(stage, wait: false); } public CultRecordHandle? TryGetHandle(T document) { if (document == null) throw new ArgumentNullException(nameof(document)); lock (_gate) return _handles.TryGetValue(document, out var box) ? new CultRecordHandle(box.Key) : null; } // The stored form of a record: for a variant, the delta beside the resolved document. public CultStoredDocument? GetStored(CultRecordKey key) { lock (_gate) return _entries.TryGetValue(key.Value, out var stored) ? stored : null; } public object? Get(CultRecordKey key) { lock (_gate) return _entries.TryGetValue(key.Value, out var stored) ? stored.Document : null; } public T? Get(CultRecordKey key) where T : class { return Get(key) as T; } // Every change whose Sequence is at or below the returned one is reflected in the returned document. public (object? Document, long Sequence) GetWithSequence(CultRecordKey key) { lock (_gate) return (_entries.TryGetValue(key.Value, out var stored) ? stored.Document : null, _sequence); } public bool TryGet(CultRecordKey key, out T? document) where T : class { document = Get(key); return document != null; } public IEnumerable GetAll() where T : class { lock (_gate) return _entries.Values.Select(entry => entry.Document).OfType().ToArray(); } public T? GetGlobal() where T : class { return Single("global", () => _globals .Where(pair => typeof(T).IsAssignableFrom(pair.Key)) .Select(pair => pair.Value)); } public T? GetByName(string name) where T : class { return Single($"name '{name}'", () => _names .Where(pair => typeof(T).IsAssignableFrom(pair.Key)) .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. 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())); } public void Dispose() { Held(() => { if (FlushAttachedStoresOnDispose && (IsDirty || _stores.Any(entry => entry.Store.HasStagedChecks))) FlushAllBackingStores(); return true; }); foreach (var store in BackingStores) store.Dispose(); Observer[] completing; lock (_observerGate) { completing = _observers; _observers = Array.Empty>(); _observersCompleted = true; } var failures = new List(); foreach (var observer in completing) { try { observer.OnCompleted(); } catch (Exception exception) { failures.Add(exception); } } if (failures.Count == 1) ExceptionDispatchInfo.Capture(failures[0]).Throw(); if (failures.Count > 1) throw new AggregateException(failures); } internal static void RequireInstanceOf(Type documentType, object document) { if (documentType == null) throw new ArgumentNullException(nameof(documentType)); if (document == null) throw new ArgumentNullException(nameof(document)); if (!documentType.IsInstanceOfType(document)) throw new ArgumentException($"Document instance must be assignable to {documentType.FullName}.", nameof(document)); } // The runtime type decides the schema; a minted storedAt is always later than the record it replaces. internal CultStoredDocument CreateStoredDocument(object document, CultRecordKey? key) { var descriptor = _registry.GetRequired(document.GetType()); lock (_gate) return Stamp( key ?? (_handles.TryGetValue(document, out var box) ? box.Key : descriptor.IsGlobal ? new CultRecordKey($"global:{descriptor.SchemaId}") : new CultRecordKey(Guid.NewGuid().ToString("N"))), descriptor, document); } // Call under the gate that admits the result, so no load can land a later storedAt in between. private CultStoredDocument Stamp(CultRecordKey key, CultDocumentDescriptor descriptor, object document) => new(key, MintStoredAt(PreviousStoredAt(key), Clock()), descriptor, document); // An unresolved variant: the admitting hold resolves it against its base before anything lands. private CultStoredDocument StampVariant(CultRecordKey key, CultDocumentDescriptor descriptor, CultVariantDelta delta) => new(key, MintStoredAt(PreviousStoredAt(key), Clock()), descriptor, delta, null, _codec); // The plain record a Flatten stages: the variant's resolved document, under the variant's own key and schema. private CultStoredDocument StampFlatten(CultRecordKey key) { if (!_entries.TryGetValue(key.Value, out var existing) || existing.Variant == null) throw new InvalidOperationException($"Flatten({key.Value}): the record is not a variant."); return new CultStoredDocument(key, MintStoredAt(existing.StoredAt, Clock()), existing.Descriptor, existing.Document) { Flattens = true }; } private string? PreviousStoredAt(CultRecordKey key) => _entries.TryGetValue(key.Value, out var previous) ? previous.StoredAt : null; // The concrete type a variant takes from its base, walking bases through what the batch stages and then the cache. // Refuses a cycle or a base that is not there, naming the keys, exactly as the admitting hold would. private CultDocumentDescriptor DescriptorOfBase(CultRecordKey variant, string baseKey, CultCacheBatch? batch) { var seen = new List { variant.Value }; var next = baseKey; while (true) { if (seen.Contains(next)) throw CycleRefusal(seen.Append(next)); var staged = batch != null && batch.Operations.TryGetValue(next, out var op) ? op : null; if (staged != null) { if (staged.Stored != null) return staged.Stored.Descriptor; if (staged.Variant != null) { seen.Add(next); next = staged.Variant.BaseKey; continue; } if (!staged.Flatten) throw MissingBaseRefusal(variant.Value, baseKey, removed: true); } if (!_entries.TryGetValue(next, out var existing)) throw MissingBaseRefusal(variant.Value, baseKey, removed: false); return existing.Descriptor; } } internal CultStoredDocument? Observe(CultRecordKey key, object? current) { if (current == null) return null; lock (_gate) { if (_handles.TryGetValue(current, out var box) && box.Key.Equals(key) && _entries.TryGetValue(key.Value, out var stored) && ReferenceEquals(stored.Document, current)) return stored; } throw new ArgumentException($"Expect({key.Value}) was given an instance this cache does not hold at that key.", nameof(current)); } // The wall clock behind every storedAt. Internal: a test that writes a golden store pins it, so the bytes reproduce. internal Func Clock { get; set; } = () => DateTimeOffset.UtcNow; internal static string MintStoredAt(string? previous, DateTimeOffset now) { if (previous != null && DateTimeOffset.TryParse(previous, CultureInfo.InvariantCulture, DateTimeStyles.RoundtripKind, out var last) && now <= last) now = last.ToUniversalTime().AddTicks(1); return now.ToString("O", CultureInfo.InvariantCulture); } private CultRecordKey Write(object document, CultRecordKey? key) => Held(() => { var stored = CreateStoredDocument(document, key); Admit(new[] { stored }, Array.Empty(), null, (home, docs) => { home?.Push(docs[0]); return CultCommitOutcome.Committed; }); return stored.Key; }); private CultCommitOutcome Land(Action stage, bool wait) { if (stage == null) throw new ArgumentNullException(nameof(stage)); var batch = new CultCacheBatch(this); try { stage(batch); } finally { batch.Sealed = true; } // The in-process gate is always entered; only the store lock attempt honors wait, so Contended means another writer holds the store. return Held(() => { // Staging stamped outside this gate; a pull since then may hold a later storedAt, so stamp again. var upserts = batch.Operations.Values .Where(op => !op.IsRemoval) .Select(op => op.Stored != null ? Stamp(op.Key, op.Stored.Descriptor, op.Stored.Document) : op.Variant != null ? StampVariant(op.Key, DescriptorOfBase(op.Key, op.Variant.BaseKey, batch), op.Variant) : StampFlatten(op.Key)) .ToArray(); // A removal of a key the cache does not hold removes nothing: it is a check every store the commit reaches judges under its lock. var checks = batch.Operations.Values .Where(op => op.IsRemoval && !_entries.ContainsKey(op.Key.Value)) .Select(op => op.Key.Value) .ToArray(); var deletes = batch.Operations.Values .Where(op => op.IsRemoval && _entries.ContainsKey(op.Key.Value)) .Select(op => _entries[op.Key.Value]) .ToArray(); return Admit(upserts, deletes, null, (home, resolved) => { // A batch that only checks keys names no home; it commits through the cache's one writable store, if it has one. if (home == null && checks.Length > 0 && _stores.Where(entry => !entry.Store.IsReadOnly).Take(2).ToArray() is { Length: 1 } only) home = only[0].Store; // Every store the batch does not commit through judges its checks first, each under its own lock, so a refusal or // contention writes nothing; the home judges its own in CommitBatch. foreach (var (store, _) in _stores) { if (store == home || checks.Length == 0) continue; var judged = store.JudgeChecks(checks, wait); if (judged != CultCommitOutcome.Committed) return judged; } // What the store holds is what the hold resolved: a variant's delta beside its resolved view. var request = new CultCommitRequest(resolved, deletes, checks, batch.Expected.ToArray(), batch.ExpectsUnchanged); if (home != null) { if (request.HasConditions && home.IsDirty) throw new InvalidOperationException($"Backing store {home} has staged writes; flush before a conditional commit."); return home.CommitBatch(request, wait); } if (request.HasConditions && _stores.Count > 0) throw new InvalidOperationException("A conditional batch names its home store through a record it upserts or removes."); var inMemory = _entries.Values.ToDictionary(entry => entry.Key.Value, entry => entry.StoredVersion, StringComparer.Ordinal); return request.ConditionsHold(inMemory, _entries.Values, Array.Empty()) ? CultCommitOutcome.Committed : CultCommitOutcome.Mismatch; }); }); } // A removal of a key the cache does not hold removes nothing, so no write carries it to a store; but a store that holds a // record there under an id no registered type declares refuses it, as it refuses every write of that record, rather than let // the caller believe it was removed. Each store judges it its own way (CacheBackingStore.RefuseRemovalOfUndeclared). private void RefuseRemovalOfUndeclared(IReadOnlyCollection keys) { if (keys.Count == 0) return; foreach (var (store, _) in _stores) store.RefuseRemovalOfUndeclared(keys); } // Every add and remove passes here: single writes, committed batches, and loads (source set). // Nothing lands unless all of it is admissible; land is the durable step between validation and memory. private CultCommitOutcome Admit( IReadOnlyList admitted, IReadOnlyList evicted, CacheBackingStore? source, Func, CultCommitOutcome> land) => Held(() => { if (_held == null) throw new InvalidOperationException("An admission reached the cache under a plain lock on its gate; every admission runs in a hold."); var home = Validate(admitted, evicted, source); // Ids are decided before anything lands, and the caller's documents change only inside this admission: a plan that // refuses (a duplicate, a malformed id) has changed nothing, and a write that does not commit gives its ids back. // A write mints random ids; a load of a record written before ids existed mints the ids its reload will mint // again, and lands in memory only (variants mint inside Resolve, into their override values). var idPlans = admitted .Where(stored => stored.Variant == null) .Select(stored => (Stored: stored, Plan: CultElementIds.Plan(stored.Descriptor.DocumentType, stored.Document, stored.Key.Value, deterministic: source != null))) .ToArray(); VariantPlan plan; CultCommitOutcome outcome; try { foreach (var (_, idPlan) in idPlans) idPlan.Apply(); // The one resolution: a variant admitted, a base admitted or evicted, all resolved here against the post-batch // set, before land and before memory. land receives the admitted records with variants resolved. plan = Resolve(admitted, evicted, source); outcome = land(home, plan.Admitted); } catch { foreach (var (_, idPlan) in idPlans) idPlan.Undo(); throw; } if (outcome != CultCommitOutcome.Committed) { foreach (var (_, idPlan) in idPlans) idPlan.Undo(); return outcome; } // A record whose ids exist only in memory stays marked until a write persists them (a flush of the whole store // does, and so does any write of the record: a new entry is unmarked). if (source != null) { foreach (var (stored, idPlan) in idPlans) stored.IdsInMemoryOnly = idPlan.Count > 0; } var applied = Apply(plan, evicted, source); foreach (var change in applied) _held.Changes.Add((change, source != null)); RunJournals(applied, _held.Failures); return CultCommitOutcome.Committed; }); // Every admission runs in a hold. A nested hold on the same cache adds to the outermost one, which publishes // exactly its own changes after it leaves the gate, before it returns, on its caller's thread: observers never // run under the gate, and a write an observer makes is a new outermost hold. Cross-thread delivery order is not // guaranteed; each change carries the Sequence it was admitted with. Journals run before that, under the gate, // in Sequence order (AddJournal); no hold may be entered on this cache while one runs. Every change reaches every observer even // if an observer or an OnUpdate handler throws; then the exception (an AggregateException for several, whatever their // origin: a journal, an observer, a second database) is rethrown. A writer looking for one type unwraps InnerExceptions. // If the body threw, its exception wins. internal T Held(Func body) { if (Monitor.IsEntered(_gate)) { if (_journaling) throw new InvalidOperationException("A journal entered the cache it journals; a journal must not write to it."); return body(); } if (_held != null) throw new InvalidOperationException( "A cache hold was entered inside another cache's hold; a thread holds one cache's gate at a time."); var mine = _held = new Hold(); var result = default(T)!; try { lock (_gate) result = body(); } catch (Exception exception) { _held = null; Publish(mine); if (mine.Failures.Count == 0) throw; throw new AggregateException(mine.Failures.Prepend(exception)); } _held = null; var failures = mine.Failures; failures.AddRange(Publish(mine)); if (failures.Count == 1) ExceptionDispatchInfo.Capture(failures[0]).Throw(); if (failures.Count > 1) throw new AggregateException(failures); return result; } private List Publish(Hold hold) { var failures = new List(); foreach (var (change, loaded) in hold.Changes) { // R3 routes a throwing subscriber to its unhandled-exception handler, and a fail-fast handler rethrows: every // observer still runs, the exception joins the failures, and the observer stays subscribed. foreach (var observer in Volatile.Read(ref _observers)) { try { observer.OnNext(change); } catch (Exception exception) { failures.Add(exception); } } if (!loaded) continue; try { OnUpdate?.Invoke(change.Previous, change.Document); } catch (Exception exception) { failures.Add(exception); } } return failures; } // Runs under the gate. A journal that throws joins the hold's failures and the remaining journals still run. private void RunJournals(List applied, List failures) { if (_journals.Count == 0 || applied.Count == 0) return; var changes = applied .Select(change => new CultCacheDocumentChange( change.Kind, change.Stored.Key, change.Document, change.Previous, change.Sequence)) .ToArray(); _journaling = true; try { foreach (var journal in _journals) { try { journal(changes); } catch (Exception exception) { failures.Add(exception); } } } finally { _journaling = false; } } private CacheBackingStore? Validate( IReadOnlyList admitted, IReadOnlyList evicted, CacheBackingStore? source) { var homes = new List(); var globals = new Dictionary(_globals); foreach (var stored in evicted) { var type = stored.Descriptor.DocumentType; if (globals.TryGetValue(type, out var key) && key == stored.Key.Value) globals.Remove(type); if (source == null) Claim(stored); } foreach (var stored in admitted) { var descriptor = stored.Descriptor; if (source == null) Claim(stored); else if (Home(descriptor.DocumentType) is var home && home != source) throw new InvalidOperationException( $"{descriptor.SchemaName} record {stored.Key.Value} was loaded from {source} but its home is {home?.ToString() ?? "no store"}."); // One key, one store: a key never moves between homes, and two stores never both deliver it. if (_entries.TryGetValue(stored.Key.Value, out var present) && Home(present.Descriptor.DocumentType) is var held && Home(descriptor.DocumentType) is var incoming && held != incoming) throw new InvalidOperationException( $"Record {stored.Key.Value} is held by {held?.ToString() ?? "memory"}; " + $"{(source == null ? "writing" : "loading")} it as {descriptor.SchemaName} from {incoming?.ToString() ?? "memory"} would give the key a second store."); if (!descriptor.IsGlobal) continue; if (globals.TryGetValue(descriptor.DocumentType, out var existing) && existing != stored.Key.Value) throw new InvalidOperationException( $"{descriptor.SchemaName} is a global and already has record {existing}; {stored.Key.Value} would be a second."); globals[descriptor.DocumentType] = stored.Key.Value; } var distinct = homes.Distinct().ToArray(); if (distinct.Length > 1) throw new InvalidOperationException( $"The batch spans {distinct[0]} and {distinct[1]}; a commit lands in one home store."); return distinct.FirstOrDefault(); void Claim(CultStoredDocument stored) { var home = Home(stored.Descriptor.DocumentType); if (home == null) { if (_stores.Count > 0) throw new InvalidOperationException($"No backing store is home to {stored.Descriptor.SchemaName}."); return; } if (home.IsReadOnly) throw new InvalidOperationException( $"{stored.Descriptor.SchemaName} record {stored.Key.Value} routes to read-only backing store {home}."); home.RefuseUndeclared(stored.Key.Value); homes.Add(home); } } private sealed class VariantPlan { public VariantPlan(IReadOnlyList admitted) { Admitted = admitted; } // The admitted records, variants resolved. public IReadOnlyList Admitted { get; } // Variants this admission did not write whose base chain it changed: re-resolved, same delta, same storedAt. public List Dependents { get; } = new(); // Name and index values of every record that lands (Admitted and Dependents), by key. Filled once by Project. public Dictionary Projections { get; } = new(StringComparer.Ordinal); } // What a landing record's [CultName] and [CultIndex] getters said. Indexes holds only non-blank values. private sealed class Projection { public Projection(string? name, (string Alias, string Value)[] indexes) { Name = name; Indexes = indexes; } public string? Name { get; } public (string Alias, string Value)[] Indexes { get; } } // The only place an admission runs a name or index getter on a landing record. It runs before land and before any // sequence is minted, so a getter that throws refuses the whole admission with nothing written and nothing minted. private static void Project(VariantPlan plan) { foreach (var stored in plan.Admitted.Concat(plan.Dependents)) { var descriptor = stored.Descriptor; var name = descriptor.NameAccessor == null ? null : NameRead(stored); var indexes = new List<(string Alias, string Value)>(); foreach (var pair in descriptor.IndexAccessors) { var member = descriptor.RichMembers.First(candidate => candidate.IndexAlias == pair.Key).Member.Name; var value = Read(stored, $"index '{pair.Key}' member {member}", pair.Value); if (!string.IsNullOrWhiteSpace(value)) indexes.Add((pair.Key, value!)); } plan.Projections[stored.Key.Value] = new Projection(name, indexes.ToArray()); } } // A blank name is no name. private static string? NameRead(CultStoredDocument stored) => Read(stored, $"[CultName] member {stored.Descriptor.NameMember}", stored.Descriptor.NameAccessor!) is { Length: > 0 } name ? name : null; private static string? Read(CultStoredDocument stored, string what, Func getter) { try { return getter(stored.Document); } catch (Exception exception) { throw new InvalidOperationException( $"Record {stored.Key.Value} ({stored.Descriptor.SchemaName}) is refused: its {what} threw: {exception.GetBaseException().Message}", exception); } } // Every admission is judged here: variants resolved, then every landing record's name and index values read once, // then the rules that compare them. It refuses naming the keys, and mutates nothing. private VariantPlan Resolve( IReadOnlyList admitted, IReadOnlyList evicted, CacheBackingStore? source) { var plan = ResolveVariants(admitted, evicted, source); Project(plan); // The variant rules are judged only when a variant is involved. if (_variantKeys.Count != 0 || admitted.Any(stored => stored.Variant != null)) { RefuseVariantNameSharing(plan); RefuseVariantIndexSharing(plan, evicted); } return plan; } // Resolves every admitted variant and every variant whose base chain this admission changes, base first, against // the post-admission set. It refuses a broken variant naming the keys, and mutates nothing. private VariantPlan ResolveVariants( IReadOnlyList admitted, IReadOnlyList evicted, CacheBackingStore? source) { if (_variantKeys.Count == 0 && admitted.All(stored => stored.Variant == null)) return new VariantPlan(admitted); var incoming = admitted.ToDictionary(stored => stored.Key.Value, StringComparer.Ordinal); var gone = evicted.Select(stored => stored.Key.Value).ToHashSet(StringComparer.Ordinal); CultStoredDocument? Lookup(string key) => incoming.TryGetValue(key, out var found) ? found : gone.Contains(key) ? null : _entries.TryGetValue(key, out var held) ? held : null; // A variant is written only through the variant API: a plain record never lands on a variant's key, and the // reverse is a rebase or a flatten, never an accident. A load reports what is on disk and is not a write. if (source == null) { foreach (var stored in admitted) { _entries.TryGetValue(stored.Key.Value, out var present); if (stored.Variant == null && !stored.Flattens && present?.Variant != null) throw new InvalidOperationException( $"Record {stored.Key.Value} is a variant of {present.Variant.BaseKey}; a plain write cannot replace it. " + "Change it through the variant API, or Flatten it."); if (stored.Variant != null && present != null && present.Variant == null) throw new InvalidOperationException( $"Record {stored.Key.Value} is a plain record; it cannot become a variant of {stored.Variant.BaseKey} in place. Use a new key."); if (stored.Variant != null && present?.Variant != null && present.Descriptor.DocumentType != stored.Descriptor.DocumentType) throw new InvalidOperationException( $"Variant {stored.Key.Value} is a {present.Descriptor.DocumentType.Name}; " + $"rebasing it onto {stored.Variant.BaseKey} would make it a {stored.Descriptor.DocumentType.Name}."); } } var changed = new HashSet(incoming.Keys.Concat(gone), StringComparer.Ordinal); // A flatten resolves like the variant it replaces, against the post-batch set, so a base edited in the same // batch is in it; it lands as the plain record of that resolution. var toResolve = new List(admitted.Where(stored => stored.Variant != null || stored.Flattens).Select(stored => stored.Key.Value)); foreach (var variantKey in _variantKeys) { if (incoming.ContainsKey(variantKey) || gone.Contains(variantKey)) continue; var next = _entries[variantKey].Variant!.BaseKey; for (var steps = 0; steps <= _entries.Count; steps++) { if (changed.Contains(next)) { toResolve.Add(variantKey); break; } var link = Lookup(next); if (link?.Variant == null) break; next = link.Variant.BaseKey; } } if (toResolve.Count == 0) return new VariantPlan(admitted); var pending = new HashSet(toResolve, StringComparer.Ordinal); var done = new Dictionary(StringComparer.Ordinal); var stack = new List(); foreach (var key in toResolve) ResolveOne(key); var plan = new VariantPlan(admitted.Select(stored => done.TryGetValue(stored.Key.Value, out var resolved) ? resolved : stored).ToList()); foreach (var key in toResolve.Where(key => !incoming.ContainsKey(key))) plan.Dependents.Add(done[key]); return plan; CultStoredDocument ResolveOne(string key) { if (done.TryGetValue(key, out var finished)) return finished; var variant = Lookup(key)!; if (stack.Contains(key)) throw CycleRefusal(stack.SkipWhile(entry => entry != key).Append(key)); stack.Add(key); // A flatten carries no delta of its own; the variant it replaces does. var isVariant = variant.Variant != null; var delta = variant.Variant ?? _entries[key].Variant!; var baseStored = Lookup(delta.BaseKey) ?? throw MissingBaseRefusal(key, delta.BaseKey, gone.Contains(delta.BaseKey)); if (pending.Contains(delta.BaseKey)) baseStored = ResolveOne(delta.BaseKey); var descriptor = variant.Descriptor; if (_codec == null) throw new InvalidOperationException( $"Variant {key} needs a codec to resolve; this cache has none. Create it with CultCacheMessagePack.Create."); if (baseStored.Descriptor.DocumentType != descriptor.DocumentType) throw new InvalidOperationException( $"Variant {key} is a {descriptor.DocumentType.Name} but its base {delta.BaseKey} is a {baseStored.Descriptor.DocumentType.Name}; " + "a variant has its base's concrete type."); var type = descriptor.DocumentType; var kept = new List>(); var overrides = new List(delta.Overrides.Count); var minted = false; var seen = new HashSet(); foreach (var entry in delta.Overrides) { if (entry.Op != CultOverrideOp.Set || entry.Path.Count != 1 || entry.Path[0].ElementId.Length != 0 || entry.Id.Length != 0) throw new InvalidOperationException( $"Variant {key} carries {entry.Op} at {entry.PathText}; this runtime resolves top-level Set overrides only."); var slot = entry.Path[0].Slot; if (!seen.Add(slot)) throw new InvalidOperationException($"Variant {key} overrides slot {slot} more than once; a variant has one override per slot."); var member = descriptor.RichMembers.FirstOrDefault(candidate => candidate.Slot == slot); // Drift is the load's business (ToStoredDocument sheds an override of a slot the type lacks), so a delta // reaching here names only slots the type has, unless someone is writing one it never had. if (member == null) throw new InvalidOperationException($"Variant {key} overrides slot {slot}, which {type.Name} does not have."); object? value; try { value = _codec.Deserialize(member.MemberType, type, entry.Value); } catch (Exception exception) { // A value that no longer decodes refuses exactly as a plain record holding it does. throw new InvalidOperationException( $"Variant {key} (schema '{descriptor.SchemaId}') overrides '{member.Member.Name}' (slot {slot}) with a value that does not decode as {member.MemberType.Name}: " + exception.GetBaseException().Message, exception); } // The override's value is minted like a plain record's members: ids the store never held are filled here, in the // decoded copy, and the delta this hold lands carries them. var bytes = entry.Value; var idPlan = CultElementIds.Plan(member.MemberType, value, key + "." + slot.ToString(CultureInfo.InvariantCulture), deterministic: source != null); if (idPlan.Count > 0) { idPlan.Apply(); bytes = _codec.Serialize(value!, member.MemberType, type); minted = true; } overrides.Add(new CultVariantOverride(entry.Op, entry.Path, entry.Id, bytes)); kept.Add(new KeyValuePair(slot, bytes)); } var payload = _codec.Serialize(baseStored.Document, type, type); var document = _codec.Deserialize(type, type, kept.Count == 0 ? payload : _codec.Overlay(payload, kept)) ?? throw new InvalidOperationException($"Variant {key} resolved to nothing."); stack.RemoveAt(stack.Count - 1); return done[key] = isVariant ? variant.Resolved(document, minted ? new CultVariantDelta(delta.BaseKey, overrides) : delta, variant.IdsInMemoryOnly || (minted && source != null), _codec) : new CultStoredDocument(variant.Key, variant.StoredAt, variant.Descriptor, document) { Flattens = true, StoredSchemaId = variant.StoredSchemaId, StoredVersion = variant.StoredVersion }; } } // R6: the name index maps one name to one key, so a variant that kept its base's name would shadow it. A variant's // name is its projection; an ancestor that lands is read from its projection, one that stays is read as held. private void RefuseVariantNameSharing(VariantPlan plan) { var landing = plan.Admitted.Concat(plan.Dependents).ToDictionary(stored => stored.Key.Value, StringComparer.Ordinal); // Resolution already refused a base that is missing or removed, so an ancestor is landing or held. CultStoredDocument? Ancestor(string key) => landing.TryGetValue(key, out var landed) ? landed : _entries.TryGetValue(key, out var held) ? held : null; string? NameOf(CultStoredDocument stored) => plan.Projections.TryGetValue(stored.Key.Value, out var projection) ? projection.Name : NameRead(stored); foreach (var variant in landing.Values.Where(stored => stored.Variant != null)) { var variantName = plan.Projections[variant.Key.Value].Name; if (variantName == null) continue; for (var ancestor = Ancestor(variant.Variant!.BaseKey); ancestor != null; ancestor = ancestor.Variant == null ? null : Ancestor(ancestor.Variant.BaseKey)) { if (NameOf(ancestor) == variantName) throw new InvalidOperationException( $"Variant {variant.Key.Value} and its base {ancestor.Key.Value} are both named '{variantName}'; a variant must override its name."); } } } // 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) { 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>(); 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 holdersOf = new Dictionary<((Type Type, string Alias) Index, string Value), (string[] All, string[] Variants)>(); foreach (var (stored, index, value) in landedValues) { if (!holdersOf.TryGetValue((index, value), out var known)) { var all = (_indexes.TryGetValue(index, out var held) && held.TryGetValue(value, out var heldKeys) ? heldKeys.Where(heldKey => !replaced.Contains(heldKey)) : Enumerable.Empty()) .Concat(landed[(index, value)]) .Distinct(StringComparer.Ordinal) .ToArray(); holdersOf[(index, value)] = known = (all, all.Where(IsVariant).ToArray()); } var self = stored.Key.Value; var holder = known.All.FirstOrDefault(candidate => candidate != self); if (stored.Variant != null && holder != null) throw new InvalidOperationException( $"Variant {self} and {holder} both hold index {index.Alias} '{value}'; a variant must override its indexed member."); var variantHolder = known.Variants.FirstOrDefault(candidate => candidate != self); if (stored.Variant == null && variantHolder != null) throw new InvalidOperationException( $"Record {self} and variant {variantHolder} both hold index {index.Alias} '{value}'; " + "a variant may not share an indexed value with another record."); } } private static InvalidOperationException CycleRefusal(IEnumerable keys) => new($"Variant bases form a cycle: {string.Join(" -> ", keys)}."); private static InvalidOperationException MissingBaseRefusal(string variant, string baseKey, bool removed) => new(removed ? $"Record {baseKey} cannot go: variant {variant} is based on it. Remove or rebase the variant in the same commit." : $"Variant {variant} is based on {baseKey}, which does not exist."); private List Apply( VariantPlan plan, IReadOnlyList evicted, CacheBackingStore? source) { var admitted = plan.Admitted; var changes = new List(admitted.Count + evicted.Count + plan.Dependents.Count); foreach (var stored in evicted) { if (!_entries.TryGetValue(stored.Key.Value, out var existing)) continue; Unindex(existing); _entries.Remove(stored.Key.Value); _handles.Remove(existing.Document); _variantKeys.Remove(stored.Key.Value); changes.Add(new Change(CultCacheDocumentChangeKind.Removed, existing, null, existing.Document, ++_sequence)); } foreach (var stored in admitted.Concat(plan.Dependents)) { _entries.TryGetValue(stored.Key.Value, out var previous); if (previous != null) { Unindex(previous); _handles.Remove(previous.Document); } _entries[stored.Key.Value] = stored; _handles.AddOrUpdate(stored.Document, new KeyBox(stored.Key)); Index(stored, plan.Projections[stored.Key.Value]); if (stored.Variant == null) _variantKeys.Remove(stored.Key.Value); else _variantKeys.Add(stored.Key.Value); changes.Add(new Change( previous == null ? CultCacheDocumentChangeKind.Added : CultCacheDocumentChangeKind.Updated, stored, stored.Document, previous?.Document, ++_sequence)); } if (source == null && _stores.Count == 0) _dirtyInMemory = true; return changes; } // The most derived routed home type assignable from the document type, else the untyped store. private CacheBackingStore? Home(Type documentType) { CacheBackingStore? untyped = null; CacheBackingStore? routed = null; Type? best = null; foreach (var (store, homes) in _stores) { if (homes.Length == 0) { untyped = store; continue; } foreach (var home in homes) { if (home.IsAssignableFrom(documentType) && (best == null || best.IsAssignableFrom(home))) { best = home; routed = store; } } } return routed ?? untyped; } private T? Single(string lookup, Func> keys) where T : class { lock (_gate) { var found = keys().Distinct().ToArray(); if (found.Length > 1) throw new InvalidOperationException( $"{typeof(T).Name} {lookup} matches records {string.Join(", ", found)}; look it up by its concrete type."); return found.Length == 1 && _entries.TryGetValue(found[0], out var stored) ? stored.Document as T : null; } } private void Index(CultStoredDocument stored, Projection projection) { var type = stored.Descriptor.DocumentType; if (stored.Descriptor.IsGlobal) _globals[type] = stored.Key.Value; 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); } // Documents are mutable, so a name or index value may have changed since it was indexed: drop by key. private void Unindex(CultStoredDocument stored) { var type = stored.Descriptor.DocumentType; if (_globals.TryGetValue(type, out var globalKey) && globalKey == stored.Key.Value) _globals.Remove(type); 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); } } private static Dictionary MapOf(Dictionary> maps, TKey key) where TKey : notnull { if (!maps.TryGetValue(key, out var map)) maps[key] = map = new Dictionary(StringComparer.Ordinal); return map; } private static void RemoveKey(Dictionary map, string key) { foreach (var stale in map.Where(pair => pair.Value == key).Select(pair => pair.Key).ToArray()) map.Remove(stale); } private sealed class KeyBox { public KeyBox(CultRecordKey key) { Key = key; } public CultRecordKey Key { get; } } // What one outermost hold admitted, and what its journals threw. private sealed class Hold { public List<(Change Change, bool Loaded)> Changes { get; } = new(); public List Failures { get; } = new(); } private sealed class Change { public Change(CultCacheDocumentChangeKind kind, CultStoredDocument stored, object? document, object? previous, long sequence) { Kind = kind; Stored = stored; Document = document; Previous = previous; Sequence = sequence; } public CultCacheDocumentChangeKind Kind { get; } public CultStoredDocument Stored { get; } public object? Document { get; } public object? Previous { get; } public long Sequence { get; } } } public abstract class CacheBackingStore : IDisposable { private CultDocumentRegistry? _registry; private CultSchemaMigrationReport[] _lastSchemaMigrationReports = Array.Empty(); protected CacheBackingStore(bool readOnly = false) { IsReadOnly = readOnly; } protected CultDocumentRegistry Registry => _registry ?? throw new InvalidOperationException("Backing store is not attached to a CultDocumentRegistry."); // Null for a store no registry was attached to (CultMesh's raw single-file write). private protected CultDocumentRegistry? AttachedRegistry => _registry; protected ConcurrentDictionary Entries { get; } = new(StringComparer.Ordinal); public bool IsReadOnly { get; } public bool IsDirty { get; protected set; } public bool FlushOnDispose { get; set; } public IReadOnlyList LastSchemaMigrationReports => _lastSchemaMigrationReports; /// /// The records this store's last load found and carries untouched, ordered by key: no registered type owns their schema id or /// lists it as compatible. See . /// public IReadOnlyList ForeignRecords => _foreignRecords; private CultForeignRecord[] _foreignRecords = Array.Empty(); // Set by the cache at attach. A pull hands over everything it loaded and dropped in one call before it adopts // any of it; if the cache refuses a record the call throws and the store keeps its previous view. protected internal Action, IReadOnlyList>? Loaded; private readonly object _detachedGate = new(); internal CultCache? Cache; private const string DirectoryFormatV4 = "cultcache.store.v4.directory-content-addressed-pages"; private const string DirectoryFormatV5 = "cultcache.store.v5.directory-content-addressed-pages"; /// /// The one header decision, for every writer of every store: the string a store file's format header must carry after /// this write. /// /// /// The marked header (v3, or the directory store's v5) says the store holds an element id: a reader older than ids refuses /// it, because it would skip the id slots and rewrite the elements without them. A writer that rewrites the whole store /// () decides by content, so a store drops back to unmarked when nothing in it holds an id. /// A writer that touches only part of the store keeps a header already marked (), because /// it cannot see what the rest holds. is about the records this write knows are on disk. /// Unmarked, a single-file store is v2 exactly when it holds a variant () and otherwise v1; /// the directory store refuses variants and is v4. Pass the header the file carries, or null when it has none, and refuse /// a file whose header you cannot read before calling this. /// protected internal static string HeaderFor(bool holdsIds, string? existingHeader, bool wholeStore, bool directoryStore, bool holdsVariants = false) { var marked = directoryStore ? DirectoryFormatV5 : CultPersistedStoreSnapshot.FormatV3; if (holdsIds || (!wholeStore && string.Equals(existingHeader, marked, StringComparison.Ordinal))) return marked; return directoryStore ? DirectoryFormatV4 : holdsVariants ? CultPersistedStoreSnapshot.FormatV2 : CultPersistedStoreSnapshot.FormatV1; } // The header a write leaves, for either store kind: HeaderFor decides. An element id is an ordinary keyed string, so only a // reader with the record's type can tell from its bytes whether it holds one; the store header is how a reader without the types // learns it. A write that copies any // record forward keeps a marked header; a write that encodes every record the store will hold decides by content. An // opaque payload (null) never marks an unmarked store, and keeps a marked one marked. A copied record the writer decodes // under the store's lock holding an id marks the store too (SingleFileBackingStore.CopiedHoldIds): another runtime may // have written ids under an unmarked header. Nothing a load or a decode records can unmark a store. protected static string HeaderForWrite(string? durableHeader, IReadOnlyCollection written, bool copiesAny, bool copiedHoldIds, bool directoryStore, bool holdsVariants) => HeaderFor( holdsIds: copiedHoldIds || written.Any(value => value == true), durableHeader, wholeStore: !copiesAny && written.All(value => value != null), directoryStore, holdsVariants); // Once attached, a hold on the cache's gate: the store and its cache share one lock, so a load calling back into // the cache can never take the gate after the store lock, and a direct call publishes what it loaded when it // returns. File locks are always taken inside it. protected T Held(Func body) { if (Cache is { } cache) return cache.Held(body); lock (_detachedGate) return body(); } protected void Held(Action body) => Held(() => { body(); return true; }); private protected CultDocumentRegistry? SwapRegistry(CultDocumentRegistry? registry) { var previous = _registry; _registry = registry; return previous; } internal void AttachRegistry(CultDocumentRegistry registry) { _registry = registry ?? throw new ArgumentNullException(nameof(registry)); } public abstract void PullAll(); public abstract void Push(CultStoredDocument entry); public abstract void Delete(CultStoredDocument entry); public abstract void PushAll(); /// /// One durable step under the store's exclusive lock: the batch and every staged write land onto the store as it is then, /// and the batch's conditions are evaluated against it. A failure restores the store's staged view. /// /// /// An implementation judges , and its own , under that /// lock with the keys it writes: a key whose record is under an id no registered type declares refuses the commit, and a /// checked key is never removed. The cache calls on every other store before this commit. Each /// foreign record a store lists comes from with the version it read, so pass to /// . /// public abstract CultCommitOutcome CommitBatch(CultCommitRequest request, bool wait); public virtual void Dispose() { if (FlushOnDispose && (IsDirty || HasStagedChecks) && !IsReadOnly) { PushAll(); } } protected void ThrowIfReadOnly() { if (IsReadOnly) throw new InvalidOperationException($"Backing store {this} is read-only."); } protected CultPersistedRecord ToPersistedRecord(CultStoredDocument entry, Func serializePayload) { // A variant persists its delta and an empty payload; the resolved view is never written. return new CultPersistedRecord { Key = entry.Key.Value, SchemaId = entry.Descriptor.SchemaId, StoredAt = entry.StoredAt, Payload = entry.Variant != null ? Array.Empty() : serializePayload(entry.Document), Variant = entry.Variant }; } // The one load surface: a persisted record resolved against the current type, with the report of what that // resolution shed. A plain record sheds a member its type dropped by never carrying it into the document; a variant // sheds the override of such a slot by never carrying it into the delta, so a later type reusing the slot number // cannot resurrect it and a flush needs no rule of its own. The report is the schema diff's ignored slots plus the // overrides whose slot the type does not have, including one the persisted schema never listed. The version is the reading // store's: it read the bytes. protected CultStoredDocument ToStoredDocument( CultPersistedRecord record, string version, IReadOnlyCollection catalog, Func deserializePayload, out CultSchemaMigrationReport report) { var resolution = Registry.ResolvePersistedSchemaDetailed(record.SchemaId, catalog); report = resolution.Report; if (record.Variant != null) { var descriptor = resolution.Descriptor; bool Has(CultVariantOverride entryOverride) => entryOverride.Path.Count != 1 || descriptor.RichMembers.Any(member => member.Slot == entryOverride.Path[0].Slot); var shed = record.Variant.Overrides .Where(entryOverride => !Has(entryOverride)) .Select(entryOverride => entryOverride.Path[0].Slot) .Distinct() .OrderBy(slot => slot) .ToArray(); if (shed.Length > 0) { // A slot the schema diff already reported is not reported twice. var unreported = shed.Except(report.IgnoredExtraSlots).ToArray(); report.IgnoredExtraSlots = report.IgnoredExtraSlots.Concat(unreported).OrderBy(slot => slot).ToArray(); report.Warnings = report.Warnings.Concat(unreported.Select(slot => new CultSchemaMigrationWarning { Code = "ignored_extra_slot", Message = $"Variant {record.Key} overrides slot {slot}, which {descriptor.DocumentType.Name} does not have; the override is ignored and shed on flush." })).ToArray(); } // The cache resolves it against its base when it admits the load. return new CultStoredDocument( new CultRecordKey(record.Key), record.StoredAt, descriptor, new CultVariantDelta(record.Variant.BaseKey, record.Variant.Overrides.Where(Has).ToArray()), null, Cache?.Codec) { StoredSchemaId = record.SchemaId, StoredVersion = version }; } object document; try { document = deserializePayload(resolution.Descriptor.DocumentType, record.Payload); } catch (Exception exception) when (exception is not OperationCanceledException) { // The codec knows what it refused but not which record; the store does. Say both, and never null the record. var causes = new List(); for (var cause = exception; cause != null; cause = cause.InnerException) causes.Add(cause.Message); throw new InvalidDataException($"Record '{record.Key}' (schema '{record.SchemaId}') cannot be decoded: {string.Join(" <- ", causes.Distinct())}", exception); } return new CultStoredDocument( new CultRecordKey(record.Key), record.StoredAt, resolution.Descriptor, document) { StoredSchemaId = record.SchemaId, StoredVersion = version }; } protected void SetLastSchemaMigrationReports(IEnumerable reports) { _lastSchemaMigrationReports = reports?.ToArray() ?? Array.Empty(); } // The one test of whether a stored record is foreign: no registered type claims its schema id through the catalog it was // stored with. It reads the record's header only, never its payload. Null for a record the registry resolves. // The one constructor path for a foreign record, so every store kind, one outside this library included, lists it at the // version it read. protected CultForeignRecord? Foreign(CultPersistedRecord record, IReadOnlyCollection catalog, string storedVersion) => Registry.Claimant(record.SchemaId, catalog, out var persisted) == null ? new CultForeignRecord(record.Key, record.SchemaId, persisted!.SchemaName, record.StoredAt, storedVersion) : null; protected void SetForeignRecords(IEnumerable foreign) { _foreignRecords = foreign.OrderBy(record => record.Key, StringComparer.Ordinal).ToArray(); } // A cache never writes, relabels or removes a record stored under an id its types do not declare: the staging half of rule 3. // Refused when the store carries a foreign record at the key, or holds one whose stored id no registered type owns or declares // compatible (a record a type reads by the catalog's schema name or a persisted entry's list: read-only). Called for every // record a cache write admits or evicts, before anything is staged. protected internal void RefuseUndeclared(string key) { var foreign = _foreignRecords.FirstOrDefault(record => string.Equals(record.Key, key, StringComparison.Ordinal)); if (foreign != null) throw Overwrites(null, new[] { (foreign.Key, foreign.SchemaId, foreign.SchemaName) }); if (Entries.TryGetValue(key, out var held) && !Registry.Declares(held.StoredSchemaId)) throw Overwrites(null, new[] { (key, held.StoredSchemaId, held.Descriptor.SchemaName) }); } /// /// Called when the cache removes keys it does not hold, outside a batch that commits through this store. Such a removal /// removes nothing, but a store that holds a record at one of these keys under an id no registered type declares must /// refuse it rather than let the caller believe the record is gone. /// /// /// /// The default refuses now, from this store's view: the foreign records its last load listed and the records it holds /// (). It reads nothing, so it cannot see a record another writer put there after that load. /// /// /// A writable store kind that reads its durable state under a lock overrides this to call , and /// judges at its next or as it judges /// : under the lock, with the keys it writes, refused through the same refusal, never /// removed. It calls when that step succeeds, which clears them. A read-only store has no /// such step and keeps the default. Both shipped store kinds do this. A batch's own checks are judged through /// . /// /// protected internal virtual void RefuseRemovalOfUndeclared(IReadOnlyCollection keys) { foreach (var key in keys) RefuseUndeclared(key); } /// /// Called for a batch's checks on every attached store that is not the batch's home, before the home commits. It judges /// only the keys it is given: it never writes, stages, clears or calls /// . /// /// /// The default answers from this store's view () and reads nothing, which is right for a /// read-only store. A writable store kind that reads its durable state under a lock overrides this to read under that /// lock and refuse through , returning when it may not wait /// and the lock is held. A refusal or contention here writes nothing anywhere, because the home has not yet committed. /// protected internal virtual CultCommitOutcome JudgeChecks(IReadOnlyCollection keys, bool wait) { foreach (var key in keys) RefuseUndeclared(key); return CultCommitOutcome.Committed; } // Keys a removal named that this store's cache does not hold, waiting for this store's next flush or commit to judge them. // A refusal takes back the keys it refused; a successful flush or commit clears them. private readonly HashSet _checks = new(StringComparer.Ordinal); /// /// The keys staged for this store's next flush or commit to judge. They are not writes: /// does not count them. /// protected IReadOnlyCollection StagedChecks => _checks; /// /// Stages keys for this store's next flush or commit to judge without removing them. See /// . /// protected void StageChecks(IEnumerable keys) => Held(() => { foreach (var key in keys) _checks.Add(key); }); internal bool HasStagedChecks => _checks.Count > 0; // A stored record under an id the writer does not declare, named by the catalog entry that publishes the id. Header only: no // payload is decoded. Null for a record under a declared id. protected static (string Key, string SchemaId, string SchemaName)? Undeclared( CultPersistedRecord record, IReadOnlyCollection catalog, Func declares) => declares(record.SchemaId) ? null : (record.Key, record.SchemaId, CultDocumentRegistry.PublishedBy(record.SchemaId, catalog)?.SchemaName ?? string.Empty); // The apply half of rule 3, for every store kind, checked where the write lands under the store's lock. A key this write names // whose durable record is under an id no registered type declares is refused, every such key at once, through Refuse, which // reloads it as the store holds it: listed foreign when no type reads it, held read-only when one does. protected void RefuseUndeclaredOnDisk( IEnumerable written, Func undeclaredAt, Func durableAt, Func foreignAt, Action unstage, bool cacheless = false) { var refused = written.Distinct(StringComparer.Ordinal) .Select(undeclaredAt) .OfType<(string Key, string SchemaId, string SchemaName)>() .OrderBy(record => record.Key, StringComparer.Ordinal) .ToArray(); if (refused.Length == 0) return; Refuse(refused.Select(record => record.Key), durableAt, foreignAt, unstage, inner => Overwrites(inner, refused, cacheless)); } /// /// The one refusal of a write, for every store kind: it reloads what it refuses, or changes nothing. For each key, /// gives the durable record decoded as a load would (null when the store holds none, or holds /// it as a foreign record), and the foreign record it holds there, if any. What differs from this /// store's view is handed to , the path a pull uses; only when that succeeds does the store forget its /// staged change at each key (), take the loaded and dropped records into its view, and list the /// keys' foreign records as a pull lists them. If the reload throws, nothing has changed and the refusal carries that /// exception. Always throws what returns. /// protected void Refuse( IEnumerable keys, Func durableAt, Func foreignAt, Action unstage, Func refusal) { var refused = keys.Distinct(StringComparer.Ordinal).OrderBy(key => key, StringComparer.Ordinal).ToArray(); var loaded = new List(); var dropped = new List(); var foreign = new List(); try { foreach (var key in refused) { if (foreignAt(key) is { } carried) foreign.Add(carried); Entries.TryGetValue(key, out var held); var durable = durableAt(key); if (durable == null) { if (held != null) dropped.Add(held); } else if (held == null || !SameStoredIdentity(held.StoredVersion, durable.StoredVersion)) loaded.Add(durable); } if (loaded.Count > 0 || dropped.Count > 0) Loaded?.Invoke(loaded, dropped); } catch (Exception ex) { throw refusal(ex); } foreach (var key in refused) { unstage(key); _checks.Remove(key); } foreach (var held in dropped) Entries.TryRemove(held.Key.Value, out _); foreach (var stored in loaded) Entries[stored.Key.Value] = stored; SetForeignRecords(_foreignRecords.Where(record => !refused.Contains(record.Key, StringComparer.Ordinal)).Concat(foreign)); throw refusal(null); } // Two stored records are the same stored bytes when their versions are equal. The one place a pull's reload filter, the // staleness check and a refusal's reload decide it. protected static bool SameStoredIdentity(string? version, string? otherVersion) => string.Equals(version, otherVersion, StringComparison.Ordinal); // The one spelling of a version in both store kinds: the lowercase hex of the SHA-256 of a record's persisted encoding. protected static string VersionOf(byte[] sha256) { var hex = new StringBuilder(sha256.Length * 2); foreach (var value in sha256) hex.Append(value.ToString("x2", CultureInfo.InvariantCulture)); return hex.ToString(); } // The one re-mint, for every store kind (storedat-on-change): a record written over one the store holds is stored at a storedAt // later than the one it replaces, whoever wrote that. The staged storedAt stands only when it is already later (both parsed // RoundtripKind; an unparsable one counts as not later); otherwise it is minted from the replaced one. protected string StoredAtOver(string storedAt, string replacedAt) => StoredLater(storedAt, replacedAt) ? storedAt : CultCache.MintStoredAt(replacedAt, (Cache?.Clock ?? (() => DateTimeOffset.UtcNow))()); private static bool StoredLater(string storedAt, string than) => DateTimeOffset.TryParse(storedAt, CultureInfo.InvariantCulture, DateTimeStyles.RoundtripKind, out var later) && DateTimeOffset.TryParse(than, CultureInfo.InvariantCulture, DateTimeStyles.RoundtripKind, out var earlier) && later > earlier; // The one refusal of a write onto records stored under ids no registered type declares, carried foreign or held read-only. // It names each record's key, stored id and schema name, never its payload. When the reload that refusal owes failed, the // refusal carries that failure as its inner exception. // A cacheless writer (a CultMesh single-file write) declares only the catalog entry it writes, and has nothing to reload. protected CultSchemaConflictException Overwrites(Exception? inner, (string Key, string SchemaId, string SchemaName)[] refused, bool cacheless = false) => new( $"{(refused.Length == 1 ? "Record" : "Records")} {string.Join(", ", refused.Select(record => $"'{record.Key}' (schema id '{record.SchemaId}', '{record.SchemaName}')"))} in {this}: " + (cacheless ? "the catalog entry this write publishes neither owns this schema id nor lists it compatible, so the write never relabels or removes it; nothing was written and nothing was reloaded. " + "List the id as compatible on the schema that should own it." : "no registered type owns this schema id or declares it compatible, so this cache never writes, relabels or removes it. " + "Declare the id compatible on the type that should own it.") + (inner == null ? string.Empty : " Reloading what the write refused failed, so the cache is unchanged."), refused[0].SchemaId, refused.Select(record => record.SchemaName).Distinct(StringComparer.Ordinal).ToArray(), refused[0].Key, inner); // A flush or commit succeeded: nothing staged is left, writes or checks. protected void MarkFlushSucceeded() { IsDirty = false; _checks.Clear(); } } public abstract class SingleFileBackingStore : CacheBackingStore { protected SingleFileBackingStore(string filePath, bool readOnly = false) : base(readOnly) { FileInfo = new FileInfo(filePath); } protected FileInfo FileInfo { get; } // Keys this cache has written or removed here since the store last wrote them: the only keys a foreign record can refuse. // A key is staged exactly while the store is dirty with it, and a pull runs only on a clean store. private readonly HashSet _staged = new(StringComparer.Ordinal); // The version of every record the store held when this store last read it, as this store last wrote it for keys it wrote. // A write into a store holding a variant lands only while the durable versions still equal these. private readonly Dictionary _lastRead = new(StringComparer.Ordinal); /// /// A store file from its parts, each already encoded: a catalog entry or record this store carries forward is the bytes its /// file held, one it writes is or of the object. /// protected abstract byte[] SerializeStore(string formatVersion, IReadOnlyList catalogEntries, IReadOnlyList records); // One record's persisted encoding: the bytes a version names, and the bytes a written record is stored as. protected abstract byte[] SerializeRecord(CultPersistedRecord record); // One catalog entry's persisted encoding, for an entry the writer built. protected abstract byte[] SerializeCatalogEntry(CultSchemaCatalogEntry entry); /// /// This store's one reader of a non-empty store file, and so its one verdict on whether the file may be replaced: open, /// flush and commit all ask it. It must throw for anything that is not exactly one /// store this runtime reads (a truncated file, bytes after the store, a missing or extra slot, a format or record it does /// not know), so a rewrite never overwrites a file it cannot see. It returns the decoded snapshot and, parallel to the /// snapshot's catalog and records, the exact bytes the file holds for each: the store keeps those, never the snapshot's objects. /// protected abstract (CultPersistedStoreSnapshot Snapshot, byte[][] CatalogBytes, byte[][] RecordBytes) ReadStore(byte[] data); // Test seam: how many record encodings this store has hashed, to prove what a write hashes. internal int RecordsHashed; // Test seam: how many copied records a write decoded to ask whether they hold an element id. internal int CopiedRecordsDecoded; private string VersionOfBytes(byte[] bytes) { RecordsHashed++; using var sha256 = SHA256.Create(); return VersionOf(sha256.ComputeHash(bytes)); } // What the store's file holds, as the store owns it: the snapshot as read, and keyed by record key the exact slice the file // holds for each record, whose version is the SHA-256 of that slice, hashed when a comparison first asks. Catalog entries // keep their slices by the entry object this read produced. A write copies a record it did not stage from its slice here // and re-encodes every record it was handed, so no object a caller edits is ever mistaken for the bytes in the file. private sealed class Durable { private readonly SingleFileBackingStore _owner; private readonly Dictionary _records; private readonly Dictionary _versions = new(StringComparer.Ordinal); private readonly Dictionary _catalog = new(); internal Durable(SingleFileBackingStore owner, (CultPersistedStoreSnapshot Snapshot, byte[][] CatalogBytes, byte[][] RecordBytes) read) { _owner = owner; Snapshot = read.Snapshot; _records = new Dictionary(StringComparer.Ordinal); for (var index = 0; index < Snapshot.Records.Length; index++) _records[Snapshot.Records[index].Key] = read.RecordBytes[index]; for (var index = 0; index < Snapshot.SchemaCatalog.Length; index++) _catalog.Add(Snapshot.SchemaCatalog[index], read.CatalogBytes[index]); } internal CultPersistedStoreSnapshot Snapshot { get; } internal byte[] Slice(string key) => _records[key]; internal string Version(string key) { if (!_versions.TryGetValue(key, out var version)) _versions[key] = version = _owner.VersionOfBytes(_records[key]); return version; } // The bytes to write for a catalog entry the write chose: the slice the file held when it is one of this read's entries, // otherwise an encoding of the entry the writer built. internal byte[] CatalogBytes(CultSchemaCatalogEntry entry) => _catalog.TryGetValue(entry, out var bytes) ? bytes : _owner.SerializeCatalogEntry(entry); } protected abstract byte[] SerializePayload(object document, Type documentType); protected abstract object DeserializePayload(Type documentType, byte[] payload); public override string ToString() => FileInfo.FullName; public override void PullAll() => Held(PullAllCore); private void PullAllCore() { // A single-file snapshot cannot distinguish local dirty keys from clean keys. Pulling while local mutations // are staged would let an older disk snapshot erase them before the next flush. Flush first. if (IsDirty) return; var durable = ReadSnapshot(); if (durable == null) { _lastRead.Clear(); SetLastSchemaMigrationReports(Array.Empty()); SetForeignRecords(Array.Empty()); return; } var snapshot = durable.Snapshot; var reports = new List(snapshot.Records.Length); var persisted = new Dictionary(StringComparer.Ordinal); var foreign = new List(); var versions = new Dictionary(StringComparer.Ordinal); foreach (var record in snapshot.Records) { var version = versions[record.Key] = durable.Version(record.Key); if (Foreign(record, snapshot.SchemaCatalog, version) is { } carried) { foreign.Add(carried); continue; } var stored = ToStoredDocument(record, version, snapshot.SchemaCatalog, DeserializePayload, out var report); reports.Add(report); persisted[stored.Key.Value] = stored; } var loaded = persisted.Values .Where(stored => !Entries.TryGetValue(stored.Key.Value, out var existing) || !SameStoredIdentity(existing.StoredVersion, stored.StoredVersion)) .ToArray(); var dropped = Entries.Values.Where(existing => !persisted.ContainsKey(existing.Key.Value)).ToArray(); if (loaded.Length > 0 || dropped.Length > 0) Loaded?.Invoke(loaded, dropped); foreach (var stored in dropped) Entries.TryRemove(stored.Key.Value, out _); foreach (var stored in loaded) Entries[stored.Key.Value] = stored; _lastRead.Clear(); foreach (var (key, version) in versions) _lastRead[key] = version; SetLastSchemaMigrationReports(reports); SetForeignRecords(foreign); } public override void Push(CultStoredDocument entry) { ThrowIfReadOnly(); Held(() => { Entries[entry.Key.Value] = entry; _staged.Add(entry.Key.Value); IsDirty = true; }); } public override void Delete(CultStoredDocument entry) { ThrowIfReadOnly(); Held(() => { Entries.TryRemove(entry.Key.Value, out _); _staged.Add(entry.Key.Value); IsDirty = true; }); } // A flush applies this store's staged writes and removals onto what the file holds now, under the lock: two writers never // interleave bytes, every record this cache did not stage stays as the file has it, and a plain flush is last-writer-wins // only per staged key. A key whose durable record is under an id this cache does not declare refuses the write, and in a store // holding a variant it lands only onto the store as this cache last read it (both ApplyWriteSet). // A file it cannot read (truncated, trailing bytes, not a store, a format it does not read) is refused and left as it is. public override void PushAll() { ThrowIfReadOnly(); Held(() => { using (AcquireLock(wait: true)) { var disk = ReadSnapshot(); var (upserts, removals) = StagedWrites(Array.Empty(), Array.Empty()); Wrote(ApplyWriteSet(disk, upserts, removals, Array.Empty())); } _staged.Clear(); MarkFlushSucceeded(); }); } public override CultCommitOutcome CommitBatch(CultCommitRequest request, bool wait) { ThrowIfReadOnly(); return Held(() => CommitBatchCore(request, wait)); } // Conditions are checked against the file under the lock, and only the condition differs between a conditional and an // unconditional commit: both apply the batch and every staged write onto the file as it is. private CultCommitOutcome CommitBatchCore(CultCommitRequest request, bool wait) { using var fileLock = AcquireLock(wait); if (fileLock == null) return CultCommitOutcome.Contended; var disk = ReadSnapshot(); if (request.HasConditions && !request.ConditionsHold( (disk?.Snapshot.Records ?? Array.Empty()).ToDictionary(record => record.Key, record => disk!.Version(record.Key), StringComparer.Ordinal), Entries.Values, ForeignRecords)) return CultCommitOutcome.Mismatch; var (upserts, removals) = StagedWrites(request.Upserts, request.Deletes); var written = ApplyWriteSet(disk, upserts, removals, request.Checks); foreach (var entry in request.Deletes) Entries.TryRemove(entry.Key.Value, out _); foreach (var entry in request.Upserts) Entries[entry.Key.Value] = entry; Wrote(written); _staged.Clear(); MarkFlushSucceeded(); return CultCommitOutcome.Committed; } // What a write puts in the store and takes out of it: the staged keys as this cache holds them now (a key it no longer holds // is a removal), and a commit's batch, which wins over a staged write of the same key. Records it did not stage are not here. private (List Upserts, List Removals) StagedWrites( IReadOnlyList batchUpserts, IReadOnlyList batchDeletes) { var documents = new Dictionary(StringComparer.Ordinal); var removals = new HashSet(StringComparer.Ordinal); foreach (var key in _staged) { if (Entries.TryGetValue(key, out var held)) documents[key] = held; else removals.Add(key); } foreach (var entry in batchDeletes) { documents.Remove(entry.Key.Value); removals.Add(entry.Key.Value); } foreach (var entry in batchUpserts) { documents[entry.Key.Value] = entry; removals.Remove(entry.Key.Value); } var catalogEntries = new Dictionary(); var upserts = documents.Values.Select(document => { if (!catalogEntries.TryGetValue(document.Descriptor, out var catalogEntry)) catalogEntries[document.Descriptor] = catalogEntry = document.Descriptor.ToCatalogEntry(); return new CultStagedRecord(ToPersistedRecord(document, payload => SerializePayload(payload, payload.GetType())), catalogEntry, document.HoldsIds); }).ToList(); return (upserts, removals.ToList()); } /// /// Writes one record to the store file without a cache. Under the store's lock the file is read by the same reader a cache's /// store uses, replaces the record at its key, and every other record and catalog entry is copied /// forward as stored. The file is replaced atomically, and its format header is decided as a flush decides it: /// true marks the store as holding element ids, false leaves it unmarked, and null (a payload /// the caller cannot read) keeps whatever the file already says. A write that finds the file unreadable, or a record at /// the key under a schema id does not declare, is refused and leaves the file as it was. /// /// The record to store. /// The schema catalog entry for the record's schema id. /// 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; omitted, it decides nothing for them. It serves this call only: the store keeps no registry. /// The store is attached to a cache; write through the 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)); RefuseWhenAttached(); ForTheCall(registry, () => WriteOne(record, catalogEntry, holdsIds)); } /// /// Writes one typed document to the store file without a cache, as a cache write would. Element ids are decided before anything /// is written: an unset id is minted, a list a cache write refuses refuses the write, and the file is marked as holding element /// ids when the document holds one. A write that fails gives the minted ids back. Then the document is encoded as /// , the type its schema id and ids are taken from, with , and stored as /// stores a record: every other record and catalog entry is copied forward as stored, under the store's /// lock, and the file is replaced atomically. /// /// The registered document type. A derived instance is stored as this type. /// The record key the document is stored under. /// The document to store. /// The stored-at stamp, or null for now. /// The registry that describes and encodes the document, or null for . It serves this call only: the store keeps no registry. /// The store is attached to a cache; write through the 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) where TDocument : class { if (document == null) throw new ArgumentNullException(nameof(document)); RefuseWhenAttached(); var resolvedRegistry = registry ?? CultDocumentRegistry.Shared; var catalogEntry = resolvedRegistry.GetRequired().ToCatalogEntry(); ForTheCall(resolvedRegistry, () => { var ids = CultElementIds.Plan(typeof(TDocument), document, key.Value, deterministic: false); ids.Apply(); try { WriteOne( new CultPersistedRecord { Key = key.Value, SchemaId = catalogEntry.SchemaId, StoredAt = string.IsNullOrWhiteSpace(storedAt) ? DateTimeOffset.UtcNow.ToString("O") : storedAt!, Payload = SerializePayload(document, typeof(TDocument)) }, catalogEntry, CultElementIds.Holds(typeof(TDocument), document)); } catch { ids.Undo(); throw; } }); } // The registry is the call's argument, not store state: it is attached for the call under the store's gate and the store's // previous registry (none, for a store no cache ever attached) comes back whatever the call does. private void ForTheCall(CultDocumentRegistry? registry, Action body) => Held(() => { var previous = SwapRegistry(registry); try { body(); } finally { SwapRegistry(previous); } }); private void WriteOne(CultPersistedRecord record, CultSchemaCatalogEntry catalogEntry, bool? holdsIds) { using (AcquireLock(wait: true)) _ = ApplyWriteSet(ReadSnapshot(), new[] { new CultStagedRecord(record, catalogEntry, holdsIds) }, Array.Empty(), Array.Empty()); } private void RefuseWhenAttached() { if (Cache != null) throw new InvalidOperationException("The store is attached to a cache, write through the cache."); } // Flush, commit and Mesh's write all end here, with the durable store they read under the lock they hold. First, a key the write // names or checks whose durable record is under an id the writer does not declare refuses the write (RefuseUndeclaredOnDisk); a // write that then has nothing to write but keys it checked writes no file. Then a store that holds // a variant, or would after this write, is written only while its record versions are what this store last read: a variant // resolves against other records, and a write decodes none it did not stage, so it cannot judge another writer's. Versions // only: a write hashes other records' bytes, never decodes them. That holds for the variant gate; the one decode of a copied // record is the element-id check (CopiedHoldIds), which runs only on an unmarked store and can only mark it. Otherwise the whole write is refused and what it refused is // reloaded (Refuse). Every record the writer staged is encoded from its object now, and its version is the hash of that // encoding; every other record is copied as the slice the file holds. Returns the records written, each with the version of // the bytes written. private IReadOnlyCollection<(CultStagedRecord Staged, string Version)> ApplyWriteSet( Durable? durable, IReadOnlyCollection upserts, IReadOnlyCollection removals, IEnumerable checks) { 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); 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); foreach (var staged in upserts) { if (replaced.TryGetValue(staged.Record.Key, out var durableAt)) staged.Record.StoredAt = StoredAtOver(staged.Record.StoredAt, durableAt); } var written = upserts.Select(staged => staged.Record.Key).ToHashSet(StringComparer.Ordinal); var removed = removals.ToHashSet(StringComparer.Ordinal); var copied = durableRecords .Where(record => !written.Contains(record.Key) && !removed.Contains(record.Key)) .ToArray(); var encoded = upserts.Select(staged => (Staged: staged, Bytes: SerializeRecord(staged.Record))).ToArray(); var records = copied.Concat(upserts.Select(staged => staged.Record)).ToArray(); var fileRecords = copied.Select(record => (record.Key, Bytes: durable!.Slice(record.Key))) .Concat(encoded.Select(pair => (pair.Staged.Record.Key, pair.Bytes))) .OrderBy(pair => pair.Key, StringComparer.Ordinal) .Select(pair => pair.Bytes) .ToArray(); var catalog = CultSchemaCatalogEntry.Derive( records, upserts.Select(staged => staged.Entry).Distinct().ToArray(), durable?.Snapshot.SchemaCatalog ?? Array.Empty()) .OrderBy(entry => entry.SchemaName, StringComparer.Ordinal) .ThenBy(entry => entry.SchemaId, StringComparer.Ordinal) .Select(entry => durable?.CatalogBytes(entry) ?? SerializeCatalogEntry(entry)) .ToArray(); var format = HeaderForWrite( durable?.Snapshot.FormatVersion, upserts.Select(staged => staged.HoldsIds).ToArray(), copiesAny: copied.Length > 0, copiedHoldIds: CopiedHoldIds(durable, copied, upserts), directoryStore: false, holdsVariants: records.Any(record => record.Variant != null)); var versions = encoded.Select(pair => (pair.Staged, VersionOfBytes(pair.Bytes))).ToArray(); Directory.CreateDirectory(FileInfo.DirectoryName!); WriteSnapshotAtomically(FileInfo.FullName, SerializeStore(format, catalog, fileRecords)); foreach (var key in removed) _lastRead.Remove(key); foreach (var (staged, version) in versions) _lastRead[staged.Record.Key] = version; return versions; } // Whether a record this write copies forward holds an element id, decided from the bytes the file holds under the store's // lock, so every C# writer with a type for the record decides the same from the same bytes. Positive-only: it runs only // while the header on disk is unmarked and can only mark the store, so a narrower type that decodes no ids changes nothing. // It never refuses a write: a record that does not resolve or decode is copied as its bytes and decides nothing. Gates, in // order, so the decode runs only where it can change the answer. private bool CopiedHoldIds(Durable? durable, CultPersistedRecord[] copied, IReadOnlyCollection upserts) { if (durable == null || string.Equals(durable.Snapshot.FormatVersion, CultPersistedStoreSnapshot.FormatV3, StringComparison.Ordinal) || upserts.Any(staged => staged.HoldsIds == true) || AttachedRegistry is not { } registry) return false; var catalog = durable.Snapshot.SchemaCatalog; foreach (var record in copied) { if (record.Variant != null) continue; try { if (registry.Claimant(record.SchemaId, catalog, out _) == null) continue; var type = registry.ResolvePersistedSchemaDetailed(record.SchemaId, catalog).Descriptor.DocumentType; if (!CultElementIds.CanHoldIds(type)) continue; CopiedRecordsDecoded++; if (CultElementIds.Holds(type, DeserializePayload(type, record.Payload))) return true; } catch (Exception e) when (e is not OperationCanceledException) { // Copied as its bytes: no type, or no decode, is no fact. } } return false; } private void RefuseWriteIntoMovedVariantStore(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)) return; var changed = new HashSet(StringComparer.Ordinal); var versions = records.ToDictionary(record => record.Key, record => durable!.Version(record.Key), StringComparer.Ordinal); foreach (var (key, version) in versions) { if (!_lastRead.TryGetValue(key, out var read) || !SameStoredIdentity(read, version)) changed.Add(key); } var durableKeys = records.Select(record => record.Key).ToHashSet(StringComparer.Ordinal); foreach (var key in _lastRead.Keys) { if (!durableKeys.Contains(key)) changed.Add(key); } if (changed.Count == 0) return; var recordKeys = _staged.Concat(upserts.Select(staged => staged.Record.Key)).Concat(removals) .Distinct(StringComparer.Ordinal).OrderBy(key => key, StringComparer.Ordinal).ToArray(); var changedKeys = changed.OrderBy(key => key, StringComparer.Ordinal).ToArray(); Refuse( recordKeys.Concat(changedKeys), key => Cache == null ? null : DurableAt(durable, key), key => ForeignAt(durable, key), key => { _staged.Remove(key); IsDirty = _staged.Count > 0; if (versions.TryGetValue(key, out var version)) _lastRead[key] = version; else _lastRead.Remove(key); }, 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. " + $"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. " + $"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." : "Reloading them failed, so the cache is unchanged."), recordKeys, changedKeys, inner)); } // A writable store judges a removal of a key its cache does not hold at its next flush or commit, under the lock (StageChecks); // a read-only store has neither, so it answers from its view and never creates the lock file. protected internal override void RefuseRemovalOfUndeclared(IReadOnlyCollection keys) { if (IsReadOnly) base.RefuseRemovalOfUndeclared(keys); else StageChecks(keys); } // A read-only store answers a batch's checks from its view and never creates the lock file; a writable one reads the file // under its lock. protected internal override CultCommitOutcome JudgeChecks(IReadOnlyCollection keys, bool wait) { if (IsReadOnly) return base.JudgeChecks(keys, wait); return Held(() => { using var fileLock = AcquireLock(wait); if (fileLock == null) return CultCommitOutcome.Contended; RefuseUndeclaredOnDisk(ReadSnapshot(), keys, Registry.Declares); return CultCommitOutcome.Committed; }); } // A cache declares through its registry. A cacheless CultMesh write declares the ids of the catalog entry it writes: its own id // and the ids it lists compatible. The keys judged are every key staged, written, removed or checked. private void RefuseUndeclaredOnDisk(Durable? durable, IReadOnlyCollection upserts, IReadOnlyCollection removals, IEnumerable checks) => RefuseUndeclaredOnDisk( durable, _staged.Concat(upserts.Select(staged => staged.Record.Key)).Concat(removals).Concat(checks).Concat(StagedChecks), Cache != null ? Registry.Declares : id => upserts.Any(staged => string.Equals(staged.Entry.SchemaId, id, StringComparison.Ordinal) || staged.Entry.CompatibleSchemaIds.Contains(id, StringComparer.Ordinal))); // A refused key's durable version is what this store has now read of it. private void RefuseUndeclaredOnDisk(Durable? durable, IEnumerable written, Func declares) { var undeclared = UndeclaredOnDisk(durable, declares); if (undeclared.Count == 0) return; RefuseUndeclaredOnDisk( written, key => undeclared.TryGetValue(key, out var carried) ? carried : null, key => Cache == null ? null : DurableAt(durable, key), key => ForeignAt(durable, key), key => { _staged.Remove(key); IsDirty = _staged.Count > 0; _lastRead[key] = durable!.Version(key); }, cacheless: Cache == null); } // Every durable record under an id the writer does not declare, by key, named by the catalog entry that publishes its id. private static Dictionary UndeclaredOnDisk(Durable? durable, Func declares) { var undeclared = new Dictionary(StringComparer.Ordinal); foreach (var record in durable?.Snapshot.Records ?? Array.Empty()) { if (Undeclared(record, durable!.Snapshot.SchemaCatalog, declares) is { } carried) undeclared[record.Key] = carried; } return undeclared; } // The durable record at a key as a load lists it when no registered type claims its id; null otherwise. private CultForeignRecord? ForeignAt(Durable? durable, string key) { if (durable?.Snapshot.Records.FirstOrDefault(candidate => string.Equals(candidate.Key, key, StringComparison.Ordinal)) is not { } record || Foreign(record, durable.Snapshot.SchemaCatalog, durable.Version(key)) is not { } carried) return null; return carried; } // The durable record at a key decoded as a load decodes it, at its durable version: null when the store holds none there or // holds it foreign. The one decode both refusals reload through. private CultStoredDocument? DurableAt(Durable? durable, string key) { var record = durable?.Snapshot.Records.FirstOrDefault(candidate => string.Equals(candidate.Key, key, StringComparison.Ordinal)); if (record == null || Foreign(record, durable!.Snapshot.SchemaCatalog, durable.Version(key)) != null) return null; return ToStoredDocument(record, durable.Version(key), durable.Snapshot.SchemaCatalog, DeserializePayload, out _); } // The file now holds every record written, under the id and at the storedAt the write gave it, and the version of those bytes: // the store's entries and the cache's say so. A copied record is on disk as the file had it, so nothing learns anything new about it: in particular the // ids it minted at load are still in memory only. private void Wrote(IEnumerable<(CultStagedRecord Staged, string Version)> written) { var staged = written.ToArray(); foreach (var (write, version) in staged) { if (!Entries.TryGetValue(write.Record.Key, out var entry)) continue; entry.StoredSchemaId = write.Record.SchemaId; entry.StoredAt = write.Record.StoredAt; entry.StoredVersion = version; } Cache?.Wrote(staged.Select(write => (write.Staged.Record, write.Version))); } // Open, flush and commit read the file through this one call, so they agree on which files are stores. A file that is // gone is an empty store: nothing to read, nothing to lose. A zero-byte file is not a store; it is refused. private Durable? ReadSnapshot() { var bytes = ReadDisk(); if (bytes == null) return null; try { return new Durable(this, ReadStore(bytes)); } catch (CultStoreUnreadableException ex) when (ex.Path == null) { throw new CultStoreUnreadableException(ex.Message, FileInfo.FullName, ex.InnerException); } } /// /// Reads the store file once, under the store's gate, with the reader a cache's store opens with, and returns what is /// durable at the path: the records, the schema catalog and the format stamp. Returns null when nothing is at the path. /// CultMesh's single-file document reads are built on this, so they share the store's one reader and its stamp rule. /// /// The file is empty, truncated, or stamped with a format this runtime does not read. public CultPersistedStoreSnapshot? ReadStoreSnapshot() => Held(() => ReadSnapshot()?.Snapshot); private byte[]? ReadDisk() { FileInfo.Refresh(); // Only nothing at the path is an empty store: a path with nothing at it reports no attributes at all (-1). A link whose // target is gone is a store that cannot be reached, and a directory is not a store file. var attributes = FileInfo.Attributes; if ((int)attributes == -1) return null; if ((attributes & FileAttributes.Directory) != 0 && (attributes & FileAttributes.ReparsePoint) == 0) throw new IOException("The store path is a directory, not a store file."); try { return ReadAllBytesShared(FileInfo.FullName); } catch (FileNotFoundException ex) when (PathIsALink()) { throw new IOException("The store path is a link whose target cannot be reached.", ex); } // A Windows junction is a directory, not a file, so opening one that dangles fails with an access error, not a missing file. catch (UnauthorizedAccessException ex) when (PathIsALink() && (FileInfo.Attributes & FileAttributes.Directory) != 0) { throw new IOException("The store path is a link whose target cannot be reached.", ex); } catch (FileNotFoundException) { return null; } } // Something is at the path even when the path cannot be opened: a link keeps the reparse-point attribute whether or not its // target is there. A path with nothing at it reports no attributes at all (-1). private bool PathIsALink() { FileInfo.Refresh(); var attributes = FileInfo.Attributes; return (int)attributes != -1 && (attributes & FileAttributes.ReparsePoint) != 0; } // The lock is a sidecar opened exclusively, which excludes other handles in this process and in others alike. private FileStream? AcquireLock(bool wait) { Directory.CreateDirectory(FileInfo.DirectoryName!); var lockPath = FileInfo.FullName + ".lock"; var started = Stopwatch.StartNew(); while (true) { try { return new FileStream(lockPath, FileMode.OpenOrCreate, FileAccess.ReadWrite, FileShare.None, bufferSize: 1); } catch (IOException) when (!wait) { return null; } catch (IOException) when (started.Elapsed < TimeSpan.FromSeconds(30)) { Thread.Sleep(10); } } } private static void WriteSnapshotAtomically(string path, byte[] payload) { var directory = Path.GetDirectoryName(path)!; var tempPath = Path.Combine(directory, $".{Path.GetFileName(path)}.{Guid.NewGuid():N}.tmp"); try { using (var stream = new FileStream( tempPath, FileMode.CreateNew, FileAccess.Write, FileShare.None, bufferSize: 4096, FileOptions.WriteThrough)) { stream.Write(payload, 0, payload.Length); stream.Flush(true); } if (File.Exists(path)) { File.Replace(tempPath, path, null); } else { File.Move(tempPath, path); } } finally { if (File.Exists(tempPath)) { File.Delete(tempPath); } } } private static byte[] ReadAllBytesShared(string path) { using var stream = new FileStream(path, FileMode.Open, FileAccess.Read, FileShare.ReadWrite | FileShare.Delete); using var buffer = new MemoryStream(); stream.CopyTo(buffer); return buffer.ToArray(); } } }