diff --git a/research/cultnet-distributed-database/summary.md b/research/cultnet-distributed-database/summary.md index 9a8f4eb3..82caa713 100644 --- a/research/cultnet-distributed-database/summary.md +++ b/research/cultnet-distributed-database/summary.md @@ -171,6 +171,8 @@ Shard catalog exchange now exists: - background shard log replicator plus schema-v0 fetcher for primary endpoints - restart-safe replica cursor store with a local MessagePack file implementation +- client authority scopes for predicted local input documents +- predicted/reconciled change events for client-side prediction ## Next Cut @@ -179,6 +181,8 @@ Build durable authoritative mutation logs: - persist per-shard primary logs - define the snapshot fallback boundary for compacted history - return explicit resync-required responses when requested history is gone +- then add simulation-frame rollback and resimulation helpers on top of + predicted input streams That is the next foundation needed before membership, Raft-style failover, or CRDT policy work can be honest. diff --git a/src/GameCult.Networking/Contracts/cultnet-distributed-database.md b/src/GameCult.Networking/Contracts/cultnet-distributed-database.md index b696b522..41594a2a 100644 --- a/src/GameCult.Networking/Contracts/cultnet-distributed-database.md +++ b/src/GameCult.Networking/Contracts/cultnet-distributed-database.md @@ -59,6 +59,8 @@ Implemented: - schema-v0 shard log fetcher for `cultnet://host:port` endpoints - restart-safe replica cursor store, including a local MessagePack file-backed implementation +- client-authority scopes for locally predicted input documents +- predicted and reconciled database change kinds for client-side prediction Not implemented yet: @@ -66,6 +68,7 @@ Not implemented yet: - leader election or automatic shard failover - durable log persistence and compaction - snapshot fallback for compacted log history +- rollback/resimulation helpers for simulation frames - declared CRDT merge policies ## Live Invariants @@ -79,6 +82,11 @@ Not implemented yet: - Raw snapshot/put/delete messages must pass through `CultNetDatabase`, not bypass shard policy. - Realtime subscriptions publish domain changes, not storage envelopes. +- Client-owned input documents may be predicted locally only inside explicit + `CultNetClientAuthorityScope` declarations. +- Predicted input state is not authoritative shared state. The shard log still + decides the committed ordering and emits reconciliation when it corrects or + confirms local prediction. ## Next Coherent Slice @@ -87,9 +95,12 @@ Add durable mutation-log storage and snapshot fallback: - persist per-shard mutation logs instead of keeping them only in memory - define the compaction boundary that turns old log reads into snapshot reads - return `ResyncRequired` when a replica asks for compacted history +- after that, add simulation-frame rollback/resimulation helpers around + predicted input streams The log is wire-readable, replicas can apply it explicitly, a pull loop can drive non-primary shards from primary endpoints, and replica cursors can survive -restart. The next useful layer is making authoritative mutation logs durable and -compaction-aware. Membership and leader election should wait until basic replica -catch-up is boring. +restart. Client-owned input documents can now be predicted locally and +reconciled when the authoritative log arrives. The next useful layer is making +authoritative mutation logs durable and compaction-aware. Membership and leader +election should wait until basic replica catch-up is boring. diff --git a/src/GameCult.Networking/CultNetDatabase.cs b/src/GameCult.Networking/CultNetDatabase.cs index 0672ea8c..f2bd0790 100644 --- a/src/GameCult.Networking/CultNetDatabase.cs +++ b/src/GameCult.Networking/CultNetDatabase.cs @@ -31,7 +31,57 @@ namespace GameCult.Networking /// /// A document change was rejected. /// - Rejected + Rejected, + /// + /// A document was applied locally before authoritative commit. + /// + Predicted, + /// + /// A predicted document was replaced by the authoritative committed state. + /// + Reconciled + } + + /// + /// Declares which documents this runtime may predict locally before authoritative commit. + /// + public sealed class CultNetClientAuthorityScope + { + /// + /// Creates a client authority scope. + /// + public CultNetClientAuthorityScope( + string ownerRuntimeId, + IEnumerable? schemaIds = null, + string? keyPrefix = null) + { + OwnerRuntimeId = string.IsNullOrWhiteSpace(ownerRuntimeId) + ? throw new ArgumentException("Value must be non-empty.", nameof(ownerRuntimeId)) + : ownerRuntimeId; + SchemaIds = schemaIds?.Where(value => !string.IsNullOrWhiteSpace(value)).Distinct(StringComparer.Ordinal).ToArray() + ?? Array.Empty(); + KeyPrefix = keyPrefix; + } + + /// + /// Gets the runtime that owns authoring for this scope. + /// + public string OwnerRuntimeId { get; } + /// + /// Gets the schema ids governed by this scope. Empty means all schemas. + /// + public IReadOnlyList SchemaIds { get; } + /// + /// Gets the optional record-key prefix governed by this scope. + /// + public string? KeyPrefix { get; } + + internal bool Matches(string runtimeId, string schemaId, CultRecordKey key) + { + return string.Equals(OwnerRuntimeId, runtimeId, StringComparison.Ordinal) && + (SchemaIds.Count == 0 || SchemaIds.Contains(schemaId, StringComparer.Ordinal)) && + (string.IsNullOrEmpty(KeyPrefix) || key.Value.StartsWith(KeyPrefix!, StringComparison.Ordinal)); + } } /// @@ -168,6 +218,10 @@ namespace GameCult.Networking /// public IReadOnlyList? Shards { get; set; } /// + /// Gets or sets document scopes this runtime may predict before authoritative commit. + /// + public IReadOnlyList? ClientAuthorityScopes { get; set; } + /// /// Gets or sets the document registry used to create and apply raw CultNet document messages. /// public CultNetDocumentRegistry? DocumentRegistry { get; set; } @@ -328,7 +382,10 @@ namespace GameCult.Networking { private readonly CultCache _cache; private readonly CultNetDocumentRegistry _documents; + private readonly string _runtimeId; private readonly List _shards; + private readonly List _clientAuthorityScopes; + private readonly Dictionary _predictedDocuments = new(StringComparer.Ordinal); private readonly Dictionary> _mutationLogs = new(StringComparer.Ordinal); private readonly Dictionary _nextLogSequences = new(StringComparer.Ordinal); @@ -343,11 +400,13 @@ namespace GameCult.Networking { _cache = cache ?? throw new ArgumentNullException(nameof(cache)); options ??= new CultNetDatabaseOptions(); + _runtimeId = options.RuntimeId; _documents = options.DocumentRegistry ?? new CultNetDocumentRegistry(cache.Registry); _shards = (options.Shards == null || options.Shards.Count == 0 ? new[] { CultNetShardDescriptor.PrimaryAll(options.RuntimeId) } : options.Shards) .ToList(); + _clientAuthorityScopes = (options.ClientAuthorityScopes ?? Array.Empty()).ToList(); _cache.OnUpdate += PublishCacheUpdate; } @@ -364,6 +423,11 @@ namespace GameCult.Networking /// public IReadOnlyList Shards => _shards; + /// + /// Gets the document scopes this runtime may predict before authoritative commit. + /// + public IReadOnlyList ClientAuthorityScopes => _clientAuthorityScopes; + /// /// Gets mutation log entries for a shard after the supplied sequence. /// @@ -486,6 +550,31 @@ namespace GameCult.Networking return handle; } + /// + /// Applies a locally predicted document for a client-owned input scope. + /// + public async Task> PutPredictedAsync(CultRecordKey key, T document) where T : class + { + ThrowIfDisposed(); + if (document == null) throw new ArgumentNullException(nameof(document)); + + var descriptor = _cache.Registry.GetRequired(); + var shard = ResolveShardInternal(descriptor.SchemaId, key); + EnsureClientAuthority(descriptor.SchemaId, key); + + var previous = _cache.Get(key); + var handle = await _cache.UpsertAsync(document, new CultRecordHandle(key)).ConfigureAwait(false); + _predictedDocuments[PredictionKey(descriptor.SchemaId, key)] = document; + Publish(new CultNetDatabaseChange( + CultNetDatabaseChangeKind.Predicted, + key, + descriptor.SchemaId, + shard, + document, + previous)); + return handle; + } + /// /// Deletes a document by key. /// @@ -537,16 +626,23 @@ namespace GameCult.Networking var descriptor = _cache.Registry.GetRequiredBySchemaId(message.Document.SchemaId); var previous = _cache.Get(key); var document = await _documents.ApplyRawDocumentPutMessageAsync(_cache, message).ConfigureAwait(false); + var kind = previous == null ? CultNetDatabaseChangeKind.Added : CultNetDatabaseChangeKind.Updated; + var predictionKey = PredictionKey(descriptor.SchemaId, key); + if (_predictedDocuments.Remove(predictionKey)) + { + kind = CultNetDatabaseChangeKind.Reconciled; + } + AppendMutationLogEntry( shard, - previous == null ? CultNetDatabaseChangeKind.Added : CultNetDatabaseChangeKind.Updated, + kind == CultNetDatabaseChangeKind.Reconciled ? CultNetDatabaseChangeKind.Updated : kind, key, descriptor.SchemaId, document, previous); PublishUntyped( descriptor.DocumentType, - previous == null ? CultNetDatabaseChangeKind.Added : CultNetDatabaseChangeKind.Updated, + kind, key, descriptor.SchemaId, shard, @@ -872,12 +968,18 @@ namespace GameCult.Networking $"Shard '{shard.ShardId}' log entry {entry.Sequence} has a removed kind with a put payload."); } + var predictionKey = PredictionKey(descriptor.SchemaId, key); + if (_predictedDocuments.Remove(predictionKey)) + { + kind = CultNetDatabaseChangeKind.Reconciled; + } + RecordMutationLogEntry(new CultNetShardMutationLogEntry( shard.ShardId, shard.Epoch, entry.Sequence, entry.CommittedAt, - kind, + kind == CultNetDatabaseChangeKind.Reconciled ? CultNetDatabaseChangeKind.Updated : kind, descriptor.SchemaId, key, document, @@ -949,6 +1051,25 @@ namespace GameCult.Networking }; } + private void EnsureClientAuthority(string schemaId, CultRecordKey key) + { + if (_clientAuthorityScopes.Any(scope => scope.Matches(_runtimeId, schemaId, key))) + { + return; + } + + var shard = ResolveShardInternal(schemaId, key); + throw new CultNetShardAuthorityException( + shard, + $"Runtime '{_runtimeId}' does not have client prediction authority for schema '{schemaId}' key '{key.Value}'.", + "not_client_authority"); + } + + private static string PredictionKey(string schemaId, CultRecordKey key) + { + return $"{schemaId}:{key.Value}"; + } + private void PublishUntyped( Type documentType, CultNetDatabaseChangeKind kind, diff --git a/src/GameCult.Networking/README.md b/src/GameCult.Networking/README.md index c092c8ca..97c2222e 100644 --- a/src/GameCult.Networking/README.md +++ b/src/GameCult.Networking/README.md @@ -148,6 +148,12 @@ transport path. Replica cursors can be stored through those cursors in one local MessagePack file so replicas can resume after a restart. +For client-side prediction, `CultNetClientAuthorityScope` declares which input +documents a runtime may author optimistically. `PutPredictedAsync` updates the +local cache immediately and emits a `Predicted` change; when the authoritative +shard log arrives for the same record, the database emits `Reconciled` and the +cache holds the committed value. + The first coherent shard policy is primary ownership. Each shard has one authoritative writer at a time. Clients may connect to any node; a non-owner node forwards a write to the owner or rejects it honestly when forwarding is diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs index dfba1bc5..fe09adff 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -923,6 +923,125 @@ namespace GameCult.Networking.Tests Assert.That(targetDatabase.GetMutationLog("players-replica"), Has.Count.EqualTo(1)); } + [Test] + public async Task CultNetDatabase_Predicts_ClientOwnedInput_AndReconciles_AuthoritativeLog() + { + var cache = new CultCache(); + var registry = new CultNetDocumentRegistry(cache.Registry) + .Register(CultNetDocumentBinding.ForDocument( + cache.Registry, + payloadSerializer: SerializePlayerDataPayload, + payloadDeserializer: DeserializePlayerDataPayload)); + var schemaId = cache.Registry.GetRequired().SchemaId; + var key = new CultRecordKey("input:client-a"); + var database = new CultNetDatabase(cache, new CultNetDatabaseOptions + { + RuntimeId = "client-a", + DocumentRegistry = registry, + Shards = + [ + new CultNetShardDescriptor( + "inputs", + "server", + epoch: 1, + isPrimary: false, + schemaIds: [schemaId], + keyPrefix: "input:", + primaryEndpoints: ["cultnet://server.example.test:3075"]) + ], + ClientAuthorityScopes = + [ + new CultNetClientAuthorityScope( + "client-a", + schemaIds: [schemaId], + keyPrefix: "input:client-a") + ] + }); + var changes = new List>(); + using var subscription = database.WatchRecord(key) + .Subscribe(change => changes.Add(change)); + var predicted = new PlayerData + { + PlayerId = Guid.NewGuid(), + Email = "input@example.test", + PasswordHash = "hash", + Username = "Predicted" + }; + var authoritative = new PlayerData + { + PlayerId = predicted.PlayerId, + Email = predicted.Email, + PasswordHash = predicted.PasswordHash, + Username = "Authoritative" + }; + var put = registry.CreateRawDocumentPutMessage("input-commit", new CultRecordHandle(key), authoritative); + put.ShardId = "inputs"; + put.ShardEpoch = 1; + + await database.PutPredictedAsync(key, predicted); + await database.ApplyShardLogResponseAsync(new CultNetShardLogResponseMessage + { + ShardId = "inputs", + ShardEpoch = 1, + Entries = + [ + new CultNetShardLogEntryMessage + { + Sequence = 1, + CommittedAt = "2026-05-19T12:00:00.0000000Z", + ChangeKind = "updated", + Put = put + } + ] + }); + + Assert.That(cache.Get(key)!.Username, Is.EqualTo("Authoritative")); + Assert.That(changes.Exists(change => change.Kind == CultNetDatabaseChangeKind.Predicted), Is.True); + Assert.That(changes.Exists(change => change.Kind == CultNetDatabaseChangeKind.Reconciled), Is.True); + } + + [Test] + public void CultNetDatabase_Rejects_Prediction_OutsideClientAuthorityScope() + { + var cache = new CultCache(); + var schemaId = cache.Registry.GetRequired().SchemaId; + var database = new CultNetDatabase(cache, new CultNetDatabaseOptions + { + RuntimeId = "client-a", + Shards = + [ + new CultNetShardDescriptor( + "inputs", + "server", + epoch: 1, + isPrimary: false, + schemaIds: [schemaId], + keyPrefix: "input:") + ], + ClientAuthorityScopes = + [ + new CultNetClientAuthorityScope( + "client-a", + schemaIds: [schemaId], + keyPrefix: "input:client-a") + ] + }); + + Assert.That( + async () => await database.PutPredictedAsync( + new CultRecordKey("input:client-b"), + new PlayerData + { + PlayerId = Guid.NewGuid(), + Email = "input-b@example.test", + PasswordHash = "hash", + Username = "WrongClient" + }), + Throws.TypeOf() + .With.Property(nameof(CultNetShardAuthorityException.Reason)) + .EqualTo("not_client_authority")); + } + [Test] public void CultNetDatabase_Rejects_ShardLogGap() {