GameCult/Aetheria
Watch
9
0
Fork
You've already forked Aetheria
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Return idle Aetheria sessions to ready

Browse source
This commit is contained in:
Meta Vens 2026-07-22 00:19:29 +01:00
commit ec00373067
3 changed files with 204 additions and 161 deletions
Download patch file Download diff file Expand all files Collapse all files

370
Aetheria.State.Daemon/Program.cs
View file

@ -32,7 +32,6 @@ void TraceStartup(string name)
$"(total {startup.Elapsed.TotalMilliseconds:0.###}ms).");
startupPhase.Restart();
}
using var worldPhysics = new AetheriaYmirWorldPhysics();
var traceClientTransport = string.Equals(
Environment.GetEnvironmentVariable("AETHERIA_TRACE_CLIENT_TRANSPORT"),
"1",
@ -92,12 +91,8 @@ Console.WriteLine($"Aetheria client CDN endpoint: {cultMeshClientHost.ContentEnd
Console.WriteLine($"Aetheria client realtime endpoint: {cultMeshClientHost.RealtimeEndpoint}");
Console.WriteLine($"Aetheria daemon transport-ready in {startup.Elapsed.TotalMilliseconds:0.###}ms.");
using var clientSubscriptions = new CultNetDatabaseSubscriptionServer(cultMeshClientHost.Control, node.Database);
var playableWorldRequested = new TaskCompletionSource<object?>(TaskCreationOptions.RunContinuationsAsynchronously);
clientSubscriptions.DemandChanged += demand =>
{
if (RequestsPlayableWorld(demand))
playableWorldRequested.TrySetResult(null);
};
var playableWorldDemand = new AetheriaPlayableWorldDemandState();
clientSubscriptions.DemandChanged += playableWorldDemand.Observe;
var managedViewportDemand = new AetheriaManagedViewportDemandState();
clientSubscriptions.DemandChanged += managedViewportDemand.Observe;
if (traceClientTransport)
@ -117,175 +112,199 @@ Console.CancelKeyPress += (_, eventArgs) =>
stopped.TrySetResult(null);
};
if (!options.Once && !options.UseTerminusFixture)
{
await PublishRuntimeSessionAsync(node, options, startedAtUtc, "ready").ConfigureAwait(false);
Console.WriteLine("Aetheria daemon ready; waiting for a client to load or generate a world.");
while (!stopped.Task.IsCompleted)
{
ThrowIfClientHostFaulted(cultMeshClientHost);
await AcceptCoreEveInvocationsAsync(node, options, latestFrame).ConfigureAwait(false);
await AcceptEveCommandsAsync(node, options, ingressState).ConfigureAwait(false);
if (await ApplyRequestedNewGameSessionAsync(node, options).ConfigureAwait(false))
{
latestFrame = null;
break;
}
if (playableWorldRequested.Task.IsCompleted &&
(HasPlayableRun(latestFrame?.Run) ||
HasPlayableRun(await ReadRuntimeRunCheckpointAsync(node, options.RenderSettings).ConfigureAwait(false))))
break;
var completed = await Task.WhenAny(stopped.Task, Task.Delay(options.TickInterval)).ConfigureAwait(false);
if (completed == stopped.Task)
break;
}
if (stopped.Task.IsCompleted)
{
await PublishRuntimeSessionAsync(node, options, startedAtUtc, "stopped").ConfigureAwait(false);
return;
}
}
else
{
await EnsurePlayableRunDocumentsAsync(node, options, startedAtUtc, latestFrame).ConfigureAwait(false);
TraceStartup("playable-run");
}
await EnsureGameSessionAsync(node, options, startedAtUtc, latestFrame).ConfigureAwait(false);
TraceStartup("game-session");
using var physicsPersistence = await AetheriaYmirPersistenceCoordinator.OpenAsync(
node,
worldPhysics,
latestFrame).ConfigureAwait(false);
TraceStartup("ymir-persistence");
var nextApiPublicationUtc = DateTimeOffset.UtcNow;
var hotState = new AetheriaHotEntityPublicationState();
var reactiveSurfaceState = new AetheriaReactiveSurfacePublicationState();
var publishRestoredFrame = !options.Once &&
latestFrame?.Run != null &&
HasPlayableRun(latestFrame.Run) &&
HasPersistedClientBootstrap(node, latestFrame);
AetheriaRuntimeDaemonTickResult? initialPublication = null;
AetheriaPreparedPublication? initialPrepared = null;
Task publicationTask;
if (publishRestoredFrame)
{
await RefreshControlPlaneInputsAsync(node, ingressState).ConfigureAwait(false);
TraceStartup("restored-client-state");
}
else
{
var firstTick = await TickAsync(
node, options, unityBundles, worldPhysics, latestFrame, ingressState, buildPublications: false).ConfigureAwait(false);
TraceStartup("first-tick");
latestFrame = firstTick.Frame;
initialPrepared = PreparePublication(
node.StatePath,
options,
firstTick,
ingressState,
physicsPersistence);
initialPublication = initialPrepared.Publication;
}
if (!publishRestoredFrame)
{
await PublishClientGameplayDocumentsAsync(
node,
options,
unityBundles,
initialPublication!,
ingressState.Catalog ?? throw new InvalidOperationException("Aetheria runtime catalog was not initialized."),
reactiveSurfaceState,
publishTopology: true).ConfigureAwait(false);
}
TraceStartup("client-bootstrap");
ThrowIfClientHostFaulted(cultMeshClientHost);
await PublishHotEntityStateAsync(
node, soaPublisher, hotState, latestFrame!, ingressState.Catalog, cultMeshClientHost).ConfigureAwait(false);
TraceStartup("hot-entity-state");
if (publishRestoredFrame)
{
publicationTask = Task.CompletedTask;
}
else
{
publicationTask = Task.Run(() => PersistPreparedDocumentsAsync(
node,
options,
physicsPersistence,
initialPrepared!,
initialTopology: true));
}
nextApiPublicationUtc = DateTimeOffset.UtcNow.Add(options.ApiPublicationInterval);
Console.WriteLine($"Aetheria Verse daemon published frame {latestFrame!.FrameId}.");
Console.WriteLine($"Aetheria daemon playable-world-ready in {startup.Elapsed.TotalMilliseconds:0.###}ms.");
if (options.Once)
{
await publicationTask.ConfigureAwait(false);
await PublishRuntimeSessionAsync(node, options, startedAtUtc, "completed").ConfigureAwait(false);
return;
}
await PublishRuntimeSessionAsync(node, options, startedAtUtc, "running").ConfigureAwait(false);
Console.WriteLine("Aetheria Verse daemon is running. Press Ctrl+C to stop.");
var nextTickUtc = DateTimeOffset.UtcNow.Add(options.TickInterval);
while (!stopped.Task.IsCompleted)
{
ThrowIfClientHostFaulted(cultMeshClientHost);
var delay = nextTickUtc - DateTimeOffset.UtcNow;
if (delay > TimeSpan.Zero)
if (!options.Once && !options.UseTerminusFixture)
{
var completed = await Task.WhenAny(stopped.Task, Task.Delay(delay)).ConfigureAwait(false);
if (completed == stopped.Task)
await PublishRuntimeSessionAsync(node, options, startedAtUtc, "ready").ConfigureAwait(false);
Console.WriteLine("Aetheria daemon ready; waiting for a client to load or generate a world.");
while (!stopped.Task.IsCompleted)
{
ThrowIfClientHostFaulted(cultMeshClientHost);
await AcceptCoreEveInvocationsAsync(node, options, latestFrame).ConfigureAwait(false);
await AcceptEveCommandsAsync(node, options, ingressState).ConfigureAwait(false);
if (await ApplyRequestedNewGameSessionAsync(node, options).ConfigureAwait(false))
latestFrame = null;
if (playableWorldDemand.IsActive &&
(HasPlayableRun(latestFrame?.Run) ||
HasPlayableRun(await ReadRuntimeRunCheckpointAsync(node, options.RenderSettings).ConfigureAwait(false))))
break;
var completed = await Task.WhenAny(stopped.Task, Task.Delay(options.TickInterval)).ConfigureAwait(false);
if (completed == stopped.Task)
break;
}
if (stopped.Task.IsCompleted)
break;
}
var buildPublications = DateTimeOffset.UtcNow >= nextApiPublicationUtc;
var tick = await TickAsync(node, options, unityBundles, worldPhysics, latestFrame, ingressState, buildPublications: false).ConfigureAwait(false);
ThrowIfClientHostFaulted(cultMeshClientHost);
latestFrame = tick.Frame;
await PublishHotEntityStateAsync(
node, soaPublisher, hotState, tick.Frame, ingressState.Catalog, cultMeshClientHost).ConfigureAwait(false);
nextTickUtc += options.TickInterval;
if (nextTickUtc < DateTimeOffset.UtcNow - options.TickInterval)
nextTickUtc = DateTimeOffset.UtcNow;
if (buildPublications && publicationTask.IsCompleted)
else
{
if (publicationTask.IsFaulted)
await publicationTask.ConfigureAwait(false);
var publication = PreparePublication(
await EnsurePlayableRunDocumentsAsync(node, options, startedAtUtc, latestFrame).ConfigureAwait(false);
TraceStartup("playable-run");
}
latestFrame = await node.MutableDocument<AetheriaRuntimeDaemonFrameDocument>(AetheriaRuntimeVerseRecordKeys.DaemonFrameLatest)
.ReadAsync().ConfigureAwait(false);
using var worldPhysics = new AetheriaYmirWorldPhysics();
await EnsureGameSessionAsync(node, options, startedAtUtc, latestFrame).ConfigureAwait(false);
TraceStartup("game-session");
using var physicsPersistence = await AetheriaYmirPersistenceCoordinator.OpenAsync(
node,
worldPhysics,
latestFrame).ConfigureAwait(false);
TraceStartup("ymir-persistence");
var nextApiPublicationUtc = DateTimeOffset.UtcNow;
var hotState = new AetheriaHotEntityPublicationState();
var reactiveSurfaceState = new AetheriaReactiveSurfacePublicationState();
var publishRestoredFrame = !options.Once &&
latestFrame?.Run != null &&
HasPlayableRun(latestFrame.Run) &&
HasPersistedClientBootstrap(node, latestFrame);
AetheriaRuntimeDaemonTickResult? initialPublication = null;
AetheriaPreparedPublication? initialPrepared = null;
Task publicationTask;
if (publishRestoredFrame)
{
await RefreshControlPlaneInputsAsync(node, ingressState).ConfigureAwait(false);
TraceStartup("restored-client-state");
}
else
{
var firstTick = await TickAsync(
node, options, unityBundles, worldPhysics, latestFrame, ingressState, buildPublications: false).ConfigureAwait(false);
TraceStartup("first-tick");
latestFrame = firstTick.Frame;
initialPrepared = PreparePublication(
node.StatePath,
options,
tick,
firstTick,
ingressState,
physicsPersistence);
initialPublication = initialPrepared.Publication;
}
if (!publishRestoredFrame)
{
await PublishClientGameplayDocumentsAsync(
node,
options,
unityBundles,
initialPublication!,
ingressState.Catalog ?? throw new InvalidOperationException("Aetheria runtime catalog was not initialized."),
reactiveSurfaceState,
publishTopology: true).ConfigureAwait(false);
}
TraceStartup("client-bootstrap");
ThrowIfClientHostFaulted(cultMeshClientHost);
await PublishHotEntityStateAsync(
node, soaPublisher, hotState, latestFrame!, ingressState.Catalog, cultMeshClientHost).ConfigureAwait(false);
TraceStartup("hot-entity-state");
if (publishRestoredFrame)
{
publicationTask = Task.CompletedTask;
}
else
{
publicationTask = Task.Run(() => PersistPreparedDocumentsAsync(
node,
options,
physicsPersistence,
publication,
initialPrepared!,
initialTopology: true));
}
nextApiPublicationUtc = DateTimeOffset.UtcNow.Add(options.ApiPublicationInterval);
Console.WriteLine($"Aetheria Verse daemon published frame {latestFrame!.FrameId}.");
Console.WriteLine($"Aetheria daemon playable-world-ready in {startup.Elapsed.TotalMilliseconds:0.###}ms.");
if (options.Once)
{
await publicationTask.ConfigureAwait(false);
await PublishRuntimeSessionAsync(node, options, startedAtUtc, "completed").ConfigureAwait(false);
return;
}
await PublishRuntimeSessionAsync(node, options, startedAtUtc, "running").ConfigureAwait(false);
Console.WriteLine("Aetheria Verse daemon is running. Press Ctrl+C to stop.");
var nextTickUtc = DateTimeOffset.UtcNow.Add(options.TickInterval);
while (!stopped.Task.IsCompleted && (options.UseTerminusFixture || playableWorldDemand.IsActive))
{
ThrowIfClientHostFaulted(cultMeshClientHost);
var delay = nextTickUtc - DateTimeOffset.UtcNow;
if (delay > TimeSpan.Zero)
{
var completed = await Task.WhenAny(stopped.Task, Task.Delay(delay)).ConfigureAwait(false);
if (completed == stopped.Task)
break;
}
var buildPublications = DateTimeOffset.UtcNow >= nextApiPublicationUtc;
var tick = await TickAsync(node, options, unityBundles, worldPhysics, latestFrame, ingressState, buildPublications: false).ConfigureAwait(false);
ThrowIfClientHostFaulted(cultMeshClientHost);
latestFrame = tick.Frame;
await PublishHotEntityStateAsync(
node, soaPublisher, hotState, tick.Frame, ingressState.Catalog, cultMeshClientHost).ConfigureAwait(false);
nextTickUtc += options.TickInterval;
if (nextTickUtc < DateTimeOffset.UtcNow - options.TickInterval)
nextTickUtc = DateTimeOffset.UtcNow;
if (buildPublications && publicationTask.IsCompleted)
{
if (publicationTask.IsFaulted)
await publicationTask.ConfigureAwait(false);
var publication = PreparePublication(
node.StatePath,
options,
tick,
ingressState,
physicsPersistence);
publicationTask = Task.Run(() => PersistPreparedDocumentsAsync(
node,
options,
physicsPersistence,
publication,
initialTopology: false,
unityBundles: unityBundles,
reactiveSurfaceState: reactiveSurfaceState));
await PublishDemandedManagedViewportsAsync(node, options, tick.Frame, managedViewportDemand).ConfigureAwait(false);
nextApiPublicationUtc = DateTimeOffset.UtcNow.Add(options.ApiPublicationInterval);
}
if (tick.Frame.FrameId % (traceClientTransport ? 10 : 120) == 0)
{
Console.WriteLine(
$"Aetheria Verse daemon published frame {tick.Frame.FrameId} at {tick.Frame.SimulationTimeSeconds:0.00}s; " +
$"control peers={cultMeshClientHost.Control.PeerCount} " +
$"quic peers={cultMeshClientHost.Realtime.ConnectionCount} " +
$"cdn chunks={cultMeshClientHost.Content.ChunkRequestsServed}.");
}
}
await publicationTask.ConfigureAwait(false);
if (!stopped.Task.IsCompleted && !options.UseTerminusFixture)
{
var checkpointRun = latestFrame?.Run ?? throw new InvalidOperationException("Aetheria cannot checkpoint an inactive playable session without its run.");
var checkpointTick = new AetheriaRuntimeDaemonTickResult(
checkpointRun,
new AetheriaRuntimeDaemonOperationResult(checkpointRun, [], []),
latestFrame!);
var checkpoint = PreparePublication(
node.StatePath,
options,
checkpointTick,
ingressState,
physicsPersistence);
await PersistPreparedDocumentsAsync(
node,
options,
physicsPersistence,
checkpoint,
initialTopology: false,
unityBundles: unityBundles,
reactiveSurfaceState: reactiveSurfaceState));
await PublishDemandedManagedViewportsAsync(node, options, tick.Frame, managedViewportDemand).ConfigureAwait(false);
nextApiPublicationUtc = DateTimeOffset.UtcNow.Add(options.ApiPublicationInterval);
}
if (tick.Frame.FrameId % (traceClientTransport ? 10 : 120) == 0)
{
Console.WriteLine(
$"Aetheria Verse daemon published frame {tick.Frame.FrameId} at {tick.Frame.SimulationTimeSeconds:0.00}s; " +
$"control peers={cultMeshClientHost.Control.PeerCount} " +
$"quic peers={cultMeshClientHost.Realtime.ConnectionCount} " +
$"cdn chunks={cultMeshClientHost.Content.ChunkRequestsServed}.");
reactiveSurfaceState: reactiveSurfaceState).ConfigureAwait(false);
Console.WriteLine($"Aetheria daemon saved frame {latestFrame!.FrameId} after the playable client disconnected.");
}
}
await PublishRuntimeSessionAsync(node, options, startedAtUtc, "stopping").ConfigureAwait(false);
await publicationTask.ConfigureAwait(false);
await PublishRuntimeSessionAsync(node, options, startedAtUtc, "stopped").ConfigureAwait(false);
Console.WriteLine("Aetheria Verse daemon stopping.");
static void ThrowIfClientHostFaulted(AetheriaClientCultMeshHost host)
@ -300,15 +319,6 @@ static void ThrowIfClientHostFaulted(AetheriaClientCultMeshHost host)
host.Realtime.BackgroundFailure.GetAwaiter().GetResult());
}
static bool RequestsPlayableWorld(CultNetDatabaseSubscriptionDemand demand) =>
demand.Active &&
(demand.RecordKeys.Any(key =>
string.Equals(key, AetheriaRuntimeVerseRecordKeys.DaemonFrameLatest.ToString(), StringComparison.Ordinal) ||
string.Equals(key, AetheriaRuntimeVerseRecordKeys.DaemonGameSurface.ToString(), StringComparison.Ordinal) ||
string.Equals(key, AetheriaRuntimeVerseRecordKeys.DaemonGameReactiveSurface.ToString(), StringComparison.Ordinal)) ||
demand.SchemaIds.Contains(AetheriaRuntimeDaemonSchemas.Frame, StringComparer.Ordinal) ||
demand.BodyIds.Contains(AetheriaRuntimeDaemonSoaFramePublisher.BodyId, StringComparer.Ordinal));
static async Task<AetheriaRuntimeDaemonTickResult> TickAsync(
AetheriaStateNode node,
AetheriaDaemonHostOptions options,
@ -3753,6 +3763,34 @@ internal sealed record AetheriaPreparedPublication(
IReadOnlyList<AetheriaYmirZonePersistenceCapture> Physics,
AetheriaRuntimeCatalogSnapshot Catalog);
internal sealed class AetheriaPlayableWorldDemandState
{
private readonly ConcurrentDictionary<string, byte> _active = new(StringComparer.Ordinal);
public bool IsActive => !_active.IsEmpty;
public void Observe(CultNetDatabaseSubscriptionDemand demand)
{
var key = demand.ConsumerRuntimeId + "\u001f" + demand.SubscriptionId;
if (!RequestsPlayableWorld(demand))
{
_active.TryRemove(key, out _);
return;
}
_active[key] = 0;
}
private static bool RequestsPlayableWorld(CultNetDatabaseSubscriptionDemand demand) =>
demand.Active &&
(demand.RecordKeys.Any(key =>
string.Equals(key, AetheriaRuntimeVerseRecordKeys.DaemonFrameLatest.ToString(), StringComparison.Ordinal) ||
string.Equals(key, AetheriaRuntimeVerseRecordKeys.DaemonGameSurface.ToString(), StringComparison.Ordinal) ||
string.Equals(key, AetheriaRuntimeVerseRecordKeys.DaemonGameReactiveSurface.ToString(), StringComparison.Ordinal)) ||
demand.SchemaIds.Contains(AetheriaRuntimeDaemonSchemas.Frame, StringComparer.Ordinal) ||
demand.BodyIds.Contains(AetheriaRuntimeDaemonSoaFramePublisher.BodyId, StringComparer.Ordinal));
}
internal sealed class AetheriaManagedViewportDemandState
{
private readonly ConcurrentDictionary<string, AetheriaManagedViewportDemand> _active =

8
Aetheria.State.Verify/Program.cs
View file

@ -1343,7 +1343,7 @@ static void RequireClientTransportPlaneOwnership(string root)
"if (!options.Once && !options.UseTerminusFixture)",
StringComparison.Ordinal);
var demandIndex = daemon.IndexOf(
"var playableWorldRequested = new TaskCompletionSource<object?>",
"var playableWorldDemand = new AetheriaPlayableWorldDemandState()",
StringComparison.Ordinal);
var gameSessionIndex = daemon.IndexOf(
"await EnsureGameSessionAsync(node, options, startedAtUtc, latestFrame)",
@ -1352,10 +1352,12 @@ static void RequireClientTransportPlaneOwnership(string root)
clientHostIndex > readyGateIndex || demandIndex > readyGateIndex ||
readyGateIndex > gameSessionIndex || gameSessionIndex > restoreIndex ||
!daemon.Contains("Aetheria daemon ready; waiting for a client to load or generate a world.", StringComparison.Ordinal) ||
!daemon.Contains("if (playableWorldRequested.Task.IsCompleted", StringComparison.Ordinal) ||
!daemon.Contains("if (playableWorldDemand.IsActive", StringComparison.Ordinal) ||
!daemon.Contains("while (!stopped.Task.IsCompleted && (options.UseTerminusFixture || playableWorldDemand.IsActive))", StringComparison.Ordinal) ||
!daemon.Contains("Aetheria daemon saved frame", StringComparison.Ordinal) ||
!daemon.Contains("if (await ApplyRequestedNewGameSessionAsync(node, options)", StringComparison.Ordinal))
throw new InvalidOperationException(
"Aetheria daemon startup must expose transport in ready mode and defer game-session/Ymir activation until playable-world demand or New Game.");
"Aetheria daemon must expose transport in ready mode, demand-gate game/Ymir activation, and checkpoint before returning to ready after the last playable client drops.");
var requiredEditorResetSymbols = new[]
{

7
docs/unity-daemon-development.md
View file

@ -27,8 +27,11 @@ Starting the daemon does not create, load, or step a game world. It starts the
typed state boundary and CultMesh transport in `ready` mode. A playable-world
subscription loads an existing run; the advertised New Game command generates
a new run. Only that activation boundary opens Ymir persistence and starts the
fixed simulation clock. A dropped client therefore leaves the daemon and saved
world alive without making Unity Play Mode a lifecycle authority.
fixed simulation clock. When the last playable-world subscription is withdrawn,
the daemon commits the final public frame and matching private Ymir checkpoint,
releases the active physics world, and returns to `ready`. A dropped client
therefore leaves the daemon and saved world alive without making Unity Play
Mode a lifecycle authority.
The editor connects through the daemon's `cultnet+tcp` control endpoint. The
advertised session then selects the dedicated content and QUIC realtime planes;