diff --git a/src/GameCult.Mesh/README.md b/src/GameCult.Mesh/README.md index 6d1a725f..725d715d 100644 --- a/src/GameCult.Mesh/README.md +++ b/src/GameCult.Mesh/README.md @@ -85,7 +85,9 @@ mesh opinion layer that can feed an authoritative shard commit path. CultMesh can persist authoritative shard logs through `ICultNetShardMutationLogStore`. The included file store writes one MessagePack log per shard, so a restarted primary can still serve replica catch-up without -depending on warm process memory. +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. ## Documentation diff --git a/src/GameCult.Mesh/docs/public-api.md b/src/GameCult.Mesh/docs/public-api.md index f1c7f390..db975902 100644 --- a/src/GameCult.Mesh/docs/public-api.md +++ b/src/GameCult.Mesh/docs/public-api.md @@ -59,7 +59,8 @@ Shard logs can be requested over schema-v0 and applied by replicas. Replica 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. +primary restart. Compacted history returns an explicit resync requirement so a +replica can fall back to a snapshot before applying newer log entries. ### Simulation Witness Consensus diff --git a/src/GameCult.Networking/Contracts/cultnet-distributed-database.md b/src/GameCult.Networking/Contracts/cultnet-distributed-database.md index b8f254e7..b140d071 100644 --- a/src/GameCult.Networking/Contracts/cultnet-distributed-database.md +++ b/src/GameCult.Networking/Contracts/cultnet-distributed-database.md @@ -62,6 +62,8 @@ Implemented: - durable authoritative shard-log storage behind `ICultNetShardMutationLogStore`, including a per-shard MessagePack file implementation +- shard-log compaction watermarks; requests for compacted history return + `ResyncRequired` with `reason = "compacted_history"` - 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 @@ -76,8 +78,7 @@ Not implemented yet: - membership/failure detection - leader election or automatic shard failover -- log compaction -- snapshot fallback for compacted log history +- automatic snapshot fallback after compacted log history - peer-to-peer fanout for observation gossip and candidate propagation - rollback/resimulation helpers for simulation frames - declared CRDT merge policies @@ -94,6 +95,9 @@ Not implemented yet: bypass shard policy. - The accepted shard log is stored in replica catch-up wire form. Object logs are for local inspection; wire logs are the durable replication authority. +- 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. - Realtime subscriptions publish domain changes, not storage envelopes. - Client-owned input documents may be predicted locally only inside explicit `CultNetClientAuthorityScope` declarations. @@ -108,10 +112,10 @@ Not implemented yet: ## Next Coherent Slice -Add compaction-aware snapshot fallback and discovery fanout: +Add automatic snapshot fallback and discovery fanout: -- define the compaction boundary that turns old log reads into snapshot reads -- return `ResyncRequired` when a replica asks for compacted history +- 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 - after that, add simulation-frame rollback/resimulation helpers around @@ -124,8 +128,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 authoritative mutation logs durable and compaction-aware. +useful layer is making the replication puller recover from compacted history +automatically. Membership and leader election should wait until basic replica catch-up is boring. Authoritative mutation logs can now survive process restart through -`ICultNetShardMutationLogStore`; compaction and snapshot fallback are the next -places where the machine needs an honest boundary. +`ICultNetShardMutationLogStore`; compaction now has an honest resync boundary, +and snapshot fallback is the next place where that boundary needs automation. diff --git a/src/GameCult.Networking/CultNetDatabase.cs b/src/GameCult.Networking/CultNetDatabase.cs index 8acb55f9..d4c9a4d0 100644 --- a/src/GameCult.Networking/CultNetDatabase.cs +++ b/src/GameCult.Networking/CultNetDatabase.cs @@ -480,6 +480,16 @@ namespace GameCult.Networking .ToArray(); } + /// + /// Gets the highest retained-log sequence that has been compacted away for a shard. + /// + public long GetCompactedMutationLogSequence(string shardId) + { + ThrowIfDisposed(); + if (string.IsNullOrWhiteSpace(shardId)) throw new ArgumentException("Value must be non-empty.", nameof(shardId)); + return _mutationLogStore?.GetCompactedThrough(shardId) ?? 0; + } + /// /// Gets the last replicated sequence applied for a shard. /// diff --git a/src/GameCult.Networking/CultNetDatabaseServer.cs b/src/GameCult.Networking/CultNetDatabaseServer.cs index 321d1c8e..0e923314 100644 --- a/src/GameCult.Networking/CultNetDatabaseServer.cs +++ b/src/GameCult.Networking/CultNetDatabaseServer.cs @@ -132,6 +132,20 @@ namespace GameCult.Networking }; } + var compactedThrough = _database.GetCompactedMutationLogSequence(request.ShardId); + if (request.AfterSequence < compactedThrough) + { + return new CultNetShardLogResponseMessage + { + MessageId = RequestMessageId(request.MessageId), + ShardId = request.ShardId, + ShardEpoch = shard.Epoch, + ResyncRequired = true, + Reason = "compacted_history", + CompactedThrough = compactedThrough + }; + } + var entries = _database.GetMutationLogMessages(request.ShardId, request.AfterSequence, request.Limit) .ToArray(); diff --git a/src/GameCult.Networking/CultNetSchemaMessages.cs b/src/GameCult.Networking/CultNetSchemaMessages.cs index 8cbac055..42e51006 100644 --- a/src/GameCult.Networking/CultNetSchemaMessages.cs +++ b/src/GameCult.Networking/CultNetSchemaMessages.cs @@ -793,6 +793,10 @@ namespace GameCult.Networking /// Gets or sets a machine-readable resync reason. /// [Key("reason")] public string? Reason { get; set; } + /// + /// Gets or sets the highest sequence compacted out of the retained log. + /// + [Key("compactedThrough")] public long? CompactedThrough { get; set; } } /// diff --git a/src/GameCult.Networking/CultNetShardMutationLogStorage.cs b/src/GameCult.Networking/CultNetShardMutationLogStorage.cs index 231c43df..5b3a9c32 100644 --- a/src/GameCult.Networking/CultNetShardMutationLogStorage.cs +++ b/src/GameCult.Networking/CultNetShardMutationLogStorage.cs @@ -21,6 +21,16 @@ namespace GameCult.Networking /// Appends or replaces one accepted log entry for a shard. /// void Append(string shardId, CultNetShardLogEntryMessage entry); + + /// + /// Gets the highest sequence that has been compacted out of the retained log. + /// + long GetCompactedThrough(string shardId); + + /// + /// Removes retained entries at or before the supplied sequence. + /// + void CompactThrough(string shardId, long sequence); } /// @@ -80,6 +90,43 @@ namespace GameCult.Networking } } + /// + public long GetCompactedThrough(string shardId) + { + if (string.IsNullOrWhiteSpace(shardId)) throw new ArgumentException("Value must be non-empty.", nameof(shardId)); + + lock (_gate) + { + return ReadMetadata(shardId).CompactedThrough; + } + } + + /// + public void CompactThrough(string shardId, long sequence) + { + if (string.IsNullOrWhiteSpace(shardId)) throw new ArgumentException("Value must be non-empty.", nameof(shardId)); + if (sequence < 0) throw new ArgumentOutOfRangeException(nameof(sequence), "Sequence must be non-negative."); + + lock (_gate) + { + var metadata = ReadMetadata(shardId); + if (sequence <= metadata.CompactedThrough) + { + return; + } + + var retained = ReadAll(shardId) + .Where(entry => entry.Sequence > sequence) + .OrderBy(entry => entry.Sequence) + .ToArray(); + File.WriteAllBytes( + GetShardPath(shardId), + MessagePackSerializer.Serialize(retained, CultNetSchemaMessageSerialization.Options)); + metadata.CompactedThrough = sequence; + WriteMetadata(shardId, metadata); + } + } + private CultNetShardLogEntryMessage[] ReadAll(string shardId) { var path = GetShardPath(shardId); @@ -99,11 +146,42 @@ namespace GameCult.Networking return Path.Combine(_rootPath, HashShardId(shardId) + ".mpack"); } + private string GetMetadataPath(string shardId) + { + return Path.Combine(_rootPath, HashShardId(shardId) + ".meta.mpack"); + } + + private CultNetShardMutationLogMetadata ReadMetadata(string shardId) + { + var path = GetMetadataPath(shardId); + if (!File.Exists(path)) + { + return new CultNetShardMutationLogMetadata(); + } + + var payload = File.ReadAllBytes(path); + return MessagePackSerializer.Deserialize( + payload, + CultNetSchemaMessageSerialization.Options) ?? new CultNetShardMutationLogMetadata(); + } + + private void WriteMetadata(string shardId, CultNetShardMutationLogMetadata metadata) + { + var payload = MessagePackSerializer.Serialize(metadata, CultNetSchemaMessageSerialization.Options); + File.WriteAllBytes(GetMetadataPath(shardId), payload); + } + private static string HashShardId(string shardId) { using var sha = SHA256.Create(); var hash = sha.ComputeHash(System.Text.Encoding.UTF8.GetBytes(shardId)); return string.Concat(hash.Select(value => value.ToString("x2"))); } + + [MessagePackObject(AllowPrivate = true)] + internal sealed class CultNetShardMutationLogMetadata + { + [Key("compactedThrough")] public long CompactedThrough { get; set; } + } } } diff --git a/src/GameCult.Networking/README.md b/src/GameCult.Networking/README.md index d80b2013..d30dab16 100644 --- a/src/GameCult.Networking/README.md +++ b/src/GameCult.Networking/README.md @@ -159,7 +159,10 @@ those cursors in one local MessagePack file so replicas can resume after a restart. Authoritative shard logs can be stored through `ICultNetShardMutationLogStore`; `CultNetFileShardMutationLogStore` keeps one MessagePack log file per shard and lets a restarted primary continue answering -`cultnet.shard_log_request.v0` from durable wire entries. +`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. 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 3fc31720..8ab4a331 100644 --- a/tests/GameCult.Networking.Tests/NetworkingTests.cs +++ b/tests/GameCult.Networking.Tests/NetworkingTests.cs @@ -285,6 +285,7 @@ namespace GameCult.Networking.Tests MessageId = "log-1", ShardId = "players-eu", ShardEpoch = 12, + CompactedThrough = 40, Entries = [ new CultNetShardLogEntryMessage @@ -314,6 +315,7 @@ namespace GameCult.Networking.Tests Assert.That(roundTrip.MessageId, Is.EqualTo("log-1")); Assert.That(roundTrip.ShardId, Is.EqualTo("players-eu")); Assert.That(roundTrip.ShardEpoch, Is.EqualTo(12)); + Assert.That(roundTrip.CompactedThrough, Is.EqualTo(40)); Assert.That(roundTrip.Entries, Has.Length.EqualTo(1)); Assert.That(roundTrip.Entries[0].Sequence, Is.EqualTo(42)); Assert.That(roundTrip.Entries[0].Put, Is.Not.Null); @@ -1022,6 +1024,66 @@ namespace GameCult.Networking.Tests Assert.That(response.Entries[0].Put!.Document.Payload, Is.EqualTo(SerializePlayerDataPayload(player))); } + [Test] + public async Task CultNetDatabaseServer_RequiresResync_ForCompactedShardLogHistory() + { + var rootPath = Path.Combine(TestContext.CurrentContext.WorkDirectory, "shard-logs", Guid.NewGuid().ToString("N")); + 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 store = new CultNetFileShardMutationLogStore(rootPath); + var database = new CultNetDatabase(cache, new CultNetDatabaseOptions + { + DocumentRegistry = registry, + MutationLogStore = store, + Shards = + [ + new CultNetShardDescriptor( + "players-compacted-log", + "runtime-a", + epoch: 8, + isPrimary: true, + schemaIds: [schemaId], + keyPrefix: "player:") + ] + }); + using var server = new Server(cache, DevelopmentServerSecurity); + using var databaseServer = new CultNetDatabaseServer(server, database); + + await database.PutAsync( + new CultRecordKey("player:first"), + new PlayerData { PlayerId = Guid.NewGuid(), Email = "first@example.test", PasswordHash = "hash", Username = "First" }); + await database.PutAsync( + new CultRecordKey("player:second"), + new PlayerData { PlayerId = Guid.NewGuid(), Email = "second@example.test", PasswordHash = "hash", Username = "Second" }); + store.CompactThrough("players-compacted-log", 1); + + var staleResponse = databaseServer.CreateShardLogResponse(new CultNetShardLogRequestMessage + { + ShardId = "players-compacted-log", + ShardEpoch = 8, + AfterSequence = 0 + }); + var currentResponse = databaseServer.CreateShardLogResponse(new CultNetShardLogRequestMessage + { + ShardId = "players-compacted-log", + ShardEpoch = 8, + AfterSequence = 1 + }); + + Assert.That(staleResponse.ResyncRequired, Is.True); + Assert.That(staleResponse.Reason, Is.EqualTo("compacted_history")); + Assert.That(staleResponse.CompactedThrough, Is.EqualTo(1)); + Assert.That(staleResponse.Entries, Is.Empty); + Assert.That(currentResponse.ResyncRequired, Is.False); + Assert.That(currentResponse.Entries, Has.Length.EqualTo(1)); + Assert.That(currentResponse.Entries[0].Sequence, Is.EqualTo(2)); + } + [Test] public async Task CultNetDatabase_Applies_ShardLogResponse_ToReplica() {