CultLib/src/GameCult.Caching/CultCache.cs
Meta Vens b4edbb1eeb Union golden: the store's clock is an internal seam, the writer pins it, the gate test runs under any switch
CultCache mints storedAt from an internal Clock (default the wall clock); the vector writer pins it, so
union-surface.cc is regenerated in process and byte-compared on every run. The gate test sets the switch
itself and writes only a probe, so it neither skips nor touches a committed vector.

This commit does not pass: union-surface.cc still holds the old wall-clock stored-at; the next commit
regenerates it. Builds.

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

3556 lines
164 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>();
}
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; }
}
public sealed class CultPersistedStoreSnapshot
{
public const string FormatV1 = "cultcache.store.v1";
public const string FormatV2 = "cultcache.store.v2";
public string FormatVersion { get; set; } = FormatV1;
public CultSchemaCatalogEntry[] SchemaCatalog { get; set; } = Array.Empty<CultSchemaCatalogEntry>();
public CultPersistedRecord[] Records { get; set; } = Array.Empty<CultPersistedRecord>();
}
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,
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;
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; }
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],
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;
_document = document;
}
// A variant: the delta is what is stored; the cache resolves it against its base in the hold that admits it.
internal CultStoredDocument(
CultRecordKey key,
string storedAt,
CultDocumentDescriptor descriptor,
CultVariantDelta variant,
object? resolved)
{
Key = key;
StoredAt = storedAt;
Descriptor = descriptor;
Variant = variant;
_document = resolved;
}
private readonly object? _document;
public CultRecordKey Key { get; }
public string StoredAt { get; }
public CultDocumentDescriptor Descriptor { get; }
// The complete document. For a variant, the resolved view: derived, never written back.
public object Document => _document ??
throw new InvalidOperationException($"Variant {Key.Value} has not been resolved; only the cache resolves a variant.");
// The base and overrides a variant record stores; null for a plain record.
public CultVariantDelta? Variant { get; }
// A plain record staged by Flatten: the one way a plain record may replace a variant at its key.
internal bool Flattens { get; set; }
internal bool IsResolved => _document != null;
internal CultStoredDocument Resolved(object document) => new(Key, StoredAt, Descriptor, Variant!, document);
}
public sealed class CultDocumentRegistry
{
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 indexes = _indexes;
if (indexes.BySchemaId.TryGetValue(schemaId, out var exact))
{
return new CultSchemaResolutionResult(
exact,
new CultSchemaMigrationReport
{
PersistedSchemaId = schemaId,
LocalSchemaId = exact.SchemaId,
PersistedSchemaName = exact.SchemaName,
LocalSchemaName = exact.SchemaName,
Kind = CultSchemaMigrationKind.Exact
});
}
var persisted = catalog.FirstOrDefault(entry => string.Equals(entry.SchemaId, schemaId, StringComparison.Ordinal));
if (persisted == null)
{
throw new InvalidOperationException($"Persisted schema '{schemaId}' is not present in the embedded catalog.");
}
foreach (var compatibleSchemaId in persisted.CompatibleSchemaIds.Where(candidate => !string.IsNullOrWhiteSpace(candidate)))
{
if (indexes.BySchemaId.TryGetValue(compatibleSchemaId, out var compatibleLocal))
{
return BuildCompatibleResolutionResult(persisted, compatibleLocal);
}
}
if (indexes.BySchemaName.TryGetValue(persisted.SchemaName, out var localVersions))
{
if (localVersions.Length == 1)
{
return BuildCompatibleResolutionResult(persisted, 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.");
}
throw new InvalidOperationException(
$"No local CultCache schema matches persisted schema '{persisted.SchemaName}' ({schemaId}).");
}
private CultDocumentDescriptor RegisterDescriptor(CultDocumentDescriptor descriptor)
{
lock (_registrationGate)
{
var current = _indexes;
if (current.ByType.TryGetValue(descriptor.DocumentType, out var registered))
return registered;
var updated = new RegistryIndexes(current);
registered = RegisterDescriptor(updated, descriptor);
_indexes = updated;
return registered;
}
}
private static CultDocumentDescriptor RegisterDescriptor(
RegistryIndexes indexes,
CultDocumentDescriptor descriptor)
{
if (indexes.ByType.TryGetValue(descriptor.DocumentType, out var registered))
return registered;
indexes.BySchemaName.TryGetValue(descriptor.SchemaName, out var schemaNameVersions);
var schemaVersionOwner = schemaNameVersions?.FirstOrDefault(candidate =>
string.Equals(candidate.SchemaVersion, descriptor.SchemaVersion, StringComparison.Ordinal));
if (schemaVersionOwner != null)
{
if (IsExactWireAlias(schemaVersionOwner, descriptor))
{
indexes.ByType[descriptor.DocumentType] = descriptor;
return descriptor;
}
throw DuplicateSchemaRegistration(
$"schema name '{descriptor.SchemaName}' version '{descriptor.SchemaVersion}'",
schemaVersionOwner.DocumentType,
descriptor.DocumentType);
}
if (indexes.BySchemaId.TryGetValue(descriptor.SchemaId, out var schemaIdOwner))
{
throw DuplicateSchemaRegistration(
$"schema id '{descriptor.SchemaId}'",
schemaIdOwner.DocumentType,
descriptor.DocumentType);
}
indexes.ByType[descriptor.DocumentType] = descriptor;
indexes.BySchemaId[descriptor.SchemaId] = descriptor;
indexes.BySchemaName[descriptor.SchemaName] = schemaNameVersions == null
? [descriptor]
: [.. schemaNameVersions, descriptor];
return descriptor;
}
private static bool IsExactWireAlias(
CultDocumentDescriptor canonical,
CultDocumentDescriptor candidate) =>
string.Equals(canonical.SchemaName, candidate.SchemaName, StringComparison.Ordinal) &&
string.Equals(canonical.SchemaVersion, candidate.SchemaVersion, StringComparison.Ordinal) &&
string.Equals(canonical.SchemaId, candidate.SchemaId, StringComparison.Ordinal) &&
string.Equals(canonical.ContentHash, candidate.ContentHash, StringComparison.Ordinal) &&
string.Equals(canonical.CanonicalSchemaJson, candidate.CanonicalSchemaJson, StringComparison.Ordinal);
private static InvalidOperationException DuplicateSchemaRegistration(
string identity,
Type existingType,
Type claimedType) =>
new(
$"CultCache {identity} is already registered to CLR type " +
$"'{existingType.FullName}' and cannot also be claimed by '{claimedType.FullName}'. " +
"Schema compatibility must be declared through explicit persisted-schema alias metadata.");
private sealed class RegistryIndexes
{
public RegistryIndexes()
{
ByType = new Dictionary<Type, CultDocumentDescriptor>();
BySchemaId = 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);
BySchemaName = new Dictionary<string, CultDocumentDescriptor[]>(source.BySchemaName, StringComparer.Ordinal);
}
public Dictionary<Type, CultDocumentDescriptor> ByType { get; }
public Dictionary<string, CultDocumentDescriptor> BySchemaId { 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 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);
return new CultDocumentDescriptor(
type,
attribute.SchemaName,
attribute.SchemaVersion,
schemaId,
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>();
if (!string.Equals(persisted.SchemaName, local.SchemaName, StringComparison.Ordinal))
{
errors.Add($"schema name drift '{persisted.SchemaName}' -> '{local.SchemaName}'");
}
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 (schemaId, storedAt), sound because every write to a key mints a later storedAt.
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.Descriptor.SchemaId, 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?.Descriptor.SchemaId, 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;
public CultCodec? Codec => _codec;
// The whole override of one member, encoded with this cache's codec, ready for UpsertVariant.
public CultVariantOverride Override(Type documentType, string member, object? value)
{
if (_codec == null) throw new InvalidOperationException("This cache has no codec; create it with CultCacheMessagePack.Create.");
var found = _registry.GetRequired(documentType).RichMembers.FirstOrDefault(entry => entry.Member.Name == member)
?? throw new ArgumentException($"{documentType.Name} has no persisted member '{member}'.", nameof(member));
return CultVariantOverride.Set(
found.Slot,
_codec.Serialize(value!, found.MemberType, documentType));
}
public CultVariantOverride Override<T>(string member, object? value) where T : class => Override(typeof(T), member, value);
public bool IsDirty
{
get
{
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);
}
/// <summary>
/// Commits the staged batch. Returns false when a condition failed: nothing was written, changed in memory, or published.
/// </summary>
/// <remarks>
/// A commit that succeeds and then fails in a journal or observer still stands and is published, and the writer
/// receives the failure. One failure arrives as itself, several as an <see cref="AggregateException"/>. A writer that
/// wants a particular failure (a CultNetShardLogException, when a database's log refused the commit) catches that type,
/// or unwraps <see cref="AggregateException.InnerExceptions"/> and looks for it.
/// </remarks>
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);
// 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));
}
// The wall clock behind every storedAt. Internal: a test that writes a golden store pins it, so the bytes reproduce.
internal Func<DateTimeOffset> Clock { get; set; } = () => DateTimeOffset.UtcNow;
private string MintStoredAt(string? previous)
{
var now = Clock();
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.Descriptor.SchemaId, 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);
// The one resolution: a variant admitted, a base admitted or evicted, all resolved here against the post-batch
// set, before land and before memory. land receives the admitted records with variants resolved.
var plan = Resolve(admitted, evicted, source);
var outcome = land(home, plan.Admitted);
if (outcome != CultCommitOutcome.Committed)
return outcome;
var applied = Apply(plan, evicted, source);
foreach (var change in applied)
_held.Changes.Add((change, source != null));
RunJournals(applied, _held.Failures);
return CultCommitOutcome.Committed;
});
// Every admission runs in a hold. A nested hold on the same cache adds to the outermost one, which publishes
// exactly its own changes after it leaves the gate, before it returns, on its caller's thread: observers never
// run under the gate, and a write an observer makes is a new outermost hold. Cross-thread delivery order is not
// guaranteed; each change carries the Sequence it was admitted with. Journals run before that, under the gate,
// in Sequence order (AddJournal); no hold may be entered on this cache while one runs. Every change reaches every observer even
// if an observer or an OnUpdate handler throws; then the exception (an AggregateException for several, whatever their
// origin: a journal, an observer, a second database) is rethrown. A writer looking for one type unwraps InnerExceptions.
// If the body threw, its exception wins.
internal T Held<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 seen = new HashSet<int>();
foreach (var entry in delta.Overrides)
{
if (entry.Op != CultOverrideOp.Set || entry.Path.Count != 1 || entry.Path[0].ElementId.Length != 0 || entry.Id.Length != 0)
throw new InvalidOperationException(
$"Variant {key} carries {entry.Op} at {entry.PathText}; this runtime resolves top-level Set overrides only.");
var slot = entry.Path[0].Slot;
if (!seen.Add(slot))
throw new InvalidOperationException($"Variant {key} overrides slot {slot} more than once; a variant has one override per slot.");
var member = descriptor.RichMembers.FirstOrDefault(candidate => candidate.Slot == slot);
// Drift is the load's business (ToStoredDocument sheds an override of a slot the type lacks), so a delta
// reaching here names only slots the type has, unless someone is writing one it never had.
if (member == null)
throw new InvalidOperationException($"Variant {key} overrides slot {slot}, which {type.Name} does not have.");
try
{
_codec.Deserialize(member.MemberType, type, entry.Value);
}
catch (Exception exception)
{
// A value that no longer decodes refuses exactly as a plain record holding it does.
throw new InvalidOperationException(
$"Variant {key} (schema '{descriptor.SchemaId}') overrides '{member.Member.Name}' (slot {slot}) with a value that does not decode as {member.MemberType.Name}: " +
exception.GetBaseException().Message, exception);
}
kept.Add(new KeyValuePair<int, byte[]>(slot, entry.Value));
}
var payload = _codec.Serialize(baseStored.Document, type, type);
var document = _codec.Deserialize(type, type, kept.Count == 0 ? payload : _codec.Overlay(payload, kept))
?? throw new InvalidOperationException($"Variant {key} resolved to nothing.");
stack.RemoveAt(stack.Count - 1);
return done[key] = isVariant
? variant.Resolved(document)
: 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;
// 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;
// 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);
}
object document;
try
{
document = deserializePayload(resolution.Descriptor.DocumentType, record.Payload);
}
catch (Exception exception) when (exception is not OperationCanceledException)
{
// The codec knows what it refused but not which record; the store does. Say both, and never null the record.
var causes = new List<string>();
for (var cause = exception; cause != null; cause = cause.InnerException) causes.Add(cause.Message);
throw new InvalidDataException($"Record '{record.Key}' (schema '{record.SchemaId}') cannot be decoded: {string.Join(" <- ", causes.Distinct())}", exception);
}
return new CultStoredDocument(
new CultRecordKey(record.Key),
record.StoredAt,
resolution.Descriptor,
document);
}
protected void SetLastSchemaMigrationReports(IEnumerable<CultSchemaMigrationReport> reports)
{
_lastSchemaMigrationReports = reports?.ToArray() ?? Array.Empty<CultSchemaMigrationReport>();
}
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);
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>());
return;
}
var reports = new List<CultSchemaMigrationReport>(snapshot.Records.Length);
var persisted = new Dictionary<string, CultStoredDocument>(StringComparer.Ordinal);
foreach (var record in snapshot.Records)
{
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.Descriptor.SchemaId != stored.Descriptor.SchemaId)
.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);
}
public override void Push(CultStoredDocument entry)
{
ThrowIfReadOnly();
Held(() =>
{
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. Processes sharing a store must use conditional commit.
public override void PushAll()
{
ThrowIfReadOnly();
Held(() =>
{
using (AcquireLock(wait: true))
{
WriteSnapshot(
Entries.Values.Select(entry => ToPersistedRecord(entry, SerializePayload)),
Entries.Values.Select(entry => entry.Descriptor.ToCatalogEntry()));
}
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 records = ontoDisk
? disk.Records.ToDictionary(record => record.Key, StringComparer.Ordinal)
: Entries.Values.Select(entry => ToPersistedRecord(entry, SerializePayload)).ToDictionary(record => record.Key, StringComparer.Ordinal);
var catalog = ontoDisk
? disk.SchemaCatalog
: Entries.Values.Select(entry => entry.Descriptor.ToCatalogEntry());
// The committer judged the batch against its own view; the file may have moved since. A merge is judged on the
// set the file will hold, exactly as that set would be judged loading: what other processes wrote (arriving),
// what they removed (departing), and this batch.
if (ontoDisk && Judging != null && MergeInvolvesVariants(request, disk))
JudgeMerge(request, disk);
foreach (var entry in request.Deletes)
records.Remove(entry.Key.Value);
foreach (var entry in request.Upserts)
records[entry.Key.Value] = ToPersistedRecord(entry, SerializePayload);
WriteSnapshot(records.Values, catalog.Concat(request.Upserts.Select(entry => entry.Descriptor.ToCatalogEntry())));
foreach (var entry in request.Deletes)
Entries.TryRemove(entry.Key.Value, out _);
foreach (var entry in request.Upserts)
Entries[entry.Key.Value] = entry;
MarkFlushSucceeded();
return CultCommitOutcome.Committed;
}
// Only a variant makes a merge more than the committer's own judgement: a variant it lands, one another writer wrote
// (read from the record's header, never its payload), or one this cache holds. A plain-only store never decodes
// another writer's records, so a schema this cache does not register cannot fail a commit that involves no variant.
private bool MergeInvolvesVariants(CultCommitRequest request, CultPersistedStoreSnapshot disk) =>
Cache?.HoldsVariants == true ||
request.Upserts.Any(entry => entry.Variant != null) ||
disk.Records.Any(record => record.Variant != null);
private void JudgeMerge(CultCommitRequest request, CultPersistedStoreSnapshot disk)
{
var landing = request.Upserts.Select(entry => entry.Key.Value).ToHashSet(StringComparer.Ordinal);
var removing = request.Deletes.Select(entry => entry.Key.Value).ToHashSet(StringComparer.Ordinal);
var onDisk = disk.Records.Select(record => record.Key).ToHashSet(StringComparer.Ordinal);
var arriving = new List<CultStoredDocument>(request.Upserts);
foreach (var record in disk.Records)
{
if (landing.Contains(record.Key) || removing.Contains(record.Key))
continue;
if (Entries.TryGetValue(record.Key, out var known) && known.StoredAt == record.StoredAt && known.Descriptor.SchemaId == record.SchemaId)
continue;
arriving.Add(ToStoredDocument(record, disk.SchemaCatalog, DeserializePayload, out _));
}
var departing = new List<CultStoredDocument>(request.Deletes);
departing.AddRange(Entries.Values.Where(known =>
!onDisk.Contains(known.Key.Value) && !landing.Contains(known.Key.Value) && !removing.Contains(known.Key.Value)));
Judging!(arriving, departing);
}
private CultPersistedStoreSnapshot? ReadSnapshot()
{
FileInfo.Refresh();
if (!FileInfo.Exists)
return null;
try
{
return DeserializeSnapshot(ReadAllBytesShared(FileInfo.FullName));
}
catch (FileNotFoundException)
{
return null;
}
}
private void WriteSnapshot(IEnumerable<CultPersistedRecord> records, IEnumerable<CultSchemaCatalogEntry> catalog)
{
var ordered = records.OrderBy(record => record.Key, StringComparer.Ordinal).ToArray();
var used = new HashSet<string>(ordered.Select(record => record.SchemaId), StringComparer.Ordinal);
var snapshot = new CultPersistedStoreSnapshot
{
// v2 exactly when the store holds a variant: a store without one stays byte-identical v1.
FormatVersion = ordered.Any(record => record.Variant != null)
? CultPersistedStoreSnapshot.FormatV2
: CultPersistedStoreSnapshot.FormatV1,
SchemaCatalog = catalog
.GroupBy(entry => entry.SchemaId, StringComparer.Ordinal)
.Select(group => group.First())
.Where(entry => used.Contains(entry.SchemaId))
.OrderBy(entry => entry.SchemaName, 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();
}
}
}