From e32f3c51879df4181a6bf8e2857935c8ec7a99ce Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 15 Jul 2026 08:27:19 +0100 Subject: [PATCH] Add incremental retained-session persistence --- README.md | 8 +- docs/architecture.md | 17 ++- docs/retained-session-contract.md | 19 ++- src/Ymir.Core/RetainedSessionModel.cs | 30 ++++ src/Ymir.Core/YmirSession.cs | 82 +++++++++++ src/Ymir.Core/YmirSessionCheckpointCodec.cs | 148 ++++++++++++++++++-- tests/Ymir.Tests/YmirSessionTests.cs | 42 ++++++ 7 files changed, 326 insertions(+), 20 deletions(-) diff --git a/README.md b/README.md index e71e3b6..9392c11 100644 --- a/README.md +++ b/README.md @@ -85,8 +85,12 @@ embeds one session per run and zone; its old process-wide simulator objects are not session identifiers. Revision-checked retained-session circle overlap and cast queries observe the current stable-id body projection without mutating it. Replay-backed checkpoint reconstruction is -implemented; host-owned incremental journal persistence and non-Windows RID -artifacts remain cutover work. +implemented. `CapturePersistence` emits only the journal suffix not yet stored +by its host plus a bounded resume descriptor; restore verifies contiguous +chunks, the complete journal digest, the projected world, and active contact +episodes before the session can emit facts. Aetheria stores those records in a +daemon-private CultCache and uses its hard-flushed public frame as the commit +marker. Journal compaction and non-Windows RID artifacts remain cutover work. See [the architecture map](docs/architecture.md) and [the executable parity contract](docs/box3d-parity.md). The diff --git a/docs/architecture.md b/docs/architecture.md index 0a3ee43..d92f186 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -56,13 +56,16 @@ step, and disposal operations. It emits revisioned receipts and typed Begin/Hit/End facts with stable Ymir identities. Ordinary retained steps do not accept complete body snapshots. -The embedded retained-session library now has a replay-backed reconstruction -checkpoint and durable-in-checkpoint command receipt ledger. Host-owned private -incremental journal storage and atomic frame/resume commits remain integration -work. Complete replay history is not product state and must not ride in public -client frames. The current CLI still has no named session registry or CultMesh -command lowerings; its CultCache and Eve output remains a regenerated -diagnostic projection. +The embedded retained-session library has replay-backed reconstruction, +durable command receipts, incremental journal-suffix capture, and bounded +resume descriptors. Aetheria stores these in a second daemon-private CultCache: +immutable chunks are flushed first, immutable per-frame resumes second, and the +public Aetheria frame last. The public frame is the only commit marker. Orphaned +private records cannot advance gameplay truth, and restart fails closed once +the persistence capability has been activated. Complete replay history is not +product state and never rides in public client frames. The current CLI still +has no named session registry or CultMesh command lowerings; its CultCache and +Eve output remains a regenerated diagnostic projection. ## Target Inputs diff --git a/docs/retained-session-contract.md b/docs/retained-session-contract.md index 9f2ad69..30ff353 100644 --- a/docs/retained-session-contract.md +++ b/docs/retained-session-contract.md @@ -101,6 +101,20 @@ world verifier, and active episodes. ABI v5 independently validates the exact pinned native build id before opening a session. It contains no native Box3D handle or contact key. +For incremental host persistence, +`gamecult.ymir.session_journal_chunk.v1` carries a contiguous command-journal +suffix and `gamecult.ymir.session_resume.v1` carries the bounded reconstruction +boundary. `CapturePersistence(persistedJournalEntryCount)` never repeats an +already acknowledged prefix. Restore orders chunks by their first entry, +requires exact coverage from zero through the descriptor count, and then uses +the same full replay and verifier path as a complete checkpoint. Both binary +forms are independently checksummed. + +The cursor is currently an index into a complete in-memory journal. Ymir does +not yet prune or renumber that journal; storage writes are incremental, memory +retention is not. Compaction requires a generation/base-sequence contract and +must not be inferred from a host acknowledging a write. + ## Forbidden Writers - recurring complete-body synchronization after restore; @@ -146,8 +160,9 @@ is derived process state: it never appears in a public fact or checkpoint. 3. Revision-checked retained-session overlap and cast queries: done. 4. One embedded retained world and physical-payload session per Aetheria run and zone: done. -5. Persist replay journals privately and incrementally behind a bounded resume - descriptor; complete history must not ride in public product frames. +5. Incremental journal chunks and bounded resume descriptors: done. Aetheria + stores them in a daemon-private CultCache; complete history does not ride in + public product frames. Journal compaction remains separate work. 6. For optional standalone hosting, add a session registry, generation-bearing routed commands, idempotent Create, durable receipts, and CultMesh lowering. diff --git a/src/Ymir.Core/RetainedSessionModel.cs b/src/Ymir.Core/RetainedSessionModel.cs index a06eaa8..cac3399 100644 --- a/src/Ymir.Core/RetainedSessionModel.cs +++ b/src/Ymir.Core/RetainedSessionModel.cs @@ -10,10 +10,40 @@ public sealed record YmirSessionCreateRequest( public static class YmirSessionCheckpointContract { public const string FormatId = "gamecult.ymir.session_checkpoint.replay.v1"; + public const string JournalChunkFormatId = "gamecult.ymir.session_journal_chunk.v1"; + public const string ResumeDescriptorFormatId = "gamecult.ymir.session_resume.v1"; public static string RuntimeFingerprint => $"gamecult.ymir.replay.v1|{Box3DRuntime.ValidatedBuildId}"; } +public sealed record YmirSessionJournalChunk( + string FormatId, + string RuntimeFingerprint, + string SessionId, + string SessionGeneration, + long FirstEntryIndex, + IReadOnlyList Entries); + +public sealed record YmirSessionResumeDescriptor( + string FormatId, + string RuntimeFingerprint, + string SessionId, + string SessionGeneration, + long Revision, + long StepIndex, + float Time, + long NextBodyGeneration, + IReadOnlyList InitialBodies, + float InitialTime, + long JournalEntryCount, + string JournalDigest, + YmirWorld World, + IReadOnlyList ActiveContactEpisodes); + +public sealed record YmirSessionPersistenceCapture( + YmirSessionJournalChunk? JournalChunk, + YmirSessionResumeDescriptor ResumeDescriptor); + public sealed record YmirSessionCheckpoint( string FormatId, string RuntimeFingerprint, diff --git a/src/Ymir.Core/YmirSession.cs b/src/Ymir.Core/YmirSession.cs index 350dbac..05b1a98 100644 --- a/src/Ymir.Core/YmirSession.cs +++ b/src/Ymir.Core/YmirSession.cs @@ -93,6 +93,49 @@ public sealed class YmirSession : IDisposable } } + public static YmirSession Restore( + YmirSessionResumeDescriptor descriptor, + IReadOnlyList chunks) + { + ArgumentNullException.ThrowIfNull(descriptor); + ArgumentNullException.ThrowIfNull(chunks); + if (!string.Equals(descriptor.FormatId, YmirSessionCheckpointContract.ResumeDescriptorFormatId, StringComparison.Ordinal) || + !string.Equals(descriptor.RuntimeFingerprint, YmirSessionCheckpointContract.RuntimeFingerprint, StringComparison.Ordinal) || + descriptor.JournalEntryCount < 0) + throw new InvalidOperationException("Ymir resume descriptor format or runtime provenance is incompatible."); + + var journal = new List(); + foreach (var chunk in chunks.OrderBy(value => value.FirstEntryIndex)) + { + if (!string.Equals(chunk.FormatId, YmirSessionCheckpointContract.JournalChunkFormatId, StringComparison.Ordinal) || + !string.Equals(chunk.RuntimeFingerprint, descriptor.RuntimeFingerprint, StringComparison.Ordinal) || + !string.Equals(chunk.SessionId, descriptor.SessionId, StringComparison.Ordinal) || + !string.Equals(chunk.SessionGeneration, descriptor.SessionGeneration, StringComparison.Ordinal) || + chunk.FirstEntryIndex != journal.Count || chunk.Entries == null) + throw new InvalidOperationException("Ymir journal chunks are incompatible, overlapping, or incomplete."); + journal.AddRange(chunk.Entries); + } + if (journal.Count != descriptor.JournalEntryCount) + throw new InvalidOperationException( + $"Ymir resume requires {descriptor.JournalEntryCount} journal entries but received {journal.Count}."); + + return Restore(new YmirSessionCheckpoint( + YmirSessionCheckpointContract.FormatId, + descriptor.RuntimeFingerprint, + descriptor.SessionId, + descriptor.SessionGeneration, + descriptor.Revision, + descriptor.StepIndex, + descriptor.Time, + descriptor.NextBodyGeneration, + descriptor.InitialBodies, + descriptor.InitialTime, + journal, + descriptor.JournalDigest, + descriptor.World, + descriptor.ActiveContactEpisodes)); + } + public YmirSessionInfo Info { get @@ -140,6 +183,45 @@ public sealed class YmirSession : IDisposable } } + public YmirSessionPersistenceCapture CapturePersistence(long persistedJournalEntryCount) + { + lock (_gate) + { + ObjectDisposedException.ThrowIf(_disposed, this); + if (_faulted) + throw new InvalidOperationException("A faulted Ymir session cannot publish persistence state."); + if (persistedJournalEntryCount < 0 || persistedJournalEntryCount > _journal.Count) + throw new ArgumentOutOfRangeException(nameof(persistedJournalEntryCount)); + + var checkpoint = Checkpoint(); + var chunk = persistedJournalEntryCount == _journal.Count + ? null + : new YmirSessionJournalChunk( + YmirSessionCheckpointContract.JournalChunkFormatId, + checkpoint.RuntimeFingerprint, + checkpoint.SessionId, + checkpoint.SessionGeneration, + persistedJournalEntryCount, + _journal.Skip(checked((int)persistedJournalEntryCount)).ToArray()); + var descriptor = new YmirSessionResumeDescriptor( + YmirSessionCheckpointContract.ResumeDescriptorFormatId, + checkpoint.RuntimeFingerprint, + checkpoint.SessionId, + checkpoint.SessionGeneration, + checkpoint.Revision, + checkpoint.StepIndex, + checkpoint.Time, + checkpoint.NextBodyGeneration, + checkpoint.InitialBodies, + checkpoint.InitialTime, + checkpoint.Journal.Count, + checkpoint.JournalDigest, + checkpoint.World, + checkpoint.ActiveContactEpisodes); + return new YmirSessionPersistenceCapture(chunk, descriptor); + } + } + public YmirCommandReceipt Spawn(YmirSpawnBodyCommand command) { ArgumentNullException.ThrowIfNull(command); diff --git a/src/Ymir.Core/YmirSessionCheckpointCodec.cs b/src/Ymir.Core/YmirSessionCheckpointCodec.cs index 1514cf1..049f63d 100644 --- a/src/Ymir.Core/YmirSessionCheckpointCodec.cs +++ b/src/Ymir.Core/YmirSessionCheckpointCodec.cs @@ -6,6 +6,8 @@ namespace Ymir.Core; public static class YmirSessionCheckpointCodec { private const uint Magic = 0x43534D59; // YMSC + private const uint JournalMagic = 0x4A534D59; // YMSJ + private const uint ResumeMagic = 0x52534D59; // YMSR private const int Version = 1; private const int MaximumPayloadBytes = 256 * 1024 * 1024; private const int MaximumCollectionCount = 4 * 1024 * 1024; @@ -43,17 +45,10 @@ public static class YmirSessionCheckpointCodec public static YmirSessionCheckpoint Decode(ReadOnlySpan payload) { - if (payload.Length <= ChecksumBytes || payload.Length > MaximumPayloadBytes) - throw new InvalidDataException("Ymir checkpoint payload size is invalid."); - var content = payload[..^ChecksumBytes]; - var expectedChecksum = payload[^ChecksumBytes..]; - Span actualChecksum = stackalloc byte[ChecksumBytes]; - SHA256.HashData(content, actualChecksum); - if (!CryptographicOperations.FixedTimeEquals(actualChecksum, expectedChecksum)) - throw new InvalidDataException("Ymir checkpoint checksum is invalid."); + var content = VerifiedContent(payload); try { - using var stream = new MemoryStream(content.ToArray(), writable: false); + using var stream = new MemoryStream(content, writable: false); using var reader = new BinaryReader(stream, Encoding.UTF8, leaveOpen: true); if (reader.ReadUInt32() != Magic || reader.ReadInt32() != Version) throw new InvalidDataException("Ymir checkpoint codec header is incompatible."); @@ -86,6 +81,141 @@ public static class YmirSessionCheckpointCodec } } + public static byte[] Encode(YmirSessionJournalChunk chunk) + { + ArgumentNullException.ThrowIfNull(chunk); + return EncodeChecksummed(writer => + { + writer.Write(JournalMagic); + writer.Write(Version); + WriteString(writer, chunk.FormatId); + WriteString(writer, chunk.RuntimeFingerprint); + WriteString(writer, chunk.SessionId); + WriteString(writer, chunk.SessionGeneration); + writer.Write(chunk.FirstEntryIndex); + WriteJournal(writer, chunk.Entries); + }); + } + + public static YmirSessionJournalChunk DecodeJournalChunk(ReadOnlySpan payload) + { + var content = VerifiedContent(payload); + try + { + using var stream = new MemoryStream(content, writable: false); + using var reader = new BinaryReader(stream, Encoding.UTF8, leaveOpen: true); + if (reader.ReadUInt32() != JournalMagic || reader.ReadInt32() != Version) + throw new InvalidDataException("Ymir journal chunk codec header is incompatible."); + var chunk = new YmirSessionJournalChunk( + ReadString(reader), + ReadString(reader), + ReadString(reader), + ReadString(reader), + reader.ReadInt64(), + ReadJournal(reader)); + if (stream.Position != stream.Length) + throw new InvalidDataException("Ymir journal chunk contains trailing data."); + return chunk; + } + catch (InvalidDataException) + { + throw; + } + catch (Exception error) when (error is EndOfStreamException or IOException or ArgumentException or OverflowException) + { + throw new InvalidDataException("Ymir journal chunk payload is corrupt.", error); + } + } + + public static byte[] Encode(YmirSessionResumeDescriptor descriptor) + { + ArgumentNullException.ThrowIfNull(descriptor); + return EncodeChecksummed(writer => + { + writer.Write(ResumeMagic); + writer.Write(Version); + WriteString(writer, descriptor.FormatId); + WriteString(writer, descriptor.RuntimeFingerprint); + WriteString(writer, descriptor.SessionId); + WriteString(writer, descriptor.SessionGeneration); + writer.Write(descriptor.Revision); + writer.Write(descriptor.StepIndex); + writer.Write(descriptor.Time); + writer.Write(descriptor.NextBodyGeneration); + WriteBodies(writer, descriptor.InitialBodies); + writer.Write(descriptor.InitialTime); + writer.Write(descriptor.JournalEntryCount); + WriteString(writer, descriptor.JournalDigest); + WriteWorld(writer, descriptor.World); + WriteEpisodes(writer, descriptor.ActiveContactEpisodes); + }); + } + + public static YmirSessionResumeDescriptor DecodeResumeDescriptor(ReadOnlySpan payload) + { + var content = VerifiedContent(payload); + try + { + using var stream = new MemoryStream(content, writable: false); + using var reader = new BinaryReader(stream, Encoding.UTF8, leaveOpen: true); + if (reader.ReadUInt32() != ResumeMagic || reader.ReadInt32() != Version) + throw new InvalidDataException("Ymir resume descriptor codec header is incompatible."); + var descriptor = new YmirSessionResumeDescriptor( + ReadString(reader), + ReadString(reader), + ReadString(reader), + ReadString(reader), + reader.ReadInt64(), + reader.ReadInt64(), + reader.ReadSingle(), + reader.ReadInt64(), + ReadBodies(reader), + reader.ReadSingle(), + reader.ReadInt64(), + ReadString(reader), + ReadWorld(reader), + ReadEpisodes(reader)); + if (stream.Position != stream.Length) + throw new InvalidDataException("Ymir resume descriptor contains trailing data."); + return descriptor; + } + catch (InvalidDataException) + { + throw; + } + catch (Exception error) when (error is EndOfStreamException or IOException or ArgumentException or OverflowException) + { + throw new InvalidDataException("Ymir resume descriptor payload is corrupt.", error); + } + } + + private static byte[] EncodeChecksummed(Action write) + { + using var stream = new MemoryStream(); + using (var writer = new BinaryWriter(stream, Encoding.UTF8, leaveOpen: true)) + { + write(writer); + writer.Flush(); + } + if (stream.Length > MaximumPayloadBytes - ChecksumBytes) + throw new InvalidOperationException("Ymir persistence payload exceeds the supported size."); + var content = stream.ToArray(); + return [.. content, .. SHA256.HashData(content)]; + } + + private static byte[] VerifiedContent(ReadOnlySpan payload) + { + if (payload.Length <= ChecksumBytes || payload.Length > MaximumPayloadBytes) + throw new InvalidDataException("Ymir persistence payload size is invalid."); + var content = payload[..^ChecksumBytes]; + var expectedChecksum = payload[^ChecksumBytes..]; + Span actualChecksum = stackalloc byte[ChecksumBytes]; + SHA256.HashData(content, actualChecksum); + if (!CryptographicOperations.FixedTimeEquals(actualChecksum, expectedChecksum)) + throw new InvalidDataException("Ymir persistence payload checksum is invalid."); + return content.ToArray(); + } + private static void WriteJournal(BinaryWriter writer, IReadOnlyList entries) { WriteCount(writer, entries?.Count ?? -1); diff --git a/tests/Ymir.Tests/YmirSessionTests.cs b/tests/Ymir.Tests/YmirSessionTests.cs index 7e1fcc8..3eda6a1 100644 --- a/tests/Ymir.Tests/YmirSessionTests.cs +++ b/tests/Ymir.Tests/YmirSessionTests.cs @@ -156,6 +156,48 @@ public sealed class YmirSessionTests YmirSessionCheckpointCodec.Decode(YmirSessionCheckpointCodec.Encode(checkpoint).Concat(new byte[] { 1 }).ToArray())); } + [Fact] + public void IncrementalPersistenceWritesOnlyNewJournalEntriesAndRestoresBoundary() + { + using var original = YmirSession.Create(new YmirSessionCreateRequest( + "incremental-contact", + [Body("ship", positionX: 0.0f, isStatic: true), Body("pickup", positionX: 1.5f)])); + var began = original.Step(Step("begin", 0, 0.01f)); + var begin = Assert.Single(began.ContactFacts, fact => fact.Kind == YmirContactFactKind.Begin); + + var first = original.CapturePersistence(0); + var firstChunk = Assert.IsType(first.JournalChunk); + Assert.Equal(0, firstChunk.FirstEntryIndex); + Assert.Single(firstChunk.Entries); + var firstChunkBytes = YmirSessionCheckpointCodec.Encode(firstChunk); + var firstDescriptorBytes = YmirSessionCheckpointCodec.Encode(first.ResumeDescriptor); + + var persistent = original.Step(Step("persistent", 1, 0.01f)); + Assert.DoesNotContain(persistent.ContactFacts, fact => fact.Kind == YmirContactFactKind.Begin); + var second = original.CapturePersistence(first.ResumeDescriptor.JournalEntryCount); + var secondChunk = Assert.IsType(second.JournalChunk); + Assert.Equal(1, secondChunk.FirstEntryIndex); + Assert.Single(secondChunk.Entries); + + var chunks = new[] + { + YmirSessionCheckpointCodec.DecodeJournalChunk(firstChunkBytes), + YmirSessionCheckpointCodec.DecodeJournalChunk(YmirSessionCheckpointCodec.Encode(secondChunk)) + }; + var descriptor = YmirSessionCheckpointCodec.DecodeResumeDescriptor( + YmirSessionCheckpointCodec.Encode(second.ResumeDescriptor)); + using var restored = YmirSession.Restore(descriptor, chunks); + restored.Teleport(new YmirTeleportBodyCommand( + Header("separate", 2), "pickup", new Vec2(10.0f, 0.0f), new Vec2(0.0f, 1.0f))); + var ended = restored.Step(Step("end", 3, 0.01f)); + Assert.Equal(begin.ContactId, Assert.Single(ended.ContactFacts, fact => fact.Kind == YmirContactFactKind.End).ContactId); + + Assert.Throws(() => YmirSession.Restore(descriptor, [chunks[1]])); + var corruptDescriptor = firstDescriptorBytes.ToArray(); + corruptDescriptor[24] ^= 0xff; + Assert.Throws(() => YmirSessionCheckpointCodec.DecodeResumeDescriptor(corruptDescriptor)); + } + [Fact] public void RemoveAndRecreateChangesBodyGeneration() {