From 002ade7a728b5ece9b870f4e15cec2cd5be99369 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 15 Jul 2026 08:27:36 +0100 Subject: [PATCH] Persist Ymir sessions behind daemon frame commits --- Aetheria.State.Daemon.Smoke/Program.cs | 112 ++++++ .../AetheriaYmirPersistence.cs | 11 + .../AetheriaYmirPersistenceCoordinator.cs | 320 ++++++++++++++++++ .../AetheriaYmirPersistenceDocuments.cs | 43 +++ .../AetheriaYmirWorldPhysics.cs | 76 +++++ Aetheria.State.Daemon/Program.cs | 68 +++- docs/aetheria-current-codebase-model.md | 22 +- docs/aetheria-perfect-machine-map.md | 25 ++ 8 files changed, 665 insertions(+), 12 deletions(-) create mode 100644 Aetheria.State.Daemon/AetheriaYmirPersistence.cs create mode 100644 Aetheria.State.Daemon/AetheriaYmirPersistenceCoordinator.cs create mode 100644 Aetheria.State.Daemon/AetheriaYmirPersistenceDocuments.cs diff --git a/Aetheria.State.Daemon.Smoke/Program.cs b/Aetheria.State.Daemon.Smoke/Program.cs index 8e1fb84f2..448722b90 100644 --- a/Aetheria.State.Daemon.Smoke/Program.cs +++ b/Aetheria.State.Daemon.Smoke/Program.cs @@ -1,4 +1,6 @@ +using Aetheria.State; using Aetheria.State.Daemon; +using Aetheria.State.Documents; using GameCult.Aetheria.State.Verse; using GameCult.Caching; using System.Globalization; @@ -45,6 +47,7 @@ internal sealed class AetheriaDaemonYmirSmokeChecks RunCheck(TractorRampsAndPullsThroughYmirWithoutTeleportingCargo); RunCheck(PickupIsCapacityCheckedExactlyOnceAndExpires); RunCheck(PickupShieldContactCollectsOrBounces); + RunCheck(YmirRestartDoesNotReplayConsumedPickupContact); } public void RunPayload() @@ -162,6 +165,7 @@ internal sealed class AetheriaDaemonYmirSmokeChecks TractorRampsAndPullsThroughYmirWithoutTeleportingCargo, PickupIsCapacityCheckedExactlyOnceAndExpires, PickupShieldContactCollectsOrBounces, + YmirRestartDoesNotReplayConsumedPickupContact, ThermalCellsUseFossilConductionAndRadiation, EnergyNetworkSettlesReactorAfterConsumers, RadiatorPumpsHeatBeforeReactorSettlement, @@ -3147,6 +3151,114 @@ internal sealed class AetheriaDaemonYmirSmokeChecks throw new InvalidOperationException($"{message}. Expected {expected}; actual {actual}."); } + private static void YmirRestartDoesNotReplayConsumedPickupContact() + { + var root = Path.Combine(Path.GetTempPath(), "aetheria-ymir-restart-" + Guid.NewGuid().ToString("N")); + Directory.CreateDirectory(root); + var statePath = Path.Combine(root, "world.cc"); + try + { + var hull = CatalogItem("restart-contact-hull"); hull.HullCapacity = PerformanceStat(1); + var salvage = CatalogItem("restart-contact-salvage"); salvage.Volume = 1; + var cargoBay = CatalogItem("restart-contact-cargo-bay"); cargoBay.InteriorOccupiedCells = 1; + var catalog = new AetheriaRuntimeCatalogSnapshot([hull, salvage, cargoBay], [], []); + var ship = Entity(0, 0, "player"); ship.HullItemKey = hull.ItemKey; + ship.Equipment = [new AetheriaRuntimeLoadoutItemSlotCommit + { + Item = new AetheriaRuntimeLoadoutItemCommit { ItemKey = cargoBay.ItemKey, Quantity = 1 } + }]; + ship.CargoContents = [Cargo((salvage.ItemKey, 1, 0, 0))]; + var run = new AetheriaRuntimeRunCheckpointCommit + { + RunId = "ymir-restart-contact-smoke", + CurrentZoneIndex = 0, + CurrentEntityKey = "zone.0.entity.0", + Zones = [new AetheriaRuntimeZoneSnapshotCommit + { + ZoneIndex = 0, + Entities = [ship], + DroppedPickups = [new AetheriaRuntimeDroppedPickupCommit + { + PickupIndex = 10, + PositionX = 20, + Item = new AetheriaRuntimeLoadoutItemCommit { ItemKey = salvage.ItemKey, Quantity = 1 }, + LifetimeSeconds = 30 + }] + }] + }; + var frame = new AetheriaRuntimeDaemonFrameDocument + { + FrameId = 1, + FixedDeltaSeconds = 0.1, + SimulationTimeSeconds = 0.1, + Run = run + }; + + using (var physics = new AetheriaYmirWorldPhysics()) + { + AetheriaRuntimeDaemonTickRunner.Tick(statePath, run, + new AetheriaRuntimeDaemonTickOptions + { + WorldPhysics = physics, + FrameId = 1, + FixedDeltaSeconds = 0.1, + SimulationTimeSeconds = 0.1, + Catalog = catalog, + BuildPublications = false + }); + RequireEqual(1, run.PickupContactReceipts.Count, + "pre-restart rejection must consume exactly one Box3D Begin fact"); + var rejectedVelocity = run.Zones[0].DroppedPickups.Single().VelocityX; + Require(rejectedVelocity > 20, "pre-restart rejection must persist the outward kick in world truth"); + + frame.Run = run; + using var node = AetheriaStateNode.OpenAsync(statePath).GetAwaiter().GetResult(); + using var persistence = AetheriaYmirPersistenceCoordinator.OpenAsync(node, physics, null) + .GetAwaiter().GetResult(); + var capture = persistence.Capture(frame); + persistence.PersistPrivateAsync(capture).GetAwaiter().GetResult(); + node.MutableDocument(AetheriaRuntimeVerseRecordKeys.DaemonFrameLatest) + .ReplaceAsync(frame).GetAwaiter().GetResult(); + node.FlushAsync(soft: false).GetAwaiter().GetResult(); + persistence.ActivateAsync().GetAwaiter().GetResult(); + Require(!AetheriaDocumentRegistry.DocumentTypes.Contains(typeof(AetheriaYmirResumeDocument)), + "daemon-private Ymir resume types must not enter the public Aetheria registry"); + Require(!node.Cache.AllEntries.Any(value => value is AetheriaYmirResumeDocument or AetheriaYmirJournalChunkDocument), + "public CultMesh cache must not enumerate daemon-private Ymir restart material"); + } + + using (var node = AetheriaStateNode.OpenAsync(statePath).GetAwaiter().GetResult()) + { + var durable = node.MutableDocument(AetheriaRuntimeVerseRecordKeys.DaemonFrameLatest) + .ReadAsync().GetAwaiter().GetResult() + ?? throw new InvalidOperationException("Durable restart smoke frame is missing."); + using var restoredPhysics = new AetheriaYmirWorldPhysics(); + using var persistence = AetheriaYmirPersistenceCoordinator.OpenAsync(node, restoredPhysics, durable) + .GetAwaiter().GetResult(); + AetheriaRuntimeDaemonTickRunner.Tick(statePath, durable.Run, + new AetheriaRuntimeDaemonTickOptions + { + WorldPhysics = restoredPhysics, + FrameId = 2, + FixedDeltaSeconds = 0.1, + SimulationTimeSeconds = 0.2, + Catalog = catalog, + BuildPublications = false + }); + RequireEqual(1, durable.Run.PickupContactReceipts.Count, + "restart must not replay an already consumed Box3D Begin fact"); + RequireEqual(1, durable.Run.GameEvents.Count(value => + value.Kind == "pickup.rejected" && value.PickupIndex == 10), + "restart must not duplicate authoritative pickup rejection feedback"); + } + } + finally + { + if (Directory.Exists(root)) + Directory.Delete(root, recursive: true); + } + } + private sealed class ScriptedWorldPhysics(params AetheriaRuntimeWorldBeginContact[] contacts) : IAetheriaRuntimeWorldPhysics { public string ImplementationId => "smoke.scripted-world"; diff --git a/Aetheria.State.Daemon/AetheriaYmirPersistence.cs b/Aetheria.State.Daemon/AetheriaYmirPersistence.cs new file mode 100644 index 000000000..a526e9813 --- /dev/null +++ b/Aetheria.State.Daemon/AetheriaYmirPersistence.cs @@ -0,0 +1,11 @@ +using Ymir.Core; + +namespace Aetheria.State.Daemon; + +public sealed record AetheriaYmirZonePersistenceCapture( + string RunId, + int ZoneIndex, + long FrameId, + int SimulationStepIndex, + YmirSessionPersistenceCapture World, + YmirSessionPersistenceCapture? Payload); diff --git a/Aetheria.State.Daemon/AetheriaYmirPersistenceCoordinator.cs b/Aetheria.State.Daemon/AetheriaYmirPersistenceCoordinator.cs new file mode 100644 index 000000000..cd4b1b5b7 --- /dev/null +++ b/Aetheria.State.Daemon/AetheriaYmirPersistenceCoordinator.cs @@ -0,0 +1,320 @@ +using System.Globalization; +using System.Security.Cryptography; +using System.Text; +using Aetheria.State.Documents; +using GameCult.Aetheria.State.Verse; +using GameCult.Caching; +using GameCult.Caching.MessagePack; +using GameCult.Mesh; +using Ymir.Core; + +namespace Aetheria.State.Daemon; + +/// +/// Owns daemon-private Ymir restart material. The public Aetheria frame remains the +/// sole durable commit marker; private chunks and resumes are never advertised. +/// +public sealed class AetheriaYmirPersistenceCoordinator : IDisposable +{ + private const string WorldChannel = "world"; + private const string PayloadChannel = "payloads"; + private readonly CultCache _cache; + private readonly AetheriaYmirWorldPhysics _physics; + private readonly Dictionary _cursors = new(); + private bool _activated; + + private AetheriaYmirPersistenceCoordinator( + CultCache cache, + AetheriaYmirWorldPhysics physics, + bool activated) + { + _cache = cache; + _physics = physics; + _activated = activated; + } + + public static string PrivateStatePath(string publicStatePath) => publicStatePath + ".ymir.cc"; + + public static async Task OpenAsync( + AetheriaStateNode node, + AetheriaYmirWorldPhysics physics, + AetheriaRuntimeDaemonFrameDocument? frame) + { + ArgumentNullException.ThrowIfNull(node); + ArgumentNullException.ThrowIfNull(physics); + var registry = CultMesh.CreateCultCacheDocumentRegistry( + typeof(AetheriaYmirPersistenceMarkerDocument), + typeof(AetheriaYmirJournalChunkDocument), + typeof(AetheriaYmirResumeDocument)); + var cache = await CultCacheMessagePack.OpenAsync( + PrivateStatePath(node.StatePath), + new CultCacheOpenOptions + { + Registry = registry, + PullOnOpen = true, + UseDirectoryStore = true, + FlushOnDispose = true, + StoreFlushOnDispose = true + }).ConfigureAwait(false); + try + { + var activated = cache.Get( + new CultRecordKey(AetheriaYmirPersistenceMarkerDocument.RecordKey)) != null; + var coordinator = new AetheriaYmirPersistenceCoordinator(cache, physics, activated); + if (frame?.Run == null || string.IsNullOrWhiteSpace(frame.Run.RunId)) + return coordinator; + + var zones = (frame.Run.Zones ?? Array.Empty()) + .Where(zone => zone != null) + .OrderBy(zone => zone.ZoneIndex) + .ToArray(); + var resumes = zones + .Select(zone => cache.Get( + new CultRecordKey(ResumeRecordKey(frame.Run.RunId, zone.ZoneIndex, frame.FrameId)))) + .ToArray(); + if (resumes.All(value => value == null) && !activated) + return coordinator; + if (resumes.Any(value => value == null)) + throw new InvalidOperationException( + $"Aetheria frame {frame.FrameId} has incomplete daemon-private Ymir resume state."); + + foreach (var pair in zones.Zip(resumes, (zone, resume) => (zone, resume: resume!))) + coordinator.RestoreZone(frame.Run.RunId, frame.FrameId, pair.zone.ZoneIndex, pair.resume); + return coordinator; + } + catch + { + cache.Dispose(); + throw; + } + } + + public IReadOnlyList Capture( + AetheriaRuntimeDaemonFrameDocument frame) + { + ArgumentNullException.ThrowIfNull(frame); + var run = frame.Run ?? throw new InvalidOperationException("Ymir persistence requires an authoritative run."); + if (string.IsNullOrWhiteSpace(run.RunId)) + throw new InvalidOperationException("Ymir persistence requires a stable run id."); + return (run.Zones ?? Array.Empty()) + .Where(zone => zone != null) + .OrderBy(zone => zone.ZoneIndex) + .Select(zone => + { + var world = CursorFor(run.RunId, zone.ZoneIndex, WorldChannel); + var payload = CursorFor(run.RunId, zone.ZoneIndex, PayloadChannel); + return _physics.CapturePersistence( + run.RunId, + zone.ZoneIndex, + frame.FrameId, + world.Generation, + world.EntryCount, + payload.Generation, + payload.EntryCount); + }) + .ToArray(); + } + + public async Task PersistPrivateAsync(IReadOnlyList captures) + { + ArgumentNullException.ThrowIfNull(captures); + foreach (var capture in captures) + { + await PutChunkAsync(capture, WorldChannel, capture.World.JournalChunk).ConfigureAwait(false); + if (capture.Payload != null) + await PutChunkAsync(capture, PayloadChannel, capture.Payload.JournalChunk).ConfigureAwait(false); + } + await _cache.FlushAsync(soft: false).ConfigureAwait(false); + + foreach (var capture in captures) + { + var world = capture.World.ResumeDescriptor; + var payload = capture.Payload?.ResumeDescriptor; + var resume = new AetheriaYmirResumeDocument + { + RunId = capture.RunId, + ZoneIndex = capture.ZoneIndex, + FrameId = capture.FrameId, + SimulationStepIndex = capture.SimulationStepIndex, + WorldSessionGeneration = world.SessionGeneration, + WorldJournalEntryCount = world.JournalEntryCount, + WorldDescriptorPayload = YmirSessionCheckpointCodec.Encode(world), + PayloadSessionGeneration = payload?.SessionGeneration ?? "", + PayloadJournalEntryCount = payload?.JournalEntryCount ?? 0, + PayloadDescriptorPayload = payload == null ? [] : YmirSessionCheckpointCodec.Encode(payload) + }; + await PutImmutableAsync( + new CultRecordKey(ResumeRecordKey(capture.RunId, capture.ZoneIndex, capture.FrameId)), + resume, + Equivalent).ConfigureAwait(false); + } + await _cache.FlushAsync(soft: false).ConfigureAwait(false); + + foreach (var capture in captures) + { + Remember(capture.RunId, capture.ZoneIndex, WorldChannel, capture.World.ResumeDescriptor); + if (capture.Payload != null) + Remember(capture.RunId, capture.ZoneIndex, PayloadChannel, capture.Payload.ResumeDescriptor); + } + } + + /// + /// Activates fail-closed restart only after the matching public frame has been hard-flushed. + /// + public async Task ActivateAsync() + { + if (_activated) + return; + await _cache.UpsertAsync( + new AetheriaYmirPersistenceMarkerDocument + { + ActivatedAtUtc = DateTimeOffset.UtcNow.ToString("O", CultureInfo.InvariantCulture) + }, + new CultRecordHandle( + new CultRecordKey(AetheriaYmirPersistenceMarkerDocument.RecordKey))).ConfigureAwait(false); + await _cache.FlushAsync(soft: false).ConfigureAwait(false); + _activated = true; + } + + public static string ResumeRecordKey(string runId, int zoneIndex, long frameId) => + $"private:aetheria.ymir.resume.v1:{RunKey(runId)}:{zoneIndex}:{frameId:D20}"; + + public void Dispose() => _cache.Dispose(); + + private async Task PutChunkAsync( + AetheriaYmirZonePersistenceCapture capture, + string channel, + YmirSessionJournalChunk? chunk) + { + if (chunk == null) + return; + var document = new AetheriaYmirJournalChunkDocument + { + RunId = capture.RunId, + ZoneIndex = capture.ZoneIndex, + Channel = channel, + SessionGeneration = chunk.SessionGeneration, + FirstEntryIndex = chunk.FirstEntryIndex, + EntryCount = chunk.Entries.Count, + Payload = YmirSessionCheckpointCodec.Encode(chunk) + }; + await PutImmutableAsync( + new CultRecordKey(JournalRecordKey( + capture.RunId, + capture.ZoneIndex, + channel, + chunk.SessionGeneration, + chunk.FirstEntryIndex)), + document, + Equivalent).ConfigureAwait(false); + } + + private async Task PutImmutableAsync( + CultRecordKey key, + T document, + Func equivalent) + where T : class + { + var existing = _cache.Get(key); + if (existing != null) + { + if (!equivalent(existing, document)) + throw new InvalidOperationException($"Immutable Ymir persistence record {key.Value} conflicts with durable state."); + return; + } + await _cache.UpsertAsync(document, new CultRecordHandle(key)).ConfigureAwait(false); + } + + private void RestoreZone(string runId, long frameId, int zoneIndex, AetheriaYmirResumeDocument resume) + { + if (!string.Equals(resume.RunId, runId, StringComparison.Ordinal) || + resume.ZoneIndex != zoneIndex || resume.FrameId != frameId || + resume.WorldDescriptorPayload.Length == 0) + throw new InvalidOperationException("Aetheria private Ymir resume identity does not match the durable frame."); + var world = YmirSessionCheckpointCodec.DecodeResumeDescriptor(resume.WorldDescriptorPayload); + var payload = resume.PayloadDescriptorPayload.Length == 0 + ? null + : YmirSessionCheckpointCodec.DecodeResumeDescriptor(resume.PayloadDescriptorPayload); + if (!string.Equals(world.SessionGeneration, resume.WorldSessionGeneration, StringComparison.Ordinal) || + world.JournalEntryCount != resume.WorldJournalEntryCount || + !string.Equals(payload?.SessionGeneration ?? "", resume.PayloadSessionGeneration, StringComparison.Ordinal) || + (payload?.JournalEntryCount ?? 0) != resume.PayloadJournalEntryCount) + throw new InvalidOperationException("Aetheria private Ymir resume metadata disagrees with its checksummed payload."); + + var worldChunks = Chunks(runId, zoneIndex, WorldChannel, world.SessionGeneration, world.JournalEntryCount); + var payloadChunks = payload == null + ? Array.Empty() + : Chunks(runId, zoneIndex, PayloadChannel, payload.SessionGeneration, payload.JournalEntryCount); + _physics.RestorePersistence( + runId, zoneIndex, frameId, resume.SimulationStepIndex, + world, worldChunks, payload, payloadChunks); + Remember(runId, zoneIndex, WorldChannel, world); + if (payload != null) + Remember(runId, zoneIndex, PayloadChannel, payload); + } + + private YmirSessionJournalChunk[] Chunks( + string runId, + int zoneIndex, + string channel, + string generation, + long entryCount) => + _cache.GetAll() + .Where(value => value != null && + string.Equals(value.RunId, runId, StringComparison.Ordinal) && + value.ZoneIndex == zoneIndex && + string.Equals(value.Channel, channel, StringComparison.Ordinal) && + string.Equals(value.SessionGeneration, generation, StringComparison.Ordinal) && + value.FirstEntryIndex < entryCount) + .OrderBy(value => value.FirstEntryIndex) + .Select(value => + { + var chunk = YmirSessionCheckpointCodec.DecodeJournalChunk(value.Payload); + if (chunk.FirstEntryIndex != value.FirstEntryIndex || chunk.Entries.Count != value.EntryCount) + throw new InvalidOperationException("Aetheria Ymir journal metadata disagrees with its checksummed payload."); + return chunk; + }) + .ToArray(); + + private Cursor CursorFor(string runId, int zoneIndex, string channel) => + _cursors.TryGetValue(new CursorKey(runId, zoneIndex, channel), out var cursor) + ? cursor + : new Cursor("", 0); + + private void Remember( + string runId, + int zoneIndex, + string channel, + YmirSessionResumeDescriptor descriptor) => + _cursors[new CursorKey(runId, zoneIndex, channel)] = + new Cursor(descriptor.SessionGeneration, descriptor.JournalEntryCount); + + private static bool Equivalent(AetheriaYmirJournalChunkDocument left, AetheriaYmirJournalChunkDocument right) => + left.RunId == right.RunId && left.ZoneIndex == right.ZoneIndex && left.Channel == right.Channel && + left.SessionGeneration == right.SessionGeneration && left.FirstEntryIndex == right.FirstEntryIndex && + left.EntryCount == right.EntryCount && left.Payload.AsSpan().SequenceEqual(right.Payload); + + private static bool Equivalent(AetheriaYmirResumeDocument left, AetheriaYmirResumeDocument right) => + left.RunId == right.RunId && left.ZoneIndex == right.ZoneIndex && left.FrameId == right.FrameId && + left.SimulationStepIndex == right.SimulationStepIndex && + left.WorldSessionGeneration == right.WorldSessionGeneration && + left.WorldJournalEntryCount == right.WorldJournalEntryCount && + left.WorldDescriptorPayload.AsSpan().SequenceEqual(right.WorldDescriptorPayload) && + left.PayloadSessionGeneration == right.PayloadSessionGeneration && + left.PayloadJournalEntryCount == right.PayloadJournalEntryCount && + left.PayloadDescriptorPayload.AsSpan().SequenceEqual(right.PayloadDescriptorPayload); + + private static string JournalRecordKey( + string runId, + int zoneIndex, + string channel, + string generation, + long firstEntryIndex) => + $"private:aetheria.ymir.journal.v1:{RunKey(runId)}:{zoneIndex}:{channel}:{generation}:{firstEntryIndex:D20}"; + + private static string RunKey(string runId) => + Convert.ToHexString(SHA256.HashData(Encoding.UTF8.GetBytes(runId))).ToLowerInvariant(); + + private readonly record struct CursorKey(string RunId, int ZoneIndex, string Channel); + private readonly record struct Cursor(string Generation, long EntryCount); +} diff --git a/Aetheria.State.Daemon/AetheriaYmirPersistenceDocuments.cs b/Aetheria.State.Daemon/AetheriaYmirPersistenceDocuments.cs new file mode 100644 index 000000000..cfe59524b --- /dev/null +++ b/Aetheria.State.Daemon/AetheriaYmirPersistenceDocuments.cs @@ -0,0 +1,43 @@ +using GameCult.Caching; +using MessagePack; + +namespace Aetheria.State.Daemon; + +[CultDocument("gamecult.aetheria.ymir_persistence_marker", "gamecult.aetheria.ymir_persistence_marker.v1")] +[MessagePackObject] +public sealed class AetheriaYmirPersistenceMarkerDocument +{ + public const string RecordKey = "private:aetheria.ymir.persistence.v1"; + + [Key(0)] public string FormatId { get; set; } = "gamecult.aetheria.ymir.persistence.v1"; + [Key(1)] public string ActivatedAtUtc { get; set; } = ""; +} + +[CultDocument("gamecult.aetheria.ymir_journal_chunk", "gamecult.aetheria.ymir_journal_chunk.v1")] +[MessagePackObject] +public sealed class AetheriaYmirJournalChunkDocument +{ + [Key(0)] public string RunId { get; set; } = ""; + [Key(1)] public int ZoneIndex { get; set; } + [Key(2)] public string Channel { get; set; } = ""; + [Key(3)] public string SessionGeneration { get; set; } = ""; + [Key(4)] public long FirstEntryIndex { get; set; } + [Key(5)] public int EntryCount { get; set; } + [Key(6)] public byte[] Payload { get; set; } = []; +} + +[CultDocument("gamecult.aetheria.ymir_resume", "gamecult.aetheria.ymir_resume.v1")] +[MessagePackObject] +public sealed class AetheriaYmirResumeDocument +{ + [Key(0)] public string RunId { get; set; } = ""; + [Key(1)] public int ZoneIndex { get; set; } + [Key(2)] public long FrameId { get; set; } + [Key(3)] public int SimulationStepIndex { get; set; } + [Key(4)] public string WorldSessionGeneration { get; set; } = ""; + [Key(5)] public long WorldJournalEntryCount { get; set; } + [Key(6)] public byte[] WorldDescriptorPayload { get; set; } = []; + [Key(7)] public string PayloadSessionGeneration { get; set; } = ""; + [Key(8)] public long PayloadJournalEntryCount { get; set; } + [Key(9)] public byte[] PayloadDescriptorPayload { get; set; } = []; +} diff --git a/Aetheria.State.Daemon/AetheriaYmirWorldPhysics.cs b/Aetheria.State.Daemon/AetheriaYmirWorldPhysics.cs index 4b2798143..9f9c5c89b 100644 --- a/Aetheria.State.Daemon/AetheriaYmirWorldPhysics.cs +++ b/Aetheria.State.Daemon/AetheriaYmirWorldPhysics.cs @@ -16,6 +16,82 @@ public sealed class AetheriaYmirWorldPhysics : IAetheriaRuntimeWorldPhysics, IDi public string ImplementationId => "ymir.box3d.retained-session.v1"; + public AetheriaYmirZonePersistenceCapture CapturePersistence( + string runId, + int zoneIndex, + long frameId, + string persistedWorldSessionGeneration, + long persistedWorldJournalEntryCount, + string persistedPayloadSessionGeneration, + long persistedPayloadJournalEntryCount) + { + if (!_sessions.TryGetValue(new WorldKey(runId, zoneIndex), out var state) || + state.LastFrameId != frameId) + throw new InvalidOperationException( + $"Cannot persist Ymir world for run '{runId}' zone {zoneIndex} at frame {frameId}."); + if (state.PayloadSession != null && + (state.LastPayloadFrameId != frameId || state.LastPayloadSimulationStepIndex != state.LastSimulationStepIndex)) + throw new InvalidOperationException( + $"Cannot persist Ymir payload world for run '{runId}' zone {zoneIndex} at frame {frameId}."); + + return new AetheriaYmirZonePersistenceCapture( + runId, + zoneIndex, + frameId, + state.LastSimulationStepIndex, + state.WorldSession.CapturePersistence( + string.Equals(persistedWorldSessionGeneration, state.WorldSession.Info.SessionGeneration, StringComparison.Ordinal) + ? persistedWorldJournalEntryCount + : 0), + state.PayloadSession?.CapturePersistence( + string.Equals(persistedPayloadSessionGeneration, state.PayloadSession.Info.SessionGeneration, StringComparison.Ordinal) + ? persistedPayloadJournalEntryCount + : 0)); + } + + public void RestorePersistence( + string runId, + int zoneIndex, + long frameId, + int simulationStepIndex, + YmirSessionResumeDescriptor worldDescriptor, + IReadOnlyList worldChunks, + YmirSessionResumeDescriptor? payloadDescriptor, + IReadOnlyList payloadChunks) + { + var key = new WorldKey(runId, zoneIndex); + if (_sessions.ContainsKey(key)) + throw new InvalidOperationException($"Ymir world for run '{runId}' zone {zoneIndex} is already live."); + var sessionId = $"aetheria.run.{runId}.zone.{zoneIndex}"; + if (!string.Equals(worldDescriptor.SessionId, sessionId + ".world", StringComparison.Ordinal) || + (payloadDescriptor != null && + !string.Equals(payloadDescriptor.SessionId, sessionId + ".payloads", StringComparison.Ordinal))) + throw new InvalidOperationException("Aetheria Ymir resume belongs to another run or zone."); + if (payloadDescriptor == null && payloadChunks.Count > 0) + throw new InvalidOperationException("Aetheria Ymir resume has payload journal chunks without a descriptor."); + + var world = YmirSession.Restore(worldDescriptor, worldChunks); + YmirSession? payload = null; + try + { + payload = payloadDescriptor == null ? null : YmirSession.Restore(payloadDescriptor, payloadChunks); + _sessions.Add(key, new SessionState(world) + { + PayloadSession = payload, + LastFrameId = frameId, + LastSimulationStepIndex = simulationStepIndex, + LastPayloadFrameId = payload == null ? -1 : frameId, + LastPayloadSimulationStepIndex = payload == null ? -1 : simulationStepIndex + }); + } + catch + { + payload?.Dispose(); + world.Dispose(); + throw; + } + } + public void RetainWorlds(string runId, IReadOnlyList zoneIndices) { var retainedZones = (zoneIndices ?? Array.Empty()).ToHashSet(); diff --git a/Aetheria.State.Daemon/Program.cs b/Aetheria.State.Daemon/Program.cs index 79d2b76b4..5bcb71954 100644 --- a/Aetheria.State.Daemon/Program.cs +++ b/Aetheria.State.Daemon/Program.cs @@ -48,15 +48,33 @@ discoveryHost.Update(verseHost); await PublishRuntimeSessionAsync(node, options, startedAtUtc, "starting").ConfigureAwait(false); await PublishStateSurfacesAsync(node, options, startedAtUtc).ConfigureAwait(false); var latestFrame = await node.MutableDocument(AetheriaRuntimeVerseRecordKeys.DaemonFrameLatest).ReadAsync().ConfigureAwait(false); +using var physicsPersistence = await AetheriaYmirPersistenceCoordinator.OpenAsync( + node, + worldPhysics, + latestFrame).ConfigureAwait(false); using var cultMeshRudpHost = StartClientCultMeshHost(node, options, () => latestFrame); using var clientSubscriptions = new CultNetDatabaseSubscriptionServer(cultMeshRudpHost, node.Database); using var clientPumpCancellation = new CancellationTokenSource(); var clientPump = RunClientCultMeshPumpAsync(cultMeshRudpHost, clientPumpCancellation.Token); var nextApiPublicationUtc = DateTimeOffset.UtcNow; var ingressState = new AetheriaDaemonIngressState(); -var firstTick = await TickAsync(node, options, worldPhysics, soaPublisher, latestFrame, ingressState, buildPublications: true).ConfigureAwait(false); +var firstTick = await TickAsync(node, options, worldPhysics, soaPublisher, latestFrame, ingressState, buildPublications: false).ConfigureAwait(false); ThrowIfClientPumpFaulted(clientPump); latestFrame = firstTick.Frame; +var firstPublication = PreparePublication( + node.StatePath, + options, + soaPublisher, + firstTick, + ingressState, + physicsPersistence); +await PublishPreparedDocumentsAsync( + node, + options, + soaPublisher, + physicsPersistence, + firstPublication, + publishTopology: true).ConfigureAwait(false); nextApiPublicationUtc = DateTimeOffset.UtcNow.Add(options.ApiPublicationInterval); Console.WriteLine($"Aetheria Verse daemon published frame {firstTick.Frame.FrameId}."); Console.WriteLine($"Aetheria client CultMesh endpoint: rudp://{options.ClientCultMeshAdvertiseHost}:{cultMeshRudpHost.LocalEndPoint.Port}"); @@ -102,8 +120,20 @@ while (!stopped.Task.IsCompleted) { if (publicationTask.IsFaulted) await publicationTask.ConfigureAwait(false); - var publication = PreparePublication(node.StatePath, options, soaPublisher, tick, ingressState); - publicationTask = PublishPreparedDocumentsAsync(node, options, soaPublisher, publication); + var publication = PreparePublication( + node.StatePath, + options, + soaPublisher, + tick, + ingressState, + physicsPersistence); + publicationTask = PublishPreparedDocumentsAsync( + node, + options, + soaPublisher, + physicsPersistence, + publication, + publishTopology: false); nextApiPublicationUtc = DateTimeOffset.UtcNow.Add(options.ApiPublicationInterval); } if (tick.Frame.FrameId % (traceClientRudp ? 10 : 120) == 0) @@ -300,20 +330,22 @@ static async Task TickAsync( return result; } -static AetheriaRuntimeDaemonTickResult PreparePublication( +static AetheriaPreparedPublication PreparePublication( string statePath, AetheriaDaemonHostOptions options, AetheriaRuntimeDaemonSoaFramePublisher soaPublisher, AetheriaRuntimeDaemonTickResult tick, - AetheriaDaemonIngressState ingressState) + AetheriaDaemonIngressState ingressState, + AetheriaYmirPersistenceCoordinator physicsPersistence) { + var physics = physicsPersistence.Capture(tick.Frame); var frameBytes = MessagePackSerializer.Serialize(tick.Frame); var frame = MessagePackSerializer.Deserialize(frameBytes); var operation = new AetheriaRuntimeDaemonOperationResult( frame.Run ?? new AetheriaRuntimeRunCheckpointCommit(), tick.OperationResult.AppliedCommandIds.ToArray(), tick.OperationResult.RejectedCommandIds.ToArray()); - return AetheriaRuntimeDaemonTickRunner.BuildPublications( + var publication = AetheriaRuntimeDaemonTickRunner.BuildPublications( statePath, operation, frame, @@ -330,15 +362,31 @@ static AetheriaRuntimeDaemonTickResult PreparePublication( SoaFramePublisher = soaPublisher }, frame.AccountedCommandIds?.Count ?? 0); + return new AetheriaPreparedPublication(publication, physics); } static async Task PublishPreparedDocumentsAsync( AetheriaStateNode node, AetheriaDaemonHostOptions options, AetheriaRuntimeDaemonSoaFramePublisher soaPublisher, - AetheriaRuntimeDaemonTickResult publication) + AetheriaYmirPersistenceCoordinator physicsPersistence, + AetheriaPreparedPublication prepared, + bool publishTopology) { - await PublishDaemonApiDocumentsAsync(node, options, soaPublisher, publication, publishTopology: false).ConfigureAwait(false); + await physicsPersistence.PersistPrivateAsync(prepared.Physics).ConfigureAwait(false); + await PublishDaemonApiDocumentsAsync( + node, + options, + soaPublisher, + prepared.Publication, + publishTopology).ConfigureAwait(false); + if (publishTopology) + { + await PublishStateSurfacesAsync(node, options, prepared.Publication.Frame.PublishedAtUtc).ConfigureAwait(false); + await PublishOdinSurfaceAnnouncementsAsync(node, options, prepared.Publication.Frame.PublishedAtUtc).ConfigureAwait(false); + } + await node.FlushAsync(soft: false).ConfigureAwait(false); + await physicsPersistence.ActivateAsync().ConfigureAwait(false); } static async Task RefreshControlPlaneInputsAsync( @@ -2993,6 +3041,10 @@ static AetheriaRuntimeLoadoutItemCommit ToLoadoutItemCommit(AetheriaLoadoutItem? }; } +internal sealed record AetheriaPreparedPublication( + AetheriaRuntimeDaemonTickResult Publication, + IReadOnlyList Physics); + internal sealed class AetheriaDaemonIngressState { public int ObservedEveCommandCount { get; set; } = -1; diff --git a/docs/aetheria-current-codebase-model.md b/docs/aetheria-current-codebase-model.md index a61026029..7162fac76 100644 --- a/docs/aetheria-current-codebase-model.md +++ b/docs/aetheria-current-codebase-model.md @@ -143,6 +143,11 @@ Unity physics is already fenced as forbidden gameplay authority by shader/material bundle are owned by EveUnity with their original Unity GUIDs; Aetheria owns only the configured prefab and its asset-manifest advertisement. +Texture production currently uses checked-in, pre-generated provider assets. +Substance is not part of the toolchain. A later asset-pipeline pass may move +texture generation into Blender baking, but that pipeline remains provider-side +and cannot become an EveUnity runtime dependency. + Current Ymir control flow: - Ymir is deliberately a Box3D wrapper and physics-daemon boundary; its name @@ -162,6 +167,14 @@ Current Ymir control flow: fields, and pickup-rejection velocity into explicit retained-session commands. Post-step body values in the run checkpoint are projections from Ymir, not an independent physics solver. +- Restart material lives in a second daemon-private CultCache at + `.ymir.cc`; it is absent from the public Aetheria document + registry and client subscription database. Ymir emits immutable incremental + journal chunks plus bounded per-frame resume descriptors. The daemon flushes + chunks, then resumes, then the public frame. That public frame is the commit + marker. Startup restores the exact `(RunId, FrameId, ZoneIndex)` records + before the client CultMesh host starts, and fails closed on incomplete active + persistence. - Cargo collection accepts only typed Ymir `Begin` facts. Aetheria persists one pickup-contact receipt per Ymir `FactId` before exposing the resulting event, so duplicate fact delivery cannot add cargo twice. Proximity and client loot @@ -199,10 +212,11 @@ Remaining debt: daemon. A future remote Ymir lowering must preserve the same ordered zone transaction, stable body identities, revisions, and query facts rather than restoring a second Aetheria physics authority. -- The in-process session registry is the live daemon implementation. CultMesh - discovery/routing, durable Ymir command receipt storage, and complete Ymir - checkpoint reconstruction remain daemon-boundary work; the checkpoint keeps - Aetheria's durable pickup-fact consumption ledger. +- The in-process session registry is the live daemon implementation. Private + replay reconstruction is wired and restart-proven for a consumed pickup + contact, including no duplicate receipt, event, or rejection kick. Ymir's + complete in-memory journal is not compacted yet; disk writes are incremental, + memory retention is not. - Clickable raycasts still construct query bodies from Unity click bounds. That is presentation picking, not simulation authority, but it should stay clearly labeled as renderer/UI picking. diff --git a/docs/aetheria-perfect-machine-map.md b/docs/aetheria-perfect-machine-map.md index 0c90c64e0..0ad43592b 100644 --- a/docs/aetheria-perfect-machine-map.md +++ b/docs/aetheria-perfect-machine-map.md @@ -2294,3 +2294,28 @@ instead of rehydrating a parallel presentation list. Any predicate that still needs legacy DTO objects must earn that dependency by using behavior objects or simulation-only methods that typed facets do not yet expose. + +Ymir restart ownership is deliberately private and asymmetric: + +- Owner: `AetheriaYmirPersistenceCoordinator` owns reconstruction durability; + the public daemon frame still owns committed gameplay time. +- Inputs: an immutable completed-tick Ymir capture, the prior durable journal + cursor, and the exact public `(RunId, FrameId, ZoneIndex)` boundary. +- Outputs: checksummed immutable journal chunks and immutable bounded resume + records in `.ymir.cc`. +- Derived state: native Box3D worlds and handles are reconstructed process + state. Resume records are recovery inputs, not client world truth. +- Forbidden writers: the public Aetheria CultCache, CultMesh subscriptions, + Eve surfaces, Unity, and broad public snapshots cannot store, enumerate, or + select private replay history. +- Shared path: first publication and periodic publication both capture before a + later simulation tick, flush private chunks, flush private resumes, publish + and hard-flush the public frame, then activate fail-closed restart. +- Cut line: the rejected design that embedded complete Ymir checkpoints in + every public zone/frame is absent; no public persistence document types are + registered. +- Verification layer: daemon smoke tears down a live rejected-pickup contact, + restores the exact durable frame into fresh Ymir sessions, and proves the + next tick cannot duplicate its receipt, feedback event, or kick. Ymir unit + tests prove incremental suffix capture, exact chunk coverage, checksums, and + replay verifier convergence.