CultLib/src/GameCult.Caching/CultCache.cs
Meta Vens c246e0c5ae Single-file store: lay back an unowned record only while the file still holds it as loaded
A whole-view write laid back the loaded bytes of an unclaimed record over whatever the file held,
reverting its owner's newer write and moving storedAt backwards. It now lays back only while the
file holds the loaded (schemaId, storedAt); a newer record is carried as the file has it and a
removed one is not written back. The v3 header is decided by records that are written as the cache
holds them, so a laid-back id-less record no longer marks the store. A document changed in place is
written as a write of the record, not dropped. A refusal drops the store's copy of the refused
records, so it no longer refuses every later write.

Does not build with the directory store untouched? It builds; the directory twin follows.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-02 16:51:54 +02:00

4067 lines
195 KiB
C#

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<string>();
public CultSchemaMemberCatalogEntry[] Members { get; set; } = Array.Empty<CultSchemaMemberCatalogEntry>();
// The catalog a write leaves, derived from the records being written and nothing else. For each schema id a record carries, one
// entry that publishes it: the entry that owns the id (a registered descriptor, else one that arrived with the records), else
// one that lists it as compatible (registered first). 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<CultSchemaCatalogEntry> Canonical(IEnumerable<CultSchemaCatalogEntry> 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<CultPersistedRecord> records,
IReadOnlyCollection<CultSchemaCatalogEntry> registered,
IReadOnlyCollection<CultSchemaCatalogEntry> arrived)
{
var chosen = new Dictionary<string, (CultSchemaCatalogEntry Entry, bool Registered)>(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(registered.Where(entry => entry.CompatibleSchemaIds.Contains(schemaId, StringComparer.Ordinal))).FirstOrDefault();
isRegistered = pick != null;
pick ??= Canonical(arrived.Where(entry => entry.CompatibleSchemaIds.Contains(schemaId, StringComparer.Ordinal))).FirstOrDefault();
}
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<byte>();
// Non-null for a variant record (slot 4): its base and overrides. Its payload is empty.
public CultVariantDelta? Variant { get; set; }
}
/// <summary>
/// A record a store carries untouched because no registered type owns its schema id or lists it as compatible, for example a
/// record of a type this build does not have, or one renamed 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
/// <see cref="CultSchemaConflictException"/>. The cache never holds it. Declaring its schema id on a type claims it at the next load.
/// </summary>
public sealed class CultForeignRecord
{
internal CultForeignRecord(string key, string schemaId, string schemaName, string storedAt)
{
Key = key;
SchemaId = schemaId;
SchemaName = schemaName;
StoredAt = storedAt;
}
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; }
}
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<CultSchemaCatalogEntry>();
public CultPersistedRecord[] Records { get; set; } = Array.Empty<CultPersistedRecord>();
}
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<int>();
public int[] DefaultedMissingSlots { get; set; } = Array.Empty<int>();
public CultSchemaMigrationWarning[] Warnings { get; set; } = Array.Empty<CultSchemaMigrationWarning>();
}
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<T>'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<string> compatibleSchemaIds,
string contentHash,
string canonicalSchemaJson,
bool isGlobal,
string? nameMember,
Func<object, string?>? nameAccessor,
IReadOnlyDictionary<string, Func<object, string>> indexAccessors,
IReadOnlyList<CultDocumentMemberDescriptor> members,
IReadOnlyList<CultDocumentRegistry.PersistedMember> 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; }
/// <summary>The ids besides <see cref="SchemaId"/> this type declared through <see cref="CultDocumentAttribute.CompatibleSchemaIds"/>.</summary>
public IReadOnlyList<string> CompatibleSchemaIds { get; }
public string ContentHash { get; }
public string CanonicalSchemaJson { get; }
public bool IsGlobal { get; }
public string? NameMember { get; }
internal Func<object, string?>? NameAccessor { get; }
internal IReadOnlyDictionary<string, Func<object, string>> IndexAccessors { get; }
internal IReadOnlyList<CultDocumentMemberDescriptor> Members { get; }
internal IReadOnlyList<CultDocumentRegistry.PersistedMember> 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<CultDocumentMemberView> 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<T>, one per element of
// an IEnumerable<CultRecordRef<T>>, one per key of an IDictionary<CultRecordRef<T>, 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;
_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;
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. With StoredSchemaId it names the record's stored bytes: a write that stores the record
// differently mints it again, including a whole-view write that re-encodes 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. With StoredAt it is the record's
// identity on disk, and a commit condition compares the pair with the store's bytes exactly.
public string StoredSchemaId { get; internal set; }
// 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(_document);
if (_codec == null)
return true;
return Variant.Overrides.Any(entry => CultElementIds.Holds(_codec.Deserialize(
Descriptor.RichMembers.First(member => member.Slot == entry.Path[0].Slot).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 };
}
public sealed class CultDocumentRegistry
{
private static readonly Lazy<CultDocumentRegistry> 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<Type> 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<CultDocumentDescriptor> 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<CultDocumentDescriptor> 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<CultDocumentAttribute>();
if (attribute == null)
{
throw new InvalidOperationException(
$"Type {type.FullName} is not marked with {nameof(CultDocumentAttribute)}.");
}
descriptor = BuildDescriptor(type);
return RegisterDescriptor(descriptor);
}
public CultDocumentDescriptor GetRequired<T>() 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<CultSchemaCatalogEntry> catalog)
{
return ResolvePersistedSchemaDetailed(schemaId, catalog).Descriptor;
}
public CultSchemaMigrationReport ResolvePersistedSchemaReport(string schemaId, IReadOnlyCollection<CultSchemaCatalogEntry> catalog)
{
return ResolvePersistedSchemaDetailed(schemaId, catalog).Report;
}
internal CultSchemaResolutionResult ResolvePersistedSchemaDetailed(string schemaId, IReadOnlyCollection<CultSchemaCatalogEntry> 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.
internal CultDocumentDescriptor? Claimant(
string schemaId,
IReadOnlyCollection<CultSchemaCatalogEntry> catalog,
out CultSchemaCatalogEntry? persisted)
{
var indexes = _indexes;
persisted = null;
if (indexes.BySchemaId.TryGetValue(schemaId, out var exact))
return exact;
// A schema is published by its id or as a compatible id.
persisted = catalog.FirstOrDefault(entry => string.Equals(entry.SchemaId, schemaId, StringComparison.Ordinal))
?? catalog.FirstOrDefault(entry => entry.CompatibleSchemaIds.Contains(schemaId, StringComparer.Ordinal));
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;
}
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<Type, CultDocumentDescriptor>();
BySchemaId = new Dictionary<string, CultDocumentDescriptor>(StringComparer.Ordinal);
ByCompatibleSchemaId = new Dictionary<string, CultDocumentDescriptor[]>(StringComparer.Ordinal);
BySchemaName = new Dictionary<string, CultDocumentDescriptor[]>(StringComparer.Ordinal);
}
public RegistryIndexes(RegistryIndexes source)
{
ByType = new Dictionary<Type, CultDocumentDescriptor>(source.ByType);
BySchemaId = new Dictionary<string, CultDocumentDescriptor>(source.BySchemaId, StringComparer.Ordinal);
ByCompatibleSchemaId = new Dictionary<string, CultDocumentDescriptor[]>(source.ByCompatibleSchemaId, StringComparer.Ordinal);
BySchemaName = new Dictionary<string, CultDocumentDescriptor[]>(source.BySchemaName, StringComparer.Ordinal);
}
public Dictionary<Type, CultDocumentDescriptor> ByType { get; }
public Dictionary<string, CultDocumentDescriptor> BySchemaId { get; }
// Every type declaring the id compatible, in registration order; resolution never depends on that order.
public Dictionary<string, CultDocumentDescriptor[]> ByCompatibleSchemaId { get; }
public Dictionary<string, CultDocumentDescriptor[]> BySchemaName { get; }
}
private static CultDocumentDescriptor BuildDescriptor(Type type)
{
var attribute = type.GetCustomAttribute<CultDocumentAttribute>()
?? 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<CultGlobalAttribute>() != null,
nameMember?.Member.Name,
nameMember?.GetterNullable,
indexAccessors,
descriptorMembers,
members);
}
private static string BuildSemanticFingerprint(
string schemaName,
string schemaVersion,
IReadOnlyList<CultDocumentMemberDescriptor> 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<CultDocumentMemberDescriptor> 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<CultSchemaMigrationWarning>();
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<string>();
var warnings = new List<CultSchemaMigrationWarning>();
var ignoredExtraSlots = new List<int>();
var defaultedMissingSlots = new List<int>();
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}'");
}
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 (!string.Equals(localMember.TargetSchemaName, persistedMember.TargetSchemaName, StringComparison.Ordinal))
{
errors.Add($"slot {localMember.Slot} changed target schema '{persistedMember.TargetSchemaName ?? "<none>"}' -> '{localMember.TargetSchemaName ?? "<none>"}'");
}
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 ?? "<none>"}' -> '{localMember.IndexAlias ?? "<none>"}'");
}
}
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<PersistedMember> DiscoverMembers(Type type)
{
const BindingFlags declared = BindingFlags.DeclaredOnly | BindingFlags.Instance | BindingFlags.Public;
var messagePackObject = type.GetCustomAttribute<MessagePackObjectAttribute>(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<string>();
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<MemberInfo>().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<T> type)
// whose CLR shape the cache cannot walk. many:false needs CultRecordRef<T> itself; many:true
// needs IEnumerable<CultRecordRef<T>> or IDictionary<CultRecordRef<T>, 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<CultReferenceAttribute>();
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<PropertyInfo> 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<int, Type>();
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<CultIndexAttribute>();
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<Type> 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<T> with a value produces a boxed T (never a boxed Nullable<T>), 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<T>, many:true needs IEnumerable<CultRecordRef<T>> or IDictionary<CultRecordRef<T>, 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<global::MessagePack.IgnoreMemberAttribute>() != null;
private static object? GetKeyValue(MemberInfo member)
{
var key = member.GetCustomAttribute<global::MessagePack.KeyAttribute>();
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<object, string> Getter { get; set; } = default!;
public Func<object, string?> GetterNullable { get; set; } = default!;
// D8/Q-J (revised 2026-09-22): non-null exactly for the closed CLR numeric set (Nullable<T>
// 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<object, string?> 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<object, IEnumerable<(CultRecordKey Target, object? Payload)>>? ReferenceEnumerator { get; set; }
public static PersistedMember FromMember(
MemberInfo member,
Type memberType,
Func<object, object?> getValue,
int slot)
{
var referenceAttribute = member.GetCustomAttribute<CultReferenceAttribute>();
var targetType = ResolveReferenceTarget(memberType, referenceAttribute?.TargetType);
var targetSchemaName = targetType?.GetCustomAttribute<CultDocumentAttribute>()?.SchemaName;
var isReference = targetType != null || referenceAttribute != null;
var many = referenceAttribute?.Many ?? false;
return new PersistedMember
{
Member = member,
MemberType = memberType,
Slot = slot,
IsName = member.GetCustomAttribute<CultNameAttribute>() != 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<object, IEnumerable<(CultRecordKey Target, object? Payload)>> BuildReferenceEnumerator(
Type memberType, bool many, Func<object, object?> 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<CultReferenceAttribute>();
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<T> element's T - the
// same narrowing a scalar CultRecordRef<T> 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<CultStoredDocument> upserts,
IReadOnlyList<CultStoredDocument> deletes,
IReadOnlyList<(CultRecordKey Key, string? SchemaId, string? StoredAt)> expected,
bool expectUnchanged)
{
Upserts = upserts;
Deletes = deletes;
Expected = expected;
ExpectUnchanged = expectUnchanged;
}
public IReadOnlyList<CultStoredDocument> Upserts { get; }
public IReadOnlyList<CultStoredDocument> Deletes { get; }
public IReadOnlyList<(CultRecordKey Key, string? SchemaId, string? StoredAt)> Expected { get; }
public bool ExpectUnchanged { get; }
public bool HasConditions => Expected.Count > 0 || ExpectUnchanged;
// durable is what the store holds on disk now; observed is what its cache last loaded or committed.
// Identity is (stored schema id, storedAt), compared exactly. It names the stored bytes: every write that stores a record
// differently mints it a later storedAt, including a whole-view write that re-encodes a record it loaded.
public bool ConditionsHold(IReadOnlyCollection<CultPersistedRecord> durable, IEnumerable<CultStoredDocument> observed)
{
var byKey = durable.ToDictionary(record => record.Key, StringComparer.Ordinal);
foreach (var (key, schemaId, storedAt) in Expected)
{
byKey.TryGetValue(key.Value, out var record);
var holds = schemaId == null
? record == null
: record != null && record.SchemaId == schemaId && record.StoredAt == storedAt;
if (!holds)
return false;
}
return !ExpectUnchanged ||
durable.Select(record => Identity(record.Key, record.SchemaId, record.StoredAt))
.OrderBy(identity => identity, StringComparer.Ordinal)
.SequenceEqual(observed
.Select(stored => Identity(stored.Key.Value, stored.StoredSchemaId, stored.StoredAt))
.OrderBy(identity => identity, StringComparer.Ordinal));
}
private static string Identity(string key, string schemaId, string storedAt) => $"{key}\n{schemaId}\n{storedAt}";
}
// One staged write: a plain record (Stored), a variant (Variant), a flatten, or, with none of those, a removal.
internal sealed class StagedOperation
{
public StagedOperation(CultRecordKey key, CultStoredDocument? stored, CultVariantDelta? variant, bool flatten)
{
Key = key;
Stored = stored;
Variant = variant;
Flatten = flatten;
}
public CultRecordKey Key { get; }
public CultStoredDocument? Stored { get; }
public CultVariantDelta? Variant { get; }
public bool Flatten { get; }
public bool IsRemoval => Stored == null && Variant == null && !Flatten;
}
// An explicit value: nothing staged here is visible to anyone until Commit returns.
public sealed class CultCacheBatch
{
private readonly CultCache _cache;
internal readonly Dictionary<string, StagedOperation> Operations = new(StringComparer.Ordinal);
internal readonly List<(CultRecordKey Key, string? SchemaId, string? StoredAt)> Expected = new();
internal bool ExpectsUnchanged;
internal bool Sealed;
internal CultCacheBatch(CultCache cache)
{
_cache = cache;
}
public CultRecordHandle<T> Upsert<T>(T document, CultRecordHandle<T>? handle = null)
{
if (document == null) throw new ArgumentNullException(nameof(document));
return new CultRecordHandle<T>(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<CultVariantOverride>? overrides = null)
{
ThrowIfSealed();
Operations[key.Value] = new StagedOperation(
key, null, new CultVariantDelta(baseKey.Value, (overrides ?? Array.Empty<CultVariantOverride>()).ToArray()), false);
return key;
}
// Replaces a variant with a plain record holding its resolved document: the explicit, separate operation (Q1a).
public void Flatten(CultRecordKey key)
{
ThrowIfSealed();
Operations[key.Value] = new StagedOperation(key, null, null, true);
}
public void Expect(CultRecordKey key, object? current)
{
ThrowIfSealed();
var observed = _cache.Observe(key, current);
Expected.Add((key, observed?.StoredSchemaId, observed?.StoredAt));
}
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<string, CultStoredDocument> _entries = new(StringComparer.Ordinal);
private readonly Dictionary<Type, Dictionary<string, string>> _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<string, List<string>>> _indexes = new();
private readonly Dictionary<Type, string> _globals = new();
private readonly CultCodec? _codec;
// Keys holding a variant: derived from the entries by Apply, never decided anywhere else.
private readonly HashSet<string> _variantKeys = new(StringComparer.Ordinal);
private readonly ConditionalWeakTable<object, KeyBox> _handles = new();
// 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<Change>[] _observers = Array.Empty<Observer<Change>>();
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<Action<IReadOnlyList<CultCacheDocumentChange<object>>>> _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;
// Read under the gate, by a store judging a merge.
internal bool HoldsVariants => _variantKeys.Count > 0;
// A store wrote its whole view: every record it wrote is stored under its type's own id at the storedAt the write gave it,
// 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 WroteWholeView(IEnumerable<CultPersistedRecord> written) => Held(() =>
{
foreach (var record in written)
{
if (!_entries.TryGetValue(record.Key, out var entry))
continue;
entry.StoredSchemaId = record.SchemaId;
entry.StoredAt = record.StoredAt;
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<T>(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.
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<object?, object?>? OnUpdate;
public IReadOnlyList<CacheBackingStore> BackingStores
{
get
{
lock (_gate)
return _stores.Select(entry => entry.Store).ToArray();
}
}
public IEnumerable<object> AllEntries
{
get
{
lock (_gate)
return _entries.Values.Select(entry => entry.Document).ToArray();
}
}
public IEnumerable<CultStoredDocument> AllStoredDocuments
{
get
{
lock (_gate)
return _entries.Values
.OrderBy(entry => entry.Descriptor.SchemaName, StringComparer.Ordinal)
.ThenBy(entry => entry.Key.Value, StringComparer.Ordinal)
.ToArray();
}
}
public Observable<CultCacheDocumentChange<T>> Watch<T>() where T : class
{
return Observable.Create<Change>(Register)
.Where(change => typeof(T).IsAssignableFrom(change.Stored.Descriptor.DocumentType))
.Select(change => new CultCacheDocumentChange<T>(
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<IReadOnlyList<CultCacheDocumentChange<object>>> _journal;
private bool _removed;
public JournalRegistration(CultCache cache, Action<IReadOnlyList<CultCacheDocumentChange<object>>> journal)
{
_cache = cache;
_journal = journal;
}
public void Dispose()
{
if (_removed)
return;
_cache.Held(() => _cache._journals.Remove(_journal));
_removed = true;
}
}
private IDisposable Register(Observer<Change> 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<CultCacheDocumentChange<T>> WatchRecord<T>(CultRecordKey key) where T : class
{
return Watch<T>().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<IReadOnlyList<CultCacheDocumentChange<object>>> 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<Type>();
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);
store.Judging = (arriving, departing) => { Resolve(arriving, departing, store); };
_stores.Add(candidate);
try
{
store.PullAll();
}
catch
{
_stores.Remove(candidate);
store.Loaded = null;
store.Judging = 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<Exception>();
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.PushAll();
}
return true;
});
public Task FlushAsync()
{
FlushAllBackingStores();
return Task.CompletedTask;
}
public Task<CultRecordHandle<T>> AddAsync<T>(T document, CultRecordHandle<T>? handle = null)
{
return UpsertAsync(document, handle);
}
public Task<CultRecordHandle<T>> UpsertAsync<T>(T document, CultRecordHandle<T>? handle = null)
{
if (document == null) throw new ArgumentNullException(nameof(document));
return Task.FromResult(new CultRecordHandle<T>(Write(document, handle?.Key)));
}
public Task<CultRecordKey> 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.
public Task<CultRecordKey> UpsertVariantAsync(CultRecordKey key, CultRecordKey baseKey, IEnumerable<CultVariantOverride>? overrides = null)
{
var delta = new CultVariantDelta(baseKey.Value, (overrides ?? Array.Empty<CultVariantOverride>()).ToArray());
return Task.FromResult(Held(() =>
{
var stored = StampVariant(key, DescriptorOfBase(key, delta.BaseKey, null), delta);
Admit(new[] { stored }, Array.Empty<CultStoredDocument>(), null, (home, docs) =>
{
home?.Push(docs[0]);
return CultCommitOutcome.Committed;
});
return key;
}));
}
// Replaces a variant with a plain record holding its resolved document (Q1a's explicit operation).
public Task<CultRecordKey> FlattenAsync(CultRecordKey key) => Task.FromResult(Held(() =>
{
var stored = StampFlatten(key);
Admit(new[] { stored }, Array.Empty<CultStoredDocument>(), null, (home, docs) =>
{
home?.Push(docs[0]);
return CultCommitOutcome.Committed;
});
return key;
}));
public bool Remove(CultRecordKey key) => Held(() =>
{
if (!_entries.TryGetValue(key.Value, out var existing))
return false;
Admit(Array.Empty<CultStoredDocument>(), new[] { existing }, null, (home, _) =>
{
home?.Delete(existing);
return CultCommitOutcome.Committed;
});
return true;
});
public void Remove<T>(CultRecordHandle<T> handle)
{
Remove(handle.Key);
}
// false: a condition failed; nothing was written, changed in memory, or published.
public bool Commit(Action<CultCacheBatch> stage)
{
return Land(stage, wait: true) == CultCommitOutcome.Committed;
}
public CultCommitOutcome TryCommit(Action<CultCacheBatch> stage)
{
return Land(stage, wait: false);
}
public CultRecordHandle<T>? TryGetHandle<T>(T document)
{
if (document == null) throw new ArgumentNullException(nameof(document));
lock (_gate)
return _handles.TryGetValue(document, out var box) ? new CultRecordHandle<T>(box.Key) : null;
}
// The stored form of a record: for a variant, the delta beside the resolved document.
public CultStoredDocument? GetStored(CultRecordKey key)
{
lock (_gate)
return _entries.TryGetValue(key.Value, out var stored) ? stored : null;
}
public object? Get(CultRecordKey key)
{
lock (_gate)
return _entries.TryGetValue(key.Value, out var stored) ? stored.Document : null;
}
public T? Get<T>(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<T>(CultRecordKey key, out T? document) where T : class
{
document = Get<T>(key);
return document != null;
}
public IEnumerable<T> GetAll<T>() where T : class
{
lock (_gate)
return _entries.Values.Select(entry => entry.Document).OfType<T>().ToArray();
}
public T? GetGlobal<T>() where T : class
{
return Single<T>("global", () => _globals
.Where(pair => typeof(T).IsAssignableFrom(pair.Key))
.Select(pair => pair.Value));
}
public T? GetByName<T>(string name) where T : class
{
return Single<T>($"name '{name}'", () => _names
.Where(pair => typeof(T).IsAssignableFrom(pair.Key))
.SelectMany(pair => pair.Value.TryGetValue(name, out var key) ? new[] { key } : Array.Empty<string>()));
}
// A value held by one record finds it. Records of one type sharing a value (plain duplicates) find the holder with the
// smallest key (ordinal), whatever order they were written or loaded in: the choice is a function of the held set, so
// it is the same live, after a flush and after a reopen, and removing that holder finds the next.
public T? GetByIndex<T>(string alias, string value) where T : class
{
return Single<T>($"index {alias}='{value}'", () => _indexes
.Where(pair => pair.Key.Alias == alias && typeof(T).IsAssignableFrom(pair.Key.Type))
.SelectMany(pair => pair.Value.TryGetValue(value, out var keys)
? new[] { keys.Aggregate((best, next) => string.CompareOrdinal(next, best) < 0 ? next : best) }
: Array.Empty<string>()));
}
public void Dispose()
{
Held(() =>
{
if (FlushAttachedStoresOnDispose && IsDirty)
FlushAllBackingStores();
return true;
});
foreach (var store in BackingStores)
store.Dispose();
Observer<Change>[] completing;
lock (_observerGate)
{
completing = _observers;
_observers = Array.Empty<Observer<Change>>();
_observersCompleted = true;
}
var failures = new List<Exception>();
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)), descriptor, document);
// An unresolved variant: the admitting hold resolves it against its base before anything lands.
private CultStoredDocument StampVariant(CultRecordKey key, CultDocumentDescriptor descriptor, CultVariantDelta delta) =>
new(key, MintStoredAt(PreviousStoredAt(key)), descriptor, delta, null, _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), existing.Descriptor, existing.Document) { Flattens = true };
}
private string? PreviousStoredAt(CultRecordKey key) => _entries.TryGetValue(key.Value, out var previous) ? previous.StoredAt : null;
// The concrete type a variant takes from its base, walking bases through what the batch stages and then the cache.
// Refuses a cycle or a base that is not there, naming the keys, exactly as the admitting hold would.
private CultDocumentDescriptor DescriptorOfBase(CultRecordKey variant, string baseKey, CultCacheBatch? batch)
{
var seen = new List<string> { variant.Value };
var next = baseKey;
while (true)
{
if (seen.Contains(next))
throw CycleRefusal(seen.Append(next));
var staged = batch != null && batch.Operations.TryGetValue(next, out var op) ? op : null;
if (staged != null)
{
if (staged.Stored != null)
return staged.Stored.Descriptor;
if (staged.Variant != null)
{
seen.Add(next);
next = staged.Variant.BaseKey;
continue;
}
if (!staged.Flatten)
throw MissingBaseRefusal(variant.Value, baseKey, removed: true);
}
if (!_entries.TryGetValue(next, out var existing))
throw MissingBaseRefusal(variant.Value, baseKey, removed: false);
return existing.Descriptor;
}
}
internal CultStoredDocument? Observe(CultRecordKey key, object? current)
{
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));
}
internal static string MintStoredAt(string? previous)
{
var now = DateTimeOffset.UtcNow;
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<CultStoredDocument>(), null, (home, docs) =>
{
home?.Push(docs[0]);
return CultCommitOutcome.Committed;
});
return stored.Key;
});
private CultCommitOutcome Land(Action<CultCacheBatch> 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();
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) =>
{
// What the store holds is what the hold resolved: a variant's delta beside its resolved view.
var request = new CultCommitRequest(resolved, deletes, batch.Expected.ToArray(), batch.ExpectsUnchanged);
if (home != null)
{
if (request.HasConditions && home.IsDirty)
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
.Select(entry => new CultPersistedRecord { Key = entry.Key.Value, SchemaId = entry.StoredSchemaId, StoredAt = entry.StoredAt })
.ToArray();
return request.ConditionsHold(inMemory, _entries.Values) ? CultCommitOutcome.Committed : CultCommitOutcome.Mismatch;
});
});
}
// 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<CultStoredDocument> admitted,
IReadOnlyList<CultStoredDocument> evicted,
CacheBackingStore? source,
Func<CacheBackingStore?, IReadOnlyList<CultStoredDocument>, CultCommitOutcome> land) => Held(() =>
{
if (_held == null)
throw new InvalidOperationException("An admission reached the cache under a plain lock on its gate; every admission runs in a hold.");
var 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.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 first exception (an AggregateException for several) is rethrown.
// If the body threw, its exception wins.
internal T Held<T>(Func<T> 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<Exception> Publish(Hold hold)
{
var failures = new List<Exception>();
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<Change> applied, List<Exception> failures)
{
if (_journals.Count == 0 || applied.Count == 0)
return;
var changes = applied
.Select(change => new CultCacheDocumentChange<object>(
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<CultStoredDocument> admitted,
IReadOnlyList<CultStoredDocument> evicted,
CacheBackingStore? source)
{
var homes = new List<CacheBackingStore>();
var globals = new Dictionary<Type, string>(_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}.");
homes.Add(home);
}
}
private sealed class VariantPlan
{
public VariantPlan(IReadOnlyList<CultStoredDocument> admitted)
{
Admitted = admitted;
}
// The admitted records, variants resolved.
public IReadOnlyList<CultStoredDocument> Admitted { get; }
// Variants this admission did not write whose base chain it changed: re-resolved, same delta, same storedAt.
public List<CultStoredDocument> Dependents { get; } = new();
// Name and index values of every record that lands (Admitted and Dependents), by key. Filled once by Project.
public Dictionary<string, Projection> 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<object, string?> 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<CultStoredDocument> admitted,
IReadOnlyList<CultStoredDocument> 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<CultStoredDocument> admitted,
IReadOnlyList<CultStoredDocument> 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<string>(incoming.Keys.Concat(gone), StringComparer.Ordinal);
// A flatten resolves like the variant it replaces, against the post-batch set, so a base edited in the same
// batch is in it; it lands as the plain record of that resolution.
var toResolve = new List<string>(admitted.Where(stored => stored.Variant != null || stored.Flattens).Select(stored => stored.Key.Value));
foreach (var variantKey in _variantKeys)
{
if (incoming.ContainsKey(variantKey) || gone.Contains(variantKey))
continue;
var next = _entries[variantKey].Variant!.BaseKey;
for (var steps = 0; steps <= _entries.Count; steps++)
{
if (changed.Contains(next))
{
toResolve.Add(variantKey);
break;
}
var link = Lookup(next);
if (link?.Variant == null)
break;
next = link.Variant.BaseKey;
}
}
if (toResolve.Count == 0)
return new VariantPlan(admitted);
var pending = new HashSet<string>(toResolve, StringComparer.Ordinal);
var done = new Dictionary<string, CultStoredDocument>(StringComparer.Ordinal);
var stack = new List<string>();
foreach (var key in toResolve)
ResolveOne(key);
var plan = new VariantPlan(admitted.Select(stored => done.TryGetValue(stored.Key.Value, out var resolved) ? resolved : stored).ToList());
foreach (var key in toResolve.Where(key => !incoming.ContainsKey(key)))
plan.Dependents.Add(done[key]);
return plan;
CultStoredDocument ResolveOne(string key)
{
if (done.TryGetValue(key, out var finished))
return finished;
var variant = Lookup(key)!;
if (stack.Contains(key))
throw CycleRefusal(stack.SkipWhile(entry => entry != key).Append(key));
stack.Add(key);
// A flatten carries no delta of its own; the variant it replaces does.
var isVariant = variant.Variant != null;
var delta = variant.Variant ?? _entries[key].Variant!;
var baseStored = Lookup(delta.BaseKey) ?? throw MissingBaseRefusal(key, delta.BaseKey, gone.Contains(delta.BaseKey));
if (pending.Contains(delta.BaseKey))
baseStored = ResolveOne(delta.BaseKey);
var descriptor = variant.Descriptor;
if (_codec == null)
throw new InvalidOperationException(
$"Variant {key} needs a codec to resolve; this cache has none. Create it with CultCacheMessagePack.Create.");
if (baseStored.Descriptor.DocumentType != descriptor.DocumentType)
throw new InvalidOperationException(
$"Variant {key} is a {descriptor.DocumentType.Name} but its base {delta.BaseKey} is a {baseStored.Descriptor.DocumentType.Name}; " +
"a variant has its base's concrete type.");
var type = descriptor.DocumentType;
var kept = new List<KeyValuePair<int, byte[]>>();
var overrides = new List<CultVariantOverride>(delta.Overrides.Count);
var minted = false;
var seen = new HashSet<int>();
foreach (var entry in delta.Overrides)
{
if (entry.Op != CultOverrideOp.Set || entry.Path.Count != 1 || entry.Path[0].ElementId.Length != 0 || entry.Id.Length != 0)
throw new InvalidOperationException(
$"Variant {key} carries {entry.Op} at {entry.PathText}; this runtime resolves top-level Set overrides only.");
var slot = entry.Path[0].Slot;
if (!seen.Add(slot))
throw new InvalidOperationException($"Variant {key} overrides slot {slot} more than once; a variant has one override per slot.");
var member = descriptor.RichMembers.FirstOrDefault(candidate => candidate.Slot == slot);
// Drift is the load's business (ToStoredDocument sheds an override of a slot the type lacks), so a delta
// reaching here names only slots the type has, unless someone is writing one it never had.
if (member == null)
throw new InvalidOperationException($"Variant {key} overrides slot {slot}, which {type.Name} does not have.");
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} 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(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<int, byte[]>(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
};
}
}
// 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<CultStoredDocument> evicted)
{
var landing = plan.Admitted.Concat(plan.Dependents).ToList();
if (landing.All(stored => stored.Descriptor.IndexAccessors.Count == 0))
return;
var replaced = new HashSet<string>(
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<string>>();
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<string>();
landedKeys.Add(stored.Key.Value);
}
}
var holdersOf = new Dictionary<((Type Type, string Alias) Index, string Value), (string[] All, string[] Variants)>();
foreach (var (stored, index, value) in landedValues)
{
if (!holdersOf.TryGetValue((index, value), out var known))
{
var all = (_indexes.TryGetValue(index, out var held) && held.TryGetValue(value, out var heldKeys)
? heldKeys.Where(heldKey => !replaced.Contains(heldKey))
: Enumerable.Empty<string>())
.Concat(landed[(index, value)])
.Distinct(StringComparer.Ordinal)
.ToArray();
holdersOf[(index, value)] = known = (all, all.Where(IsVariant).ToArray());
}
var self = stored.Key.Value;
var holder = known.All.FirstOrDefault(candidate => candidate != self);
if (stored.Variant != null && holder != null)
throw new InvalidOperationException(
$"Variant {self} and {holder} both hold index {index.Alias} '{value}'; a variant must override its indexed member.");
var variantHolder = known.Variants.FirstOrDefault(candidate => candidate != self);
if (stored.Variant == null && variantHolder != null)
throw new InvalidOperationException(
$"Record {self} and variant {variantHolder} both hold index {index.Alias} '{value}'; " +
"a variant may not share an indexed value with another record.");
}
}
private static InvalidOperationException CycleRefusal(IEnumerable<string> keys) =>
new($"Variant bases form a cycle: {string.Join(" -> ", keys)}.");
private static InvalidOperationException MissingBaseRefusal(string variant, string baseKey, bool removed) =>
new(removed
? $"Record {baseKey} cannot go: variant {variant} is based on it. Remove or rebase the variant in the same commit."
: $"Variant {variant} is based on {baseKey}, which does not exist.");
private List<Change> Apply(
VariantPlan plan,
IReadOnlyList<CultStoredDocument> evicted,
CacheBackingStore? source)
{
var admitted = plan.Admitted;
var changes = new List<Change>(admitted.Count + evicted.Count + plan.Dependents.Count);
foreach (var stored in evicted)
{
if (!_entries.TryGetValue(stored.Key.Value, out var existing))
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<T>(string lookup, Func<IEnumerable<string>> 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<string, List<string>>> indexes, (Type, string) index, string value, string key)
{
if (!indexes.TryGetValue(index, out var map))
indexes[index] = map = new Dictionary<string, List<string>>(StringComparer.Ordinal);
if (!map.TryGetValue(value, out var keys))
map[value] = keys = new List<string>();
keys.Remove(key);
keys.Add(key);
}
private static void RemoveHolder(Dictionary<string, List<string>> map, string key)
{
foreach (var pair in map.ToArray())
{
if (pair.Value.Remove(key) && pair.Value.Count == 0)
map.Remove(pair.Key);
}
}
private static Dictionary<string, string> MapOf<TKey>(Dictionary<TKey, Dictionary<string, string>> maps, TKey key)
where TKey : notnull
{
if (!maps.TryGetValue(key, out var map))
maps[key] = map = new Dictionary<string, string>(StringComparer.Ordinal);
return map;
}
private static void RemoveKey(Dictionary<string, string> 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<Exception> 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<CultSchemaMigrationReport>();
protected CacheBackingStore(bool readOnly = false)
{
IsReadOnly = readOnly;
}
protected CultDocumentRegistry Registry =>
_registry ?? throw new InvalidOperationException("Backing store is not attached to a CultDocumentRegistry.");
protected ConcurrentDictionary<string, CultStoredDocument> Entries { get; } =
new(StringComparer.Ordinal);
public bool IsReadOnly { get; }
public bool IsDirty { get; protected set; }
public bool FlushOnDispose { get; set; }
public IReadOnlyList<CultSchemaMigrationReport> LastSchemaMigrationReports => _lastSchemaMigrationReports;
/// <summary>
/// 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 <see cref="CultForeignRecord"/>.
/// </summary>
public IReadOnlyList<CultForeignRecord> ForeignRecords => _foreignRecords;
private CultForeignRecord[] _foreignRecords = Array.Empty<CultForeignRecord>();
// 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<CultStoredDocument>, IReadOnlyList<CultStoredDocument>>? Loaded;
// Set by the cache at attach: the cache's own resolution (variants, their indexed values, their bases) run over a
// set of records a store is about to hold, mutating nothing. It throws the refusal the cache would give the same
// set arriving as a load.
protected internal Action<IReadOnlyList<CultStoredDocument>, IReadOnlyList<CultStoredDocument>>? Judging;
private readonly object _detachedGate = new();
internal CultCache? Cache;
private const string DirectoryFormatV4 = "cultcache.store.v4.directory-content-addressed-pages";
private const string DirectoryFormatV5 = "cultcache.store.v5.directory-content-addressed-pages";
/// <summary>
/// The one header decision, for every writer of every store: the string a store file's format header must carry after
/// this write.
/// </summary>
/// <remarks>
/// 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
/// (<paramref name="wholeStore"/>) 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 (<paramref name="existingHeader"/>), because
/// it cannot see what the rest holds. <paramref name="holdsIds"/> is about the records this write knows are on disk.
/// Unmarked, a single-file store is v2 exactly when it holds a variant (<paramref name="holdsVariants"/>) 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.
/// </remarks>
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;
}
// 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<T>(Func<T> body)
{
if (Cache is { } cache)
return cache.Held(body);
lock (_detachedGate)
return body();
}
protected void Held(Action body) => Held(() => { body(); return true; });
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; conditions are evaluated against what is on disk then.
// A failure restores the store's staged view.
public abstract CultCommitOutcome CommitBatch(CultCommitRequest request, bool wait);
public virtual void Dispose()
{
if (FlushOnDispose && IsDirty && !IsReadOnly)
{
PushAll();
}
}
protected void ThrowIfReadOnly()
{
if (IsReadOnly)
throw new InvalidOperationException($"Backing store {this} is read-only.");
}
protected CultPersistedRecord ToPersistedRecord(CultStoredDocument entry, Func<object, byte[]> serializePayload)
{
// A variant persists its delta and an empty payload; the resolved view is never written.
return new CultPersistedRecord
{
Key = entry.Key.Value,
SchemaId = entry.Descriptor.SchemaId,
StoredAt = entry.StoredAt,
Payload = entry.Variant != null ? Array.Empty<byte>() : serializePayload(entry.Document),
Variant = entry.Variant
};
}
// The one load surface: a persisted record resolved against the current type, with the report of what that
// resolution shed. A plain record sheds a member its type dropped by never carrying it into the document; a variant
// sheds the override of such a slot by never carrying it into the delta, so a later type reusing the slot number
// cannot resurrect it and a flush needs no rule of its own. The report is the schema diff's ignored slots plus the
// overrides whose slot the type does not have, including one the persisted schema never listed.
protected CultStoredDocument ToStoredDocument(
CultPersistedRecord record,
IReadOnlyCollection<CultSchemaCatalogEntry> catalog,
Func<Type, byte[], object> deserializePayload,
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
};
}
var document = deserializePayload(resolution.Descriptor.DocumentType, record.Payload);
return new CultStoredDocument(
new CultRecordKey(record.Key),
record.StoredAt,
resolution.Descriptor,
document)
{
StoredSchemaId = record.SchemaId
};
}
protected void SetLastSchemaMigrationReports(IEnumerable<CultSchemaMigrationReport> reports)
{
_lastSchemaMigrationReports = reports?.ToArray() ?? Array.Empty<CultSchemaMigrationReport>();
}
// 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.
protected CultForeignRecord? Foreign(CultPersistedRecord record, IReadOnlyCollection<CultSchemaCatalogEntry> catalog) =>
Registry.Claimant(record.SchemaId, catalog, out var persisted) == null
? new CultForeignRecord(record.Key, record.SchemaId, persisted!.SchemaName, record.StoredAt)
: null;
protected void SetForeignRecords(IEnumerable<CultForeignRecord> foreign)
{
_foreignRecords = foreign.OrderBy(record => record.Key, StringComparer.Ordinal).ToArray();
}
// A write never replaces or removes a record this store carries for a schema it does not know.
protected void RefuseOverwrite(string key)
{
var foreign = _foreignRecords.FirstOrDefault(record => string.Equals(record.Key, key, StringComparison.Ordinal));
if (foreign != null)
throw Overwrites(foreign);
}
protected CultSchemaConflictException Overwrites(CultForeignRecord foreign) => new(
$"Record '{foreign.Key}' in {this} is stored under schema id '{foreign.SchemaId}' ('{foreign.SchemaName}'), which " +
"no registered type owns or lists as compatible. The store carries it untouched and refuses a write that would replace or remove it; " +
"declare the id on a type to claim it.",
foreign.SchemaId,
new[] { foreign.SchemaName },
foreign.Key);
protected void MarkFlushSucceeded()
{
IsDirty = false;
}
}
public abstract class SingleFileBackingStore : CacheBackingStore
{
protected SingleFileBackingStore(string filePath, bool readOnly = false)
: base(readOnly)
{
FileInfo = new FileInfo(filePath);
}
protected FileInfo FileInfo { get; }
protected abstract byte[] SerializeSnapshot(CultPersistedStoreSnapshot snapshot);
/// <summary>
/// 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 <see cref="CultStoreUnreadableException"/> 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.
/// </summary>
protected abstract CultPersistedStoreSnapshot DeserializeSnapshot(byte[] data);
protected abstract byte[] SerializePayload(object document);
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 snapshot = ReadSnapshot();
if (snapshot == null)
{
SetLastSchemaMigrationReports(Array.Empty<CultSchemaMigrationReport>());
SetForeignRecords(Array.Empty<CultForeignRecord>());
return;
}
var reports = new List<CultSchemaMigrationReport>(snapshot.Records.Length);
var persisted = new Dictionary<string, CultStoredDocument>(StringComparer.Ordinal);
var foreign = new List<CultForeignRecord>();
foreach (var record in snapshot.Records)
{
if (Foreign(record, snapshot.SchemaCatalog) is { } carried)
{
foreign.Add(carried);
continue;
}
var stored = ToStoredDocument(record, 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) ||
existing.StoredAt != stored.StoredAt ||
existing.StoredSchemaId != stored.StoredSchemaId)
.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;
SetLastSchemaMigrationReports(reports);
SetForeignRecords(foreign);
_unowned.Clear();
foreach (var record in snapshot.Records)
{
if (!Entries.TryGetValue(record.Key, out var held) || Owns(held.Descriptor, record.SchemaId))
continue;
// An entry under the type's own id that lists the record's id says the record is this schema's: it is re-encoded.
var publisher = Publisher(record.SchemaId, snapshot.SchemaCatalog);
if (!Owns(held.Descriptor, publisher.SchemaId))
_unowned[record.Key] = (held, record, publisher, SerializePayload(held.Document));
}
}
// The records this store loaded under an id their type neither owns nor lists as compatible (another runtime's id, or an
// earlier declaration of the schema, resolved through the catalog's name or its pointer to a local id), each with the
// catalog entry that published it, keyed while the store holds the entry it loaded. A whole-view write lays such a record
// back exactly as it was stored, id and catalog entry included; only a write of the record itself stores it under its
// type's own id. Payload is what the loaded document serialized to when it was loaded, so a change made to the document in
// place is told from a record nobody wrote.
private readonly Dictionary<string, (CultStoredDocument Loaded, CultPersistedRecord Record, CultSchemaCatalogEntry Entry, byte[] Payload)> _unowned =
new(StringComparer.Ordinal);
private static bool Owns(CultDocumentDescriptor descriptor, string schemaId) =>
string.Equals(descriptor.SchemaId, schemaId, StringComparison.Ordinal) ||
descriptor.CompatibleSchemaIds.Contains(schemaId, StringComparer.Ordinal);
// The entry that publishes a schema id: the one that owns it, else the first that lists it.
private static CultSchemaCatalogEntry Publisher(string schemaId, IReadOnlyCollection<CultSchemaCatalogEntry> catalog) =>
catalog.FirstOrDefault(entry => string.Equals(entry.SchemaId, schemaId, StringComparison.Ordinal))
?? catalog.First(entry => entry.CompatibleSchemaIds.Contains(schemaId, StringComparer.Ordinal));
public override void Push(CultStoredDocument entry)
{
ThrowIfReadOnly();
Held(() =>
{
RefuseOverwrite(entry.Key.Value);
Entries[entry.Key.Value] = entry;
IsDirty = true;
});
}
public override void Delete(CultStoredDocument entry)
{
ThrowIfReadOnly();
Held(() =>
{
Entries.TryRemove(entry.Key.Value, out _);
IsDirty = true;
});
}
// A plain flush writes this store's whole view under the lock: two writers never interleave bytes, but the last
// one wins over any file this runtime can read. Processes sharing a store must use conditional commit. 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 foreign = ForeignOnDisk(disk);
var view = WholeView(Entries.Values, disk, foreign);
WriteSnapshot(
view.Records,
Entries.Values.Select(entry => entry.Descriptor.ToCatalogEntry()).ToArray(),
view.Arrived,
Entries.Values.Any(entry => entry.HoldsIds && !view.LaidBack.Contains(entry.Key.Value)),
disk?.FormatVersion,
wholeStore: foreign.Count == 0);
WroteWholeView(view.Records, view.LaidBack);
}
MarkFlushSucceeded();
});
}
public override CultCommitOutcome CommitBatch(CultCommitRequest request, bool wait)
{
ThrowIfReadOnly();
return Held(() => CommitBatchCore(request, wait));
}
private CultCommitOutcome CommitBatchCore(CultCommitRequest request, bool wait)
{
using var fileLock = AcquireLock(wait);
if (fileLock == null)
return CultCommitOutcome.Contended;
var disk = ReadSnapshot() ?? new CultPersistedStoreSnapshot();
if (!request.ConditionsHold(disk.Records, Entries.Values))
return CultCommitOutcome.Mismatch;
// An unconditional commit writes what a flush would, this store's whole view plus the batch: last-writer-wins,
// and staged single writes land with it. A conditional commit (store clean) lands the batch onto the file as it is.
var ontoDisk = request.HasConditions && !IsDirty;
var foreign = ForeignOnDisk(disk);
RefuseForeign(request.Upserts.Concat(request.Deletes), foreign);
var view = ontoDisk ? null : WholeView(Entries.Values, disk, foreign);
var records = (view?.Records ?? disk.Records).ToDictionary(record => record.Key, StringComparer.Ordinal);
// The committer judged the batch against its own view; the file may have moved since. A merge is judged on the
// set the file will hold, exactly as that set would be judged loading: what other processes wrote (arriving),
// what they removed (departing), and this batch.
if (ontoDisk && Judging != null && MergeInvolvesVariants(request, disk))
JudgeMerge(request, disk);
foreach (var entry in request.Deletes)
records.Remove(entry.Key.Value);
foreach (var entry in request.Upserts)
records[entry.Key.Value] = ToPersistedRecord(entry, SerializePayload);
// Onto the file, only the batch's records are seen: the rest stay as they are, and the header they carry stays. An
// unconditional commit writes this store's whole view, so what its records hold decides. A record this commit
// replaces or removes no longer counts.
var replaced = request.Upserts.Concat(request.Deletes).Select(entry => entry.Key.Value).ToHashSet(StringComparer.Ordinal);
var holdsIds = request.Upserts.Any(entry => entry.HoldsIds) ||
(!ontoDisk && Entries.Values.Any(entry => entry.HoldsIds && !replaced.Contains(entry.Key.Value) && !view!.LaidBack.Contains(entry.Key.Value)));
var registered = Entries.Values.Select(entry => entry.Descriptor.ToCatalogEntry())
.Concat(request.Upserts.Select(entry => entry.Descriptor.ToCatalogEntry()))
.ToArray();
WriteSnapshot(records.Values, registered, view?.Arrived ?? disk.SchemaCatalog, holdsIds, disk.FormatVersion, wholeStore: !ontoDisk && foreign.Count == 0);
foreach (var entry in request.Deletes)
Entries.TryRemove(entry.Key.Value, out _);
foreach (var entry in request.Upserts)
Entries[entry.Key.Value] = entry;
if (view != null)
WroteWholeView(records.Values, view.LaidBack);
MarkFlushSucceeded();
return CultCommitOutcome.Committed;
}
// A record's (schema id, storedAt) names its stored bytes, and a commit condition trusts that name. So a whole-view write
// that stores a record differently is a new store of it and mints it a later storedAt: under another id than it was loaded
// under (a type re-encoding a record it read under an older id it declares), or with other
// bytes than the file holds at that storedAt. A record written exactly as it is stored keeps its storedAt.
// Records and catalog entries this runtime does not own survive the write: a record loaded under an id its type does not own
// and that nothing has written since is laid back as it was loaded, with the catalog entry that published it, and every
// foreign record the file holds now is laid back as the file holds it. An entry whose key the file holds as a foreign record
// is refused. Arrived is the catalog the write publishes those records by.
private sealed class WholeViewWrite
{
public CultPersistedRecord[] Records = Array.Empty<CultPersistedRecord>();
public CultSchemaCatalogEntry[] Arrived = Array.Empty<CultSchemaCatalogEntry>();
// Keys written as the file holds or held them, not as this cache holds them.
public HashSet<string> LaidBack = new(StringComparer.Ordinal);
}
// A write over a record the file holds as foreign is refused, and what this store held for the refused keys goes with the
// refusal: its copy is not what the file holds, and kept it would refuse every later write to other records too.
private void RefuseForeign(
IEnumerable<CultStoredDocument> written,
IReadOnlyDictionary<string, (CultForeignRecord Foreign, CultPersistedRecord Record)> foreign)
{
var refused = written.Where(entry => foreign.ContainsKey(entry.Key.Value)).ToArray();
if (refused.Length == 0)
return;
foreach (var entry in refused)
Entries.TryRemove(entry.Key.Value, out _);
throw Overwrites(foreign[refused[0].Key.Value].Foreign);
}
private WholeViewWrite WholeView(
IEnumerable<CultStoredDocument> view,
CultPersistedStoreSnapshot? disk,
IReadOnlyDictionary<string, (CultForeignRecord Foreign, CultPersistedRecord Record)> foreign)
{
var entries = view.ToArray();
RefuseForeign(entries, foreign);
var write = new WholeViewWrite();
var laidBackEntries = new List<CultSchemaCatalogEntry>();
var onDisk = (disk?.Records ?? Array.Empty<CultPersistedRecord>()).ToDictionary(record => record.Key, StringComparer.Ordinal);
write.Records = entries.Select(entry =>
{
if (_unowned.TryGetValue(entry.Key.Value, out var loaded) && ReferenceEquals(loaded.Loaded, entry) &&
SerializePayload(entry.Document).AsSpan().SequenceEqual(loaded.Payload))
{
// Laid back only while the file still holds what was loaded: after that the file's record is its owner's newer
// write, which stays as it is, and a record the file no longer holds is not written back.
write.LaidBack.Add(entry.Key.Value);
if (!onDisk.TryGetValue(entry.Key.Value, out var current))
return null;
if (current.SchemaId != loaded.Record.SchemaId || current.StoredAt != loaded.Record.StoredAt)
return current;
laidBackEntries.Add(loaded.Entry);
return loaded.Record;
}
var record = ToPersistedRecord(entry, SerializePayload);
var unchanged = onDisk.TryGetValue(record.Key, out var stored) && stored.StoredAt == record.StoredAt
? SameStoredBytes(stored, record)
: entry.StoredSchemaId == record.SchemaId;
if (!unchanged)
record.StoredAt = CultCache.MintStoredAt(record.StoredAt);
return record;
}).OfType<CultPersistedRecord>().Concat(foreign.Values.Select(carried => carried.Record)).ToArray();
// An entry that published a laid-back record is written as it was loaded, whatever the file's entry for that id says now.
var laidBackIds = laidBackEntries.Select(entry => entry.SchemaId).ToHashSet(StringComparer.Ordinal);
write.Arrived = laidBackEntries.Distinct()
.Concat((disk?.SchemaCatalog ?? Array.Empty<CultSchemaCatalogEntry>()).Where(entry => !laidBackIds.Contains(entry.SchemaId)))
.ToArray();
return write;
}
// The records the file holds that no registered type claims, by key.
private Dictionary<string, (CultForeignRecord Foreign, CultPersistedRecord Record)> ForeignOnDisk(CultPersistedStoreSnapshot? disk)
{
var foreign = new Dictionary<string, (CultForeignRecord, CultPersistedRecord)>(StringComparer.Ordinal);
foreach (var record in disk?.Records ?? Array.Empty<CultPersistedRecord>())
{
if (Foreign(record, disk!.SchemaCatalog) is { } carried)
foreign[record.Key] = (carried, record);
}
return foreign;
}
private static bool SameStoredBytes(CultPersistedRecord stored, CultPersistedRecord written) =>
stored.SchemaId == written.SchemaId &&
stored.Payload.AsSpan().SequenceEqual(written.Payload) &&
(stored.Variant == null
? written.Variant == null
: written.Variant != null &&
stored.Variant.BaseKey == written.Variant.BaseKey &&
stored.Variant.Overrides.Count == written.Variant.Overrides.Count &&
stored.Variant.Overrides.Zip(written.Variant.Overrides, (a, b) =>
a.Op == b.Op && a.Id == b.Id && a.Value.AsSpan().SequenceEqual(b.Value) &&
a.Path.Select(step => (step.Slot, step.ElementId)).SequenceEqual(b.Path.Select(step => (step.Slot, step.ElementId))))
.All(same => same));
// The file now holds every record written, under the id and at the storedAt the write gave it: the store's entries and the
// cache's say so. A laid-back or carried record is on disk as the file had it, so the cache learns nothing new about it: in
// particular the ids it minted at load are still in memory only.
private void WroteWholeView(IReadOnlyCollection<CultPersistedRecord> written, HashSet<string> laidBack)
{
var asHeld = written.Where(record => !laidBack.Contains(record.Key)).ToArray();
foreach (var record in asHeld)
{
if (!Entries.TryGetValue(record.Key, out var entry))
continue;
entry.StoredSchemaId = record.SchemaId;
entry.StoredAt = record.StoredAt;
}
Cache?.WroteWholeView(asHeld);
}
// Only a variant makes a merge more than the committer's own judgement: a variant it lands, one another writer wrote
// (read from the record's header, never its payload), or one this cache holds. A plain-only store never decodes
// another writer's records, so a schema this cache does not register cannot fail a commit that involves no variant.
private bool MergeInvolvesVariants(CultCommitRequest request, CultPersistedStoreSnapshot disk) =>
Cache?.HoldsVariants == true ||
request.Upserts.Any(entry => entry.Variant != null) ||
disk.Records.Any(record => record.Variant != null);
private void JudgeMerge(CultCommitRequest request, CultPersistedStoreSnapshot disk)
{
var landing = request.Upserts.Select(entry => entry.Key.Value).ToHashSet(StringComparer.Ordinal);
var removing = request.Deletes.Select(entry => entry.Key.Value).ToHashSet(StringComparer.Ordinal);
var onDisk = disk.Records.Select(record => record.Key).ToHashSet(StringComparer.Ordinal);
var arriving = new List<CultStoredDocument>(request.Upserts);
var departing = new List<CultStoredDocument>(request.Deletes);
foreach (var record in disk.Records)
{
if (landing.Contains(record.Key) || removing.Contains(record.Key))
continue;
// A foreign record is not part of what the cache judges; one that replaced a record this store holds takes it away.
if (Foreign(record, disk.SchemaCatalog) != null)
{
if (Entries.TryGetValue(record.Key, out var replaced))
departing.Add(replaced);
continue;
}
if (Entries.TryGetValue(record.Key, out var known) && known.StoredAt == record.StoredAt && known.StoredSchemaId == record.SchemaId)
continue;
arriving.Add(ToStoredDocument(record, disk.SchemaCatalog, DeserializePayload, out _));
}
departing.AddRange(Entries.Values.Where(known =>
!onDisk.Contains(known.Key.Value) && !landing.Contains(known.Key.Value) && !removing.Contains(known.Key.Value)));
Judging!(arriving, departing);
}
// Open, flush and commit read the file through this one call, so they agree on which files are stores. A file that is
// gone or zero bytes is an empty store: nothing to read, nothing to lose.
private CultPersistedStoreSnapshot? ReadSnapshot()
{
var bytes = ReadDisk();
if (bytes == null)
return null;
try
{
return DeserializeSnapshot(bytes);
}
catch (CultStoreUnreadableException ex) when (ex.Path == null)
{
throw new CultStoreUnreadableException(ex.Message, FileInfo.FullName, ex.InnerException);
}
}
private byte[]? ReadDisk()
{
FileInfo.Refresh();
if (!FileInfo.Exists)
return null;
try
{
return ReadAllBytesShared(FileInfo.FullName) is { Length: > 0 } bytes ? bytes : null;
}
catch (FileNotFoundException)
{
return null;
}
}
private void WriteSnapshot(IEnumerable<CultPersistedRecord> records, IReadOnlyCollection<CultSchemaCatalogEntry> registered, IReadOnlyCollection<CultSchemaCatalogEntry> arrived, bool holdsElementIds, string? existingHeader, bool wholeStore)
{
var ordered = records.OrderBy(record => record.Key, StringComparer.Ordinal).ToArray();
var snapshot = new CultPersistedStoreSnapshot
{
FormatVersion = HeaderFor(holdsElementIds, existingHeader, wholeStore, directoryStore: false, holdsVariants: ordered.Any(record => record.Variant != null)),
SchemaCatalog = CultSchemaCatalogEntry.Derive(ordered, registered, arrived)
.OrderBy(entry => entry.SchemaName, StringComparer.Ordinal)
.ThenBy(entry => entry.SchemaId, StringComparer.Ordinal)
.ToArray(),
Records = ordered
};
Directory.CreateDirectory(FileInfo.DirectoryName!);
WriteSnapshotAtomically(FileInfo.FullName, SerializeSnapshot(snapshot));
}
// 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();
}
}
}