Add shard log compaction boundary

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
 

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
 

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.

src/GameCult.Networking/CultNetDatabase.cs

@@ -480,6 +480,16 @@ namespace GameCult.Networking
                 .ToArray();
         }
 
+        /// <summary>
+        /// Gets the highest retained-log sequence that has been compacted away for a shard.
+        /// </summary>
+        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;
+        }
+
         /// <summary>
         /// Gets the last replicated sequence applied for a shard.
         /// </summary>

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();
 

src/GameCult.Networking/CultNetSchemaMessages.cs

@@ -793,6 +793,10 @@ namespace GameCult.Networking
         /// Gets or sets a machine-readable resync reason.
         /// </summary>
         [Key("reason")] public string? Reason { get; set; }
+        /// <summary>
+        /// Gets or sets the highest sequence compacted out of the retained log.
+        /// </summary>
+        [Key("compactedThrough")] public long? CompactedThrough { get; set; }
     }
 
     /// <summary>

src/GameCult.Networking/CultNetShardMutationLogStorage.cs

@@ -21,6 +21,16 @@ namespace GameCult.Networking
         /// Appends or replaces one accepted log entry for a shard.
         /// </summary>
         void Append(string shardId, CultNetShardLogEntryMessage entry);
+
+        /// <summary>
+        /// Gets the highest sequence that has been compacted out of the retained log.
+        /// </summary>
+        long GetCompactedThrough(string shardId);
+
+        /// <summary>
+        /// Removes retained entries at or before the supplied sequence.
+        /// </summary>
+        void CompactThrough(string shardId, long sequence);
     }
 
     /// <summary>
@@ -80,6 +90,43 @@ namespace GameCult.Networking
             }
         }
 
+        /// <inheritdoc />
+        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;
+            }
+        }
+
+        /// <inheritdoc />
+        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<CultNetShardMutationLogMetadata>(
+                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; }
+        }
     }
 }

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

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<PlayerData>(
+                    cache.Registry,
+                    payloadSerializer: SerializePlayerDataPayload,
+                    payloadDeserializer: DeserializePlayerDataPayload));
+            var schemaId = cache.Registry.GetRequired<PlayerData>().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()
         {