Persist Ymir sessions behind daemon frame commits
This commit is contained in:
parent
72128da11e
commit
002ade7a72
8 changed files with 665 additions and 12 deletions
|
|
@ -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<AetheriaRuntimeDaemonFrameDocument>(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<AetheriaRuntimeDaemonFrameDocument>(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";
|
||||
|
|
|
|||
11
Aetheria.State.Daemon/AetheriaYmirPersistence.cs
Normal file
11
Aetheria.State.Daemon/AetheriaYmirPersistence.cs
Normal file
|
|
@ -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);
|
||||
320
Aetheria.State.Daemon/AetheriaYmirPersistenceCoordinator.cs
Normal file
320
Aetheria.State.Daemon/AetheriaYmirPersistenceCoordinator.cs
Normal file
|
|
@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// Owns daemon-private Ymir restart material. The public Aetheria frame remains the
|
||||
/// sole durable commit marker; private chunks and resumes are never advertised.
|
||||
/// </summary>
|
||||
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<CursorKey, Cursor> _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<AetheriaYmirPersistenceCoordinator> 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<AetheriaYmirPersistenceMarkerDocument>(
|
||||
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<AetheriaRuntimeZoneSnapshotCommit>())
|
||||
.Where(zone => zone != null)
|
||||
.OrderBy(zone => zone.ZoneIndex)
|
||||
.ToArray();
|
||||
var resumes = zones
|
||||
.Select(zone => cache.Get<AetheriaYmirResumeDocument>(
|
||||
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<AetheriaYmirZonePersistenceCapture> 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<AetheriaRuntimeZoneSnapshotCommit>())
|
||||
.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<AetheriaYmirZonePersistenceCapture> 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);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Activates fail-closed restart only after the matching public frame has been hard-flushed.
|
||||
/// </summary>
|
||||
public async Task ActivateAsync()
|
||||
{
|
||||
if (_activated)
|
||||
return;
|
||||
await _cache.UpsertAsync(
|
||||
new AetheriaYmirPersistenceMarkerDocument
|
||||
{
|
||||
ActivatedAtUtc = DateTimeOffset.UtcNow.ToString("O", CultureInfo.InvariantCulture)
|
||||
},
|
||||
new CultRecordHandle<AetheriaYmirPersistenceMarkerDocument>(
|
||||
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<T>(
|
||||
CultRecordKey key,
|
||||
T document,
|
||||
Func<T, T, bool> equivalent)
|
||||
where T : class
|
||||
{
|
||||
var existing = _cache.Get<T>(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<T>(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<YmirSessionJournalChunk>()
|
||||
: 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<AetheriaYmirJournalChunkDocument>()
|
||||
.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);
|
||||
}
|
||||
43
Aetheria.State.Daemon/AetheriaYmirPersistenceDocuments.cs
Normal file
43
Aetheria.State.Daemon/AetheriaYmirPersistenceDocuments.cs
Normal file
|
|
@ -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; } = [];
|
||||
}
|
||||
|
|
@ -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<YmirSessionJournalChunk> worldChunks,
|
||||
YmirSessionResumeDescriptor? payloadDescriptor,
|
||||
IReadOnlyList<YmirSessionJournalChunk> 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<int> zoneIndices)
|
||||
{
|
||||
var retainedZones = (zoneIndices ?? Array.Empty<int>()).ToHashSet();
|
||||
|
|
|
|||
|
|
@ -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<AetheriaRuntimeDaemonFrameDocument>(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<AetheriaRuntimeDaemonTickResult> 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<AetheriaRuntimeDaemonFrameDocument>(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<AetheriaYmirZonePersistenceCapture> Physics);
|
||||
|
||||
internal sealed class AetheriaDaemonIngressState
|
||||
{
|
||||
public int ObservedEveCommandCount { get; set; } = -1;
|
||||
|
|
|
|||
|
|
@ -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
|
||||
`<public-state>.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.
|
||||
|
|
|
|||
|
|
@ -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 `<state>.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.
|
||||
|
|
|
|||
Loading…
Reference in a new issue