From c2b2239853ea876b9ec2035d3d7f63e03f084928 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 19 May 2026 21:07:46 +0100 Subject: [PATCH] Recover replicas from compacted shard history --- src/GameCult.Mesh/README.md | 4 +- src/GameCult.Mesh/docs/public-api.md | 4 +- .../Contracts/cultnet-distributed-database.md | 18 +- src/GameCult.Networking/CultNetDatabase.cs | 176 ++++++++++++++++++ .../CultNetDatabaseServer.cs | 23 +++ .../CultNetSchemaMessages.cs | 20 ++ .../CultNetShardReplication.cs | 146 +++++++++++++++ src/GameCult.Networking/README.md | 6 +- .../NetworkingTests.cs | 99 ++++++++++ 9 files changed, 486 insertions(+), 10 deletions(-) diff --git a/src/GameCult.Mesh/README.md b/src/GameCult.Mesh/README.md index 725d715d..a926c180 100644 --- a/src/GameCult.Mesh/README.md +++ b/src/GameCult.Mesh/README.md @@ -87,7 +87,9 @@ CultMesh can persist authoritative shard logs through log per shard, so a restarted primary can still serve replica catch-up without depending on warm process memory. When retained log history is compacted, stale replicas get an explicit resync requirement instead of a partial log pretending -to be enough. +to be enough. The replicator can answer that by fetching a shard-bounded +snapshot, replacing the local shard view, and advancing its cursor to the +snapshot's represented log sequence. ## Documentation diff --git a/src/GameCult.Mesh/docs/public-api.md b/src/GameCult.Mesh/docs/public-api.md index db975902..1c4fff05 100644 --- a/src/GameCult.Mesh/docs/public-api.md +++ b/src/GameCult.Mesh/docs/public-api.md @@ -60,7 +60,9 @@ cursors can be stored with `ICultNetShardReplicaCursorStore`. Authoritative shard logs can be persisted with `ICultNetShardMutationLogStore`; the file implementation stores one MessagePack log per shard so catch-up survives a primary restart. Compacted history returns an explicit resync requirement so a -replica can fall back to a snapshot before applying newer log entries. +replica can fall back to a shard-bounded snapshot before applying newer log +entries. Applying that snapshot replaces the local shard view and advances the +replica cursor to the snapshot's represented log sequence. ### Simulation Witness Consensus diff --git a/src/GameCult.Networking/Contracts/cultnet-distributed-database.md b/src/GameCult.Networking/Contracts/cultnet-distributed-database.md index b140d071..d620ca3d 100644 --- a/src/GameCult.Networking/Contracts/cultnet-distributed-database.md +++ b/src/GameCult.Networking/Contracts/cultnet-distributed-database.md @@ -64,6 +64,9 @@ Implemented: implementation - shard-log compaction watermarks; requests for compacted history return `ResyncRequired` with `reason = "compacted_history"` +- shard-bounded snapshots with shard id, epoch, and represented log sequence +- replica snapshot recovery after compacted history, including local shard + replacement and cursor advancement - client-authority scopes for locally predicted input documents - predicted and reconciled database change kinds for client-side prediction - simulation witness observations for mesh-side opinions about frame facts @@ -78,7 +81,7 @@ Not implemented yet: - membership/failure detection - leader election or automatic shard failover -- automatic snapshot fallback after compacted log history +- peer-to-peer snapshot fanout and throttling - peer-to-peer fanout for observation gossip and candidate propagation - rollback/resimulation helpers for simulation frames - declared CRDT merge policies @@ -98,6 +101,8 @@ Not implemented yet: - Log retention is part of the replication contract. A replica asking before the compaction watermark must resync from a snapshot before applying newer log entries. +- A shard snapshot represents a log sequence. Applying it replaces the local + shard view and advances the replica cursor to that sequence. - Realtime subscriptions publish domain changes, not storage envelopes. - Client-owned input documents may be predicted locally only inside explicit `CultNetClientAuthorityScope` declarations. @@ -112,12 +117,11 @@ Not implemented yet: ## Next Coherent Slice -Add automatic snapshot fallback and discovery fanout: +Add discovery fanout and operational polish: -- teach the replication puller to request and apply a shard snapshot after - `compacted_history` - expose operator-facing defaults for durable shard-log paths in CultMesh nodes - add Verse discovery over the schema-v0 mesh lane +- add snapshot throttling and size limits for public mesh edges - after that, add simulation-frame rollback/resimulation helpers around predicted input streams - add peer-to-peer fanout for observation gossip and candidate propagation @@ -128,9 +132,9 @@ restart. Client-owned input documents can now be predicted locally and reconciled when the authoritative log arrives. Nodes can also aggregate witness observations into deterministic consensus candidates for simulation facts like "who shot first," and those reports now have schema-v0 wire contracts. The next -useful layer is making the replication puller recover from compacted history -automatically. +useful layer is making Verse discovery and operator defaults first-class. Membership and leader election should wait until basic replica catch-up is boring. Authoritative mutation logs can now survive process restart through `ICultNetShardMutationLogStore`; compaction now has an honest resync boundary, -and snapshot fallback is the next place where that boundary needs automation. +and the shard replicator can recover from that boundary through snapshot +replacement. diff --git a/src/GameCult.Networking/CultNetDatabase.cs b/src/GameCult.Networking/CultNetDatabase.cs index d4c9a4d0..61db9043 100644 --- a/src/GameCult.Networking/CultNetDatabase.cs +++ b/src/GameCult.Networking/CultNetDatabase.cs @@ -490,6 +490,135 @@ namespace GameCult.Networking return _mutationLogStore?.GetCompactedThrough(shardId) ?? 0; } + /// + /// Gets the latest known shard-log sequence for a shard. + /// + public long GetLatestMutationLogSequence(string shardId) + { + ThrowIfDisposed(); + if (string.IsNullOrWhiteSpace(shardId)) throw new ArgumentException("Value must be non-empty.", nameof(shardId)); + var compactedThrough = GetCompactedMutationLogSequence(shardId); + var retained = _mutationLogStore != null + ? _mutationLogStore.Read(shardId).Select(entry => entry.Sequence) + : GetMutationLog(shardId).Select(entry => entry.Sequence); + return retained.DefaultIfEmpty(compactedThrough).Max(); + } + + /// + /// Creates a raw document snapshot bounded to one shard. + /// + public CultNetSnapshotResponseRawMessage CreateShardSnapshotResponse( + CultNetShardDescriptor shard, + string messageId, + CultNetSnapshotRequestMessage? filter = null) + { + ThrowIfDisposed(); + if (shard == null) throw new ArgumentNullException(nameof(shard)); + + var requestedSchemaIds = filter?.SchemaIds != null + ? new HashSet(filter.SchemaIds, StringComparer.Ordinal) + : null; + var requestedRecordKeys = filter?.RecordKeys != null + ? new HashSet(filter.RecordKeys, StringComparer.Ordinal) + : null; + var documents = new List(); + foreach (var document in _cache.AllEntries) + { + var documentType = document.GetType(); + var descriptor = _cache.Registry.GetRequired(documentType); + var key = GetTrackedKey(document, documentType); + if (string.IsNullOrWhiteSpace(key.Value) || + !shard.Matches(descriptor.SchemaId, key) || + (requestedSchemaIds != null && !requestedSchemaIds.Contains(descriptor.SchemaId)) || + (requestedRecordKeys != null && !requestedRecordKeys.Contains(key.Value))) + { + continue; + } + + documents.Add(CreateRawDocumentRecord(key, document)); + } + + return new CultNetSnapshotResponseRawMessage + { + MessageId = string.IsNullOrWhiteSpace(messageId) ? Guid.NewGuid().ToString("N") : messageId, + Documents = documents.ToArray(), + ShardId = shard.ShardId, + ShardEpoch = shard.Epoch, + ShardLogSequence = GetLatestMutationLogSequence(shard.ShardId) + }; + } + + /// + /// Applies a shard-bounded snapshot and advances the replica cursor to its represented sequence. + /// + public async Task ApplyShardSnapshotResponseAsync( + CultNetShardDescriptor shard, + CultNetSnapshotResponseRawMessage response) + { + ThrowIfDisposed(); + if (shard == null) throw new ArgumentNullException(nameof(shard)); + if (response == null) throw new ArgumentNullException(nameof(response)); + if (!string.IsNullOrWhiteSpace(response.ShardId) && + !string.Equals(response.ShardId, shard.ShardId, StringComparison.Ordinal)) + { + throw new InvalidOperationException($"Snapshot for shard '{response.ShardId}' cannot be applied to shard '{shard.ShardId}'."); + } + + if (response.ShardEpoch.HasValue && response.ShardEpoch.Value != shard.Epoch) + { + throw new CultNetShardAuthorityException( + shard, + $"Shard '{shard.ShardId}' is at epoch {shard.Epoch}, not snapshot epoch {response.ShardEpoch.Value}.", + "stale_epoch"); + } + + var incomingKeys = response.Documents + .Where(document => ShardMatchesRawDocument(shard, document)) + .Select(document => document.RecordKey) + .ToHashSet(StringComparer.Ordinal); + var local = GetLocalShardDocuments(shard).ToArray(); + foreach (var existing in local.Where(item => !incomingKeys.Contains(item.Key.Value))) + { + RemoveTrackedDocument(existing.DocumentType, existing.Key); + PublishUntyped( + existing.DocumentType, + CultNetDatabaseChangeKind.Removed, + existing.Key, + existing.SchemaId, + shard, + document: null, + previousDocument: existing.Document); + } + + foreach (var document in response.Documents.Where(document => ShardMatchesRawDocument(shard, document))) + { + var key = new CultRecordKey(document.RecordKey); + var descriptor = _cache.Registry.GetRequiredBySchemaId(document.SchemaId); + var previous = _cache.Get(key); + var applied = await _documents.ApplyRawDocumentPutMessageAsync( + _cache, + new CultNetDocumentPutRawMessage + { + MessageId = response.MessageId, + Document = document, + ShardId = shard.ShardId, + ShardEpoch = shard.Epoch + }).ConfigureAwait(false); + PublishUntyped( + descriptor.DocumentType, + previous == null ? CultNetDatabaseChangeKind.Added : CultNetDatabaseChangeKind.Updated, + key, + document.SchemaId, + shard, + applied, + previous); + } + + var sequence = response.ShardLogSequence ?? 0; + SetAppliedShardSequence(shard.ShardId, sequence); + return sequence; + } + /// /// Gets the last replicated sequence applied for a shard. /// @@ -914,11 +1043,58 @@ namespace GameCult.Networking : new CultRecordKey(string.Empty); } + private IEnumerable GetLocalShardDocuments(CultNetShardDescriptor shard) + { + foreach (var document in _cache.AllEntries) + { + var documentType = document.GetType(); + var descriptor = _cache.Registry.GetRequired(documentType); + var key = GetTrackedKey(document, documentType); + if (!string.IsNullOrWhiteSpace(key.Value) && shard.Matches(descriptor.SchemaId, key)) + { + yield return new TrackedShardDocument(documentType, descriptor.SchemaId, key, document); + } + } + } + + private bool ShardMatchesRawDocument(CultNetShardDescriptor shard, CultNetRawDocumentRecord document) + { + return document != null && + !string.IsNullOrWhiteSpace(document.SchemaId) && + !string.IsNullOrWhiteSpace(document.RecordKey) && + shard.Matches(document.SchemaId, new CultRecordKey(document.RecordKey)); + } + + private void RemoveTrackedDocument(Type documentType, CultRecordKey key) + { + var removeMethod = typeof(CultCache).GetMethod(nameof(CultCache.Remove))! + .MakeGenericMethod(documentType); + var handleType = typeof(CultRecordHandle<>).MakeGenericType(documentType); + var handle = Activator.CreateInstance(handleType, new object[] { key }); + removeMethod.Invoke(_cache, new[] { handle }); + } + private void Publish(CultNetDatabaseChange change) where T : class { _changes.OnNext(change); } + private sealed class TrackedShardDocument + { + public TrackedShardDocument(Type documentType, string schemaId, CultRecordKey key, object document) + { + DocumentType = documentType; + SchemaId = schemaId; + Key = key; + Document = document; + } + + public Type DocumentType { get; } + public string SchemaId { get; } + public CultRecordKey Key { get; } + public object Document { get; } + } + private void AppendMutationLogEntry( CultNetShardDescriptor shard, CultNetDatabaseChangeKind kind, diff --git a/src/GameCult.Networking/CultNetDatabaseServer.cs b/src/GameCult.Networking/CultNetDatabaseServer.cs index 0e923314..34dd3fcf 100644 --- a/src/GameCult.Networking/CultNetDatabaseServer.cs +++ b/src/GameCult.Networking/CultNetDatabaseServer.cs @@ -82,6 +82,29 @@ namespace GameCult.Networking public CultNetSnapshotResponseRawMessage CreateSnapshotResponse(CultNetSnapshotRequestMessage request) { if (request == null) throw new ArgumentNullException(nameof(request)); + if (!string.IsNullOrWhiteSpace(request.ShardId)) + { + var shard = _database.Shards.FirstOrDefault(candidate => + string.Equals(candidate.ShardId, request.ShardId, StringComparison.Ordinal)); + if (shard == null) + { + throw new InvalidOperationException($"Shard '{request.ShardId}' is not known by this database."); + } + + if (request.ShardEpoch.HasValue && request.ShardEpoch.Value != shard.Epoch) + { + throw new CultNetShardAuthorityException( + shard, + $"Shard '{shard.ShardId}' is at epoch {shard.Epoch}, not request epoch {request.ShardEpoch.Value}.", + "stale_epoch"); + } + + return _database.CreateShardSnapshotResponse( + shard, + string.IsNullOrWhiteSpace(request.MessageId) ? Guid.NewGuid().ToString("N") : request.MessageId, + request); + } + return _database.Documents.CreateRawSnapshotResponse( _database.Cache, string.IsNullOrWhiteSpace(request.MessageId) ? Guid.NewGuid().ToString("N") : request.MessageId, diff --git a/src/GameCult.Networking/CultNetSchemaMessages.cs b/src/GameCult.Networking/CultNetSchemaMessages.cs index 42e51006..e7849193 100644 --- a/src/GameCult.Networking/CultNetSchemaMessages.cs +++ b/src/GameCult.Networking/CultNetSchemaMessages.cs @@ -525,6 +525,14 @@ namespace GameCult.Networking /// Gets or sets the record keys. /// [Key("recordKeys")] public string[]? RecordKeys { get; set; } + /// + /// Gets or sets the shard id requested for a shard-bounded snapshot. + /// + [Key("shardId")] public string? ShardId { get; set; } + /// + /// Gets or sets the expected shard epoch. + /// + [Key("shardEpoch")] public long? ShardEpoch { get; set; } } /// @@ -545,6 +553,18 @@ namespace GameCult.Networking /// Gets or sets the documents. /// [Key("documents")] public CultNetRawDocumentRecord[] Documents { get; set; } = Array.Empty(); + /// + /// Gets or sets the shard id represented by this snapshot. + /// + [Key("shardId")] public string? ShardId { get; set; } + /// + /// Gets or sets the shard epoch represented by this snapshot. + /// + [Key("shardEpoch")] public long? ShardEpoch { get; set; } + /// + /// Gets or sets the shard-log sequence represented by this snapshot. + /// + [Key("shardLogSequence")] public long? ShardLogSequence { get; set; } } /// diff --git a/src/GameCult.Networking/CultNetShardReplication.cs b/src/GameCult.Networking/CultNetShardReplication.cs index 14e614a7..2524e0c2 100644 --- a/src/GameCult.Networking/CultNetShardReplication.cs +++ b/src/GameCult.Networking/CultNetShardReplication.cs @@ -23,6 +23,17 @@ namespace GameCult.Networking int? limit = null); } + /// + /// Fetches shard-bounded snapshots from an authoritative shard owner. + /// + public interface ICultNetShardSnapshotFetcher + { + /// + /// Fetches a shard snapshot. + /// + Task FetchAsync(CultNetShardDescriptor shard); + } + /// /// Persisted replica cursor for one shard. /// @@ -83,6 +94,11 @@ namespace GameCult.Networking /// public ICultNetShardLogFetcher? Fetcher { get; set; } + /// + /// Gets or sets the fetcher used when log history has been compacted. + /// + public ICultNetShardSnapshotFetcher? SnapshotFetcher { get; set; } + /// /// Gets or sets a callback for background pull failures. /// @@ -180,6 +196,21 @@ namespace GameCult.Networking var response = await _options.Fetcher .FetchAsync(shard, afterSequence, _options.BatchSize) .ConfigureAwait(false); + if (response.ResyncRequired && + string.Equals(response.Reason, "compacted_history", StringComparison.Ordinal)) + { + if (_options.SnapshotFetcher == null) + { + throw new InvalidOperationException( + $"Shard '{shard.ShardId}' requires snapshot resync, but no shard snapshot fetcher is configured."); + } + + var snapshot = await _options.SnapshotFetcher.FetchAsync(shard).ConfigureAwait(false); + var snapshotSequence = await _database.ApplyShardSnapshotResponseAsync(shard, snapshot).ConfigureAwait(false); + await WriteCursorAsync(shard, snapshotSequence).ConfigureAwait(false); + return snapshotSequence; + } + var appliedSequence = await _database.ApplyShardLogResponseAsync(response).ConfigureAwait(false); await WriteCursorAsync(shard, appliedSequence).ConfigureAwait(false); return appliedSequence; @@ -456,4 +487,119 @@ namespace GameCult.Networking $"Shard '{shard.ShardId}' does not advertise a primary endpoint."); } } + + /// + /// Options for the schema-v0 client-based shard snapshot fetcher. + /// + public sealed class CultNetSchemaShardSnapshotFetcherOptions + { + /// + /// Gets or sets client security options used to connect to primary endpoints. + /// + public ClientSecurityOptions? Security { get; set; } + + /// + /// Gets or sets how long to wait for a connection before failing. + /// + public TimeSpan ConnectTimeout { get; set; } = TimeSpan.FromSeconds(5); + + /// + /// Gets or sets how long to wait for a snapshot response before failing. + /// + public TimeSpan ResponseTimeout { get; set; } = TimeSpan.FromSeconds(5); + + /// + /// Gets or sets a callback used to customize each ephemeral fetch client. + /// + public Action? ConfigureClient { get; set; } + } + + /// + /// Fetches shard snapshots over the CultNet schema-v0 client transport. + /// + public sealed class CultNetSchemaShardSnapshotFetcher : ICultNetShardSnapshotFetcher + { + private readonly CultNetSchemaShardSnapshotFetcherOptions _options; + + /// + /// Creates a schema-v0 shard snapshot fetcher. + /// + public CultNetSchemaShardSnapshotFetcher(CultNetSchemaShardSnapshotFetcherOptions? options = null) + { + _options = options ?? new CultNetSchemaShardSnapshotFetcherOptions(); + } + + /// + public async Task FetchAsync(CultNetShardDescriptor shard) + { + if (shard == null) throw new ArgumentNullException(nameof(shard)); + var endpoint = ResolvePrimaryEndpoint(shard); + var (host, port) = CultNetSchemaWriteForwarder.ParseEndpoint(endpoint); + var messageId = Guid.NewGuid().ToString("N"); + var completion = new TaskCompletionSource( + TaskCreationOptions.RunContinuationsAsynchronously); + + using var client = new Client(_options.Security ?? ClientSecurityOptions.Development()) + { + AllowUnverifiedCultNetMessages = true + }; + _options.ConfigureClient?.Invoke(client); + client.OnCultNet(response => + { + if (string.Equals(response.MessageId, messageId, StringComparison.Ordinal)) + { + completion.TrySetResult(response); + } + }); + client.OnCultNet(error => + completion.TrySetException(new InvalidOperationException(error.Error))); + + client.Connect(host, port); + await WaitForConnectionAsync(client, endpoint).ConfigureAwait(false); + client.SendCultNet(new CultNetSnapshotRequestMessage + { + MessageId = messageId, + SchemaIds = shard.SchemaIds.Count == 0 ? null : shard.SchemaIds.ToArray(), + ShardId = shard.ShardId, + ShardEpoch = shard.Epoch + }); + + return await WaitForResponseAsync(completion.Task, endpoint).ConfigureAwait(false); + } + + private async Task WaitForConnectionAsync(Client client, string endpoint) + { + var deadline = DateTimeOffset.UtcNow + _options.ConnectTimeout; + while (!client.Connected) + { + if (DateTimeOffset.UtcNow >= deadline) + { + throw new TimeoutException($"Timed out connecting to shard primary endpoint {endpoint}."); + } + + await Task.Delay(TimeSpan.FromMilliseconds(25)).ConfigureAwait(false); + } + } + + private async Task WaitForResponseAsync( + Task responseTask, + string endpoint) + { + var timeoutTask = Task.Delay(_options.ResponseTimeout); + var completed = await Task.WhenAny(responseTask, timeoutTask).ConfigureAwait(false); + if (completed != responseTask) + { + throw new TimeoutException($"Timed out waiting for shard snapshot response from {endpoint}."); + } + + return await responseTask.ConfigureAwait(false); + } + + private static string ResolvePrimaryEndpoint(CultNetShardDescriptor shard) + { + return shard.PrimaryEndpoints.FirstOrDefault() + ?? throw new InvalidOperationException( + $"Shard '{shard.ShardId}' does not advertise a primary endpoint."); + } + } } diff --git a/src/GameCult.Networking/README.md b/src/GameCult.Networking/README.md index d30dab16..44592186 100644 --- a/src/GameCult.Networking/README.md +++ b/src/GameCult.Networking/README.md @@ -162,7 +162,11 @@ MessagePack log file per shard and lets a restarted primary continue answering `cultnet.shard_log_request.v0` from durable wire entries. The same store owns log compaction: once entries are compacted through a sequence, older catch-up requests return `ResyncRequired` with `reason = "compacted_history"` instead of -silently handing a replica a partial history. +silently handing a replica a partial history. `CultNetShardReplicator` can use +an `ICultNetShardSnapshotFetcher` to recover from that response: it fetches a +shard-bounded snapshot, replaces the local shard view, advances the cursor to +the snapshot's represented log sequence, and resumes normal log pulls on later +ticks. For client-side prediction, `CultNetClientAuthorityScope` declares which input documents a runtime may author optimistically. `PutPredictedAsync` updates the diff --git a/tests/GameCult.Networking.Tests/NetworkingTests.cs b/tests/GameCult.Networking.Tests/NetworkingTests.cs index 8ab4a331..6c662033 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -187,6 +187,9 @@ namespace GameCult.Networking.Tests var message = new CultNetSnapshotResponseRawMessage { MessageId = "snapshot-1", + ShardId = "players-eu", + ShardEpoch = 12, + ShardLogSequence = 42, Documents = [ new CultNetRawDocumentRecord @@ -208,6 +211,9 @@ namespace GameCult.Networking.Tests var roundTrip = (CultNetSnapshotResponseRawMessage)CultNetSchemaMessageSerialization.Deserialize(payload); Assert.That(roundTrip.MessageId, Is.EqualTo("snapshot-1")); + Assert.That(roundTrip.ShardId, Is.EqualTo("players-eu")); + Assert.That(roundTrip.ShardEpoch, Is.EqualTo(12)); + Assert.That(roundTrip.ShardLogSequence, Is.EqualTo(42)); Assert.That(roundTrip.Documents, Has.Length.EqualTo(1)); Assert.That(roundTrip.Documents[0].SchemaId, Is.EqualTo("sha256:ghostlight-agent-state")); Assert.That(roundTrip.Documents[0].PayloadEncoding, Is.EqualTo("messagepack")); @@ -1448,6 +1454,79 @@ namespace GameCult.Networking.Tests } } + [Test] + public async Task CultNetShardReplicator_AppliesSnapshot_WhenLogHistoryWasCompacted() + { + 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 shard = new CultNetShardDescriptor( + "players-snapshot-resync", + "runtime-a", + epoch: 9, + isPrimary: false, + schemaIds: [schemaId], + keyPrefix: "player:", + primaryEndpoints: ["cultnet://primary.example.test:3075"]); + var keepKey = new CultRecordKey("player:keep"); + var goneKey = new CultRecordKey("player:gone"); + await cache.UpsertAsync( + new PlayerData { PlayerId = Guid.NewGuid(), Email = "keep-old@example.test", PasswordHash = "hash", Username = "KeepOld" }, + new CultRecordHandle(keepKey)); + await cache.UpsertAsync( + new PlayerData { PlayerId = Guid.NewGuid(), Email = "gone@example.test", PasswordHash = "hash", Username = "Gone" }, + new CultRecordHandle(goneKey)); + var database = new CultNetDatabase(cache, new CultNetDatabaseOptions + { + DocumentRegistry = registry, + Shards = [shard] + }); + var keep = new PlayerData + { + PlayerId = Guid.NewGuid(), + Email = "keep-new@example.test", + PasswordHash = "hash", + Username = "KeepNew" + }; + var put = registry.CreateRawDocumentPutMessage( + "snapshot-keep", + new CultRecordHandle(keepKey), + keep); + var logFetcher = new CapturingShardLogFetcher(new CultNetShardLogResponseMessage + { + ShardId = shard.ShardId, + ShardEpoch = shard.Epoch, + ResyncRequired = true, + Reason = "compacted_history", + CompactedThrough = 2 + }); + var snapshotFetcher = new CapturingShardSnapshotFetcher(new CultNetSnapshotResponseRawMessage + { + MessageId = "snapshot-resync", + ShardId = shard.ShardId, + ShardEpoch = shard.Epoch, + ShardLogSequence = 4, + Documents = [put.Document] + }); + using var replicator = new CultNetShardReplicator(database, new CultNetShardReplicatorOptions + { + Fetcher = logFetcher, + SnapshotFetcher = snapshotFetcher + }); + + var sequence = await replicator.PullOnceAsync(shard); + + Assert.That(sequence, Is.EqualTo(4)); + Assert.That(database.GetAppliedShardSequence(shard.ShardId), Is.EqualTo(4)); + Assert.That(snapshotFetcher.FetchCount, Is.EqualTo(1)); + Assert.That(cache.Get(goneKey), Is.Null); + Assert.That(cache.Get(keepKey)!.Username, Is.EqualTo("KeepNew")); + } + [Test] public void CultNetSimulationConsensus_Builds_QuorumCandidate_FromWitnesses() { @@ -1913,6 +1992,26 @@ namespace GameCult.Networking.Tests } } + private sealed class CapturingShardSnapshotFetcher : ICultNetShardSnapshotFetcher + { + private readonly CultNetSnapshotResponseRawMessage _response; + + public CapturingShardSnapshotFetcher(CultNetSnapshotResponseRawMessage response) + { + _response = response; + } + + public int FetchCount { get; private set; } + public CultNetShardDescriptor? LastShard { get; private set; } + + public Task FetchAsync(CultNetShardDescriptor shard) + { + FetchCount++; + LastShard = shard; + return Task.FromResult(_response); + } + } + [MessagePack.MessagePackObject] public sealed class PlayerDataPayload {