From 7e8acb74dd20931d5ba88430fd100cd074a32e4d Mon Sep 17 00:00:00 2001 From: Erik Date: Mon, 27 Jul 2026 07:51:49 +0200 Subject: [PATCH] feat(headless): add deterministic multi-session scheduler --- .../Hosting/HeadlessMonotonicTime.cs | 25 ++ .../Hosting/HeadlessProcessHost.cs | 113 ++++--- .../Hosting/HeadlessProcessScheduler.cs | 249 ++++++++++++++++ .../Hosting/HeadlessSessionHost.cs | 77 ++++- .../Policies/HeadlessBotPolicy.cs | 4 +- .../HeadlessProcessSchedulerTests.cs | 280 ++++++++++++++++++ 6 files changed, 694 insertions(+), 54 deletions(-) create mode 100644 src/AcDream.Headless/Hosting/HeadlessMonotonicTime.cs create mode 100644 src/AcDream.Headless/Hosting/HeadlessProcessScheduler.cs create mode 100644 tests/AcDream.Headless.Tests/HeadlessProcessSchedulerTests.cs diff --git a/src/AcDream.Headless/Hosting/HeadlessMonotonicTime.cs b/src/AcDream.Headless/Hosting/HeadlessMonotonicTime.cs new file mode 100644 index 00000000..8740db57 --- /dev/null +++ b/src/AcDream.Headless/Hosting/HeadlessMonotonicTime.cs @@ -0,0 +1,25 @@ +namespace AcDream.Headless.Hosting; + +internal static class HeadlessMonotonicTime +{ + internal static long Add( + TimeProvider timeProvider, + long timestamp, + TimeSpan duration) + { + ArgumentNullException.ThrowIfNull(timeProvider); + if (duration < TimeSpan.Zero) + throw new ArgumentOutOfRangeException(nameof(duration)); + + double ticks = + duration.TotalSeconds * timeProvider.TimestampFrequency; + if (!double.IsFinite(ticks) || ticks > long.MaxValue) + return long.MaxValue; + long delta = Math.Max( + 1L, + checked((long)Math.Ceiling(ticks))); + return timestamp > long.MaxValue - delta + ? long.MaxValue + : timestamp + delta; + } +} diff --git a/src/AcDream.Headless/Hosting/HeadlessProcessHost.cs b/src/AcDream.Headless/Hosting/HeadlessProcessHost.cs index 04a4c64e..4edc4910 100644 --- a/src/AcDream.Headless/Hosting/HeadlessProcessHost.cs +++ b/src/AcDream.Headless/Hosting/HeadlessProcessHost.cs @@ -1,4 +1,3 @@ -using System.Diagnostics; using AcDream.Headless.Configuration; using AcDream.Headless.Credentials; using AcDream.Headless.Diagnostics; @@ -10,12 +9,10 @@ namespace AcDream.Headless.Hosting; internal sealed class HeadlessProcessHost : IDisposable { - private static readonly TimeSpan K1TickInterval = - TimeSpan.FromMilliseconds(15); - - private readonly HeadlessSessionHost _session; + private readonly HeadlessSessionHost[] _sessions; + private readonly HeadlessProcessScheduler _scheduler; private readonly HeadlessDiagnosticWriter _diagnostics; - private readonly string _sessionId; + private int _disposeIndex; private bool _disposed; internal HeadlessProcessHost( @@ -30,67 +27,97 @@ internal sealed class HeadlessProcessHost : IDisposable ArgumentNullException.ThrowIfNull(paths); ArgumentNullException.ThrowIfNull(standardInput); ArgumentNullException.ThrowIfNull(diagnostics); - if (configuration.Sessions.Count != 1 - || configuration.Sessions[0] is not { } descriptor) + if (configuration.Sessions.Count == 0) { throw new HeadlessConfigurationException( - "Slice K1 run mode requires exactly one session."); + "Headless run mode requires at least one session."); } - _sessionId = descriptor.Id; _diagnostics = new HeadlessDiagnosticWriter(diagnostics); var credentials = new HeadlessCredentialResolver( standardInput, paths.ConfigDirectory); - HeadlessCredentialSecret secret = - credentials.Resolve(descriptor.Id, descriptor.Credential); + var sessions = new List( + configuration.Sessions.Count); try { - _session = new HeadlessSessionHost( - descriptor, - secret, - _diagnostics, - sessionOperations, + foreach (HeadlessSessionDescriptor? candidate + in configuration.Sessions) + { + HeadlessSessionDescriptor descriptor = candidate + ?? throw new HeadlessConfigurationException( + "Headless run mode cannot contain a null session."); + HeadlessCredentialSecret secret = + credentials.Resolve( + descriptor.Id, + descriptor.Credential); + try + { + sessions.Add(new HeadlessSessionHost( + descriptor, + secret, + _diagnostics, + sessionOperations, + timeProvider)); + } + catch + { + secret.Dispose(); + throw; + } + } + + _sessions = sessions.ToArray(); + _scheduler = new HeadlessProcessScheduler( + _sessions, timeProvider); + _disposeIndex = _sessions.Length - 1; } catch { - secret.Dispose(); + for (int index = sessions.Count - 1; index >= 0; index--) + sessions[index].Dispose(); throw; } } - internal HeadlessSessionHost Session => _session; + internal HeadlessSessionHost Session => _sessions.Length == 1 + ? _sessions[0] + : throw new InvalidOperationException( + "The process owns more than one headless session."); + internal IReadOnlyList Sessions => _sessions; + internal HeadlessSchedulerSnapshot Scheduler => + _scheduler.CaptureSnapshot(); internal async Task RunAsync( CancellationToken cancellationToken) { ObjectDisposedException.ThrowIf(_disposed, this); - RuntimeSessionStartResult started = _session.Start(); - if (started.Status != RuntimeSessionStartStatus.Connected) + foreach (HeadlessSessionHost session in _sessions) { - if (started.Error is { } error) - _diagnostics.Failure(_sessionId, "start", error); - return HeadlessExitCode.ConnectionError; + RuntimeSessionStartResult started = session.Start(); + if (started.Status != RuntimeSessionStartStatus.Connected) + { + if (started.Error is { } error) + { + _diagnostics.Failure( + session.SessionId, + "start", + error); + } + return HeadlessExitCode.ConnectionError; + } + + _diagnostics.Lifecycle( + session.SessionId, + "running", + session.Runtime); } - _diagnostics.Lifecycle(_sessionId, "running", _session.Runtime); - long previous = Stopwatch.GetTimestamp(); try { - while (!cancellationToken.IsCancellationRequested) - { - await Task.Delay( - K1TickInterval, - cancellationToken).ConfigureAwait(false); - long current = Stopwatch.GetTimestamp(); - double deltaSeconds = - (current - previous) / (double)Stopwatch.Frequency; - previous = current; - _session.Tick(deltaSeconds); - if (_session.IsPolicyComplete) - break; - } + await _scheduler.RunAsync(cancellationToken) + .ConfigureAwait(false); } catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) @@ -98,7 +125,7 @@ internal sealed class HeadlessProcessHost : IDisposable } catch (Exception error) { - _diagnostics.Failure(_sessionId, "tick", error); + _diagnostics.Failure("process", "tick", error); return HeadlessExitCode.RuntimeError; } @@ -109,7 +136,11 @@ internal sealed class HeadlessProcessHost : IDisposable { if (_disposed) return; - _session.Dispose(); + while (_disposeIndex >= 0) + { + _sessions[_disposeIndex].Dispose(); + _disposeIndex--; + } _disposed = true; } } diff --git a/src/AcDream.Headless/Hosting/HeadlessProcessScheduler.cs b/src/AcDream.Headless/Hosting/HeadlessProcessScheduler.cs new file mode 100644 index 00000000..43b336be --- /dev/null +++ b/src/AcDream.Headless/Hosting/HeadlessProcessScheduler.cs @@ -0,0 +1,249 @@ +using AcDream.Runtime; + +namespace AcDream.Headless.Hosting; + +internal sealed class HeadlessProcessScheduler +{ + private sealed class Slot + { + internal Slot( + HeadlessSessionHost session, + long startTimestamp, + long periodTicks) + { + Session = session; + LastTurnTimestamp = startTimestamp; + NextDeadline = AddTicks(startTimestamp, periodTicks); + } + + internal HeadlessSessionHost Session { get; } + internal long LastTurnTimestamp { get; set; } + internal long NextDeadline { get; set; } + } + + private readonly TimeProvider _timeProvider; + private readonly long _periodTicks; + private readonly int _maximumCatchUpTurns; + private readonly Slot[] _slots; + private long _waitCount; + private long _turnCount; + private long _catchUpCollapseCount; + + internal HeadlessProcessScheduler( + IReadOnlyList sessions, + TimeProvider? timeProvider = null, + TimeSpan? turnPeriod = null, + int maximumCatchUpTurns = 8) + { + ArgumentNullException.ThrowIfNull(sessions); + if (sessions.Count == 0) + { + throw new ArgumentException( + "A headless scheduler requires at least one session.", + nameof(sessions)); + } + if (maximumCatchUpTurns <= 0) + { + throw new ArgumentOutOfRangeException( + nameof(maximumCatchUpTurns)); + } + + _timeProvider = timeProvider ?? TimeProvider.System; + TimeSpan configuredTurnPeriod = + turnPeriod ?? TimeSpan.FromMilliseconds(15); + if (configuredTurnPeriod <= TimeSpan.Zero) + throw new ArgumentOutOfRangeException(nameof(turnPeriod)); + _periodTicks = DurationToTimestampTicks( + _timeProvider, + configuredTurnPeriod); + _maximumCatchUpTurns = maximumCatchUpTurns; + + long start = _timeProvider.GetTimestamp(); + _slots = new Slot[sessions.Count]; + for (int index = 0; index < sessions.Count; index++) + { + _slots[index] = new Slot( + sessions[index] + ?? throw new ArgumentException( + "A headless scheduler session cannot be null.", + nameof(sessions)), + start, + _periodTicks); + } + } + + internal HeadlessSchedulerSnapshot CaptureSnapshot() + { + int activeSessionCount = 0; + for (int index = 0; index < _slots.Length; index++) + { + if (!_slots[index].Session.IsPolicyComplete) + activeSessionCount++; + } + + return new HeadlessSchedulerSnapshot( + _slots.Length, + Interlocked.Read(ref _waitCount), + Interlocked.Read(ref _turnCount), + Interlocked.Read(ref _catchUpCollapseCount), + activeSessionCount, + NextDeadline()); + } + + internal async Task RunAsync(CancellationToken cancellationToken) + { + while (!cancellationToken.IsCancellationRequested + && HasActiveSession()) + { + long now = _timeProvider.GetTimestamp(); + if (DispatchDue(now)) + continue; + + long deadline = NextDeadline(); + TimeSpan delay = deadline <= now + ? TimeSpan.Zero + : _timeProvider.GetElapsedTime(now, deadline); + Interlocked.Increment(ref _waitCount); + await Task.Delay( + delay, + _timeProvider, + cancellationToken) + .ConfigureAwait(false); + } + } + + internal bool DispatchDue(long nowTimestamp) + { + bool dispatched = false; + foreach (Slot slot in _slots) + { + HeadlessSessionHost session = slot.Session; + if (session.IsPolicyComplete) + continue; + + if (session.IsReconnectPending) + { + if (nowTimestamp < session.ReconnectDeadline) + continue; + + RuntimeSessionStartResult result = + session.CompletePendingReconnect(nowTimestamp); + if (result.Status + is not RuntimeSessionStartStatus.Connected) + { + throw result.Error + ?? new InvalidOperationException( + $"Headless session reconnect completed with {result.Status}."); + } + slot.LastTurnTimestamp = nowTimestamp; + slot.NextDeadline = AddTicks( + nowTimestamp, + _periodTicks); + dispatched = true; + continue; + } + + if (nowTimestamp < slot.NextDeadline) + continue; + + DispatchSessionDue(slot, nowTimestamp); + dispatched = true; + } + return dispatched; + } + + private void DispatchSessionDue(Slot slot, long nowTimestamp) + { + long dueCount = + 1L + ((nowTimestamp - slot.NextDeadline) / _periodTicks); + int turns = (int)Math.Min( + dueCount, + _maximumCatchUpTurns); + bool collapse = dueCount > _maximumCatchUpTurns; + + for (int turn = 0; turn < turns; turn++) + { + if (slot.Session.IsPolicyComplete + || slot.Session.IsReconnectPending) + { + break; + } + + bool finalCollapsedTurn = + collapse && turn == turns - 1; + long targetTimestamp = finalCollapsedTurn + ? nowTimestamp + : slot.NextDeadline; + double deltaSeconds = _timeProvider + .GetElapsedTime( + slot.LastTurnTimestamp, + targetTimestamp) + .TotalSeconds; + slot.Session.Tick(deltaSeconds); + Interlocked.Increment(ref _turnCount); + slot.LastTurnTimestamp = targetTimestamp; + slot.NextDeadline = AddTicks( + targetTimestamp, + _periodTicks); + } + + if (collapse) + Interlocked.Increment(ref _catchUpCollapseCount); + } + + private long NextDeadline() + { + long next = long.MaxValue; + foreach (Slot slot in _slots) + { + if (slot.Session.IsPolicyComplete) + continue; + long candidate = slot.Session.IsReconnectPending + ? slot.Session.ReconnectDeadline + : slot.NextDeadline; + if (candidate < next) + next = candidate; + } + return next; + } + + private bool HasActiveSession() + { + for (int index = 0; index < _slots.Length; index++) + { + if (!_slots[index].Session.IsPolicyComplete) + return true; + } + return false; + } + + private static long DurationToTimestampTicks( + TimeProvider timeProvider, + TimeSpan duration) + { + double ticks = + duration.TotalSeconds * timeProvider.TimestampFrequency; + if (!double.IsFinite(ticks) + || ticks <= 0d + || ticks > long.MaxValue) + { + throw new ArgumentOutOfRangeException(nameof(duration)); + } + return Math.Max( + 1L, + checked((long)Math.Ceiling(ticks))); + } + + private static long AddTicks(long timestamp, long ticks) => + timestamp > long.MaxValue - ticks + ? long.MaxValue + : timestamp + ticks; +} + +internal readonly record struct HeadlessSchedulerSnapshot( + int SessionCount, + long WaitCount, + long TurnCount, + long CatchUpCollapseCount, + int ActiveSessionCount, + long NextDeadline); diff --git a/src/AcDream.Headless/Hosting/HeadlessSessionHost.cs b/src/AcDream.Headless/Hosting/HeadlessSessionHost.cs index 011b5ad2..4f8f2de2 100644 --- a/src/AcDream.Headless/Hosting/HeadlessSessionHost.cs +++ b/src/AcDream.Headless/Hosting/HeadlessSessionHost.cs @@ -1,4 +1,3 @@ -using System.Diagnostics; using AcDream.Headless.Configuration; using AcDream.Headless.Credentials; using AcDream.Headless.Diagnostics; @@ -47,12 +46,15 @@ internal sealed class HeadlessSessionHost : IDisposable private readonly HeadlessCredentialSecret _credential; private readonly HeadlessDiagnosticWriter _diagnostics; private readonly TimeSpan _reconnectQuiescence; + private readonly TimeProvider _timeProvider; private readonly HeadlessGenerationResetHost _resetHost = new(); private readonly IDisposable _hostLease; private readonly IHeadlessBotPolicy _policy; private readonly IDisposable _policySubscription; private readonly LiveSessionHost _liveSession; private int _disposeStage; + private long _reconnectDeadline; + private bool _reconnectPending; private ulong _stoppedGeneration; private bool _disposed; @@ -70,6 +72,7 @@ internal sealed class HeadlessSessionHost : IDisposable ?? throw new ArgumentNullException(nameof(credential)); _diagnostics = diagnostics ?? throw new ArgumentNullException(nameof(diagnostics)); + _timeProvider = timeProvider ?? TimeProvider.System; _reconnectQuiescence = reconnectQuiescence ?? (sessionOperations is null ? TimeSpan.FromMilliseconds(2500) @@ -92,7 +95,7 @@ internal sealed class HeadlessSessionHost : IDisposable gameplay, gameplay, gameplay, - TimeProvider: timeProvider, + TimeProvider: _timeProvider, Log: message => diagnostics.Message( descriptor.Id, message), @@ -169,9 +172,15 @@ internal sealed class HeadlessSessionHost : IDisposable internal GameRuntime Runtime { get; } internal DirectGameRuntimeCommandAdapter Commands { get; } + internal string SessionId => _descriptor.Id; internal string ActiveCharacterName { get; private set; } = string.Empty; internal bool IsPolicyComplete => _policy.IsComplete; + internal bool IsReconnectPending => _reconnectPending; + internal long ReconnectDeadline => _reconnectPending + ? _reconnectDeadline + : throw new InvalidOperationException( + "The headless session has no pending reconnect."); internal RuntimeSessionStartResult Start() => Commands.Session.Start(Runtime.Generation); @@ -182,6 +191,8 @@ internal sealed class HeadlessSessionHost : IDisposable internal void Tick(double deltaSeconds) { ObjectDisposedException.ThrowIf(_disposed, this); + if (_reconnectPending) + return; _ = Runtime.Clock.Advance(deltaSeconds); Runtime.Session.Tick(); Runtime.ActionOwner.CombatAttack.Tick(); @@ -191,6 +202,28 @@ internal sealed class HeadlessSessionHost : IDisposable internal RuntimeTeardownAcknowledgement Stop() => Commands.Session.Stop(Runtime.Generation); + internal RuntimeSessionStartResult CompletePendingReconnect( + long nowTimestamp) + { + ObjectDisposedException.ThrowIf(_disposed, this); + if (!_reconnectPending) + { + return new RuntimeSessionStartResult( + RuntimeSessionStartStatus.Inactive, + Runtime.Generation); + } + if (nowTimestamp < _reconnectDeadline) + { + return new RuntimeSessionStartResult( + RuntimeSessionStartStatus.Deferred, + Runtime.Generation); + } + + _reconnectPending = false; + _reconnectDeadline = 0L; + return StartLive(reconnect: true); + } + public void Dispose() { if (_disposed) @@ -202,6 +235,8 @@ internal sealed class HeadlessSessionHost : IDisposable { case 0: { + _reconnectPending = false; + _reconnectDeadline = 0L; RuntimeTeardownAcknowledgement stopped = Stop(); if (!stopped.IsComplete) { @@ -266,6 +301,12 @@ internal sealed class HeadlessSessionHost : IDisposable RuntimeSessionStartStatus.StaleGeneration, Runtime.Generation); } + if (_reconnectPending) + { + return new RuntimeSessionStartResult( + RuntimeSessionStartStatus.Deferred, + Runtime.Generation); + } if (reconnect) { @@ -281,19 +322,32 @@ internal sealed class HeadlessSessionHost : IDisposable "The prior headless session did not quiesce before reconnect.")); } - // ACE confirms CharacterLogOff before its account/session index - // releases the retiring socket. Starting a replacement in that - // short suffix makes ACE reject both sessions as Account In Use. - // This is a one-time connection-lifecycle wait, never a steady - // scheduler sleep; K2 moves it onto the absolute-deadline host. if (_reconnectQuiescence > TimeSpan.Zero) { - Task.Delay(_reconnectQuiescence) - .GetAwaiter() - .GetResult(); + // ACE confirms CharacterLogOff before its account/session + // index releases the retiring socket. The process scheduler + // owns this monotonic deadline so other sessions keep moving + // and no command or worker thread blocks. + _reconnectDeadline = HeadlessMonotonicTime.Add( + _timeProvider, + _timeProvider.GetTimestamp(), + _reconnectQuiescence); + _reconnectPending = true; + _diagnostics.Lifecycle( + _descriptor.Id, + "reconnect-deferred", + Runtime); + return new RuntimeSessionStartResult( + RuntimeSessionStartStatus.Deferred, + Runtime.Generation); } } + return StartLive(reconnect); + } + + private RuntimeSessionStartResult StartLive(bool reconnect) + { string password = _credential.Reveal(); try { @@ -366,8 +420,7 @@ internal sealed class HeadlessSessionHost : IDisposable OnConfirmationRequest: null, OnConfirmationDone: null, ClientTime: () => - Stopwatch.GetTimestamp() - / (double)Stopwatch.Frequency), + Runtime.Clock.SimulationTimeSeconds), new LiveSocialSessionBindings( Runtime.CommunicationOwner.Chat, Runtime.CommunicationOwner.TurbineChat, diff --git a/src/AcDream.Headless/Policies/HeadlessBotPolicy.cs b/src/AcDream.Headless/Policies/HeadlessBotPolicy.cs index 41f02a99..5a966537 100644 --- a/src/AcDream.Headless/Policies/HeadlessBotPolicy.cs +++ b/src/AcDream.Headless/Policies/HeadlessBotPolicy.cs @@ -113,7 +113,9 @@ internal sealed class LifecycleSmokeHeadlessBotPolicy RuntimeSessionStartResult reconnect = commands.Session.Reconnect(view.Generation); if (reconnect.Status - != RuntimeSessionStartStatus.Connected) + is not ( + RuntimeSessionStartStatus.Connected + or RuntimeSessionStartStatus.Deferred)) { throw new InvalidOperationException( $"Lifecycle gate reconnect was rejected with {reconnect.Status}."); diff --git a/tests/AcDream.Headless.Tests/HeadlessProcessSchedulerTests.cs b/tests/AcDream.Headless.Tests/HeadlessProcessSchedulerTests.cs new file mode 100644 index 00000000..eee8c675 --- /dev/null +++ b/tests/AcDream.Headless.Tests/HeadlessProcessSchedulerTests.cs @@ -0,0 +1,280 @@ +using System.Net; +using AcDream.Core.Net; +using AcDream.Core.Net.Messages; +using AcDream.Headless.Configuration; +using AcDream.Headless.Credentials; +using AcDream.Headless.Diagnostics; +using AcDream.Headless.Hosting; +using AcDream.Headless.Platform; +using AcDream.Runtime; +using AcDream.Runtime.Session; + +namespace AcDream.Headless.Tests; + +public sealed class HeadlessProcessSchedulerTests +{ + [Fact] + public void AbsoluteDeadlinesTickSessionsInStableElapsedTime() + { + var time = new ManualTimeProvider(); + var operations = new FixtureSessionOperations(); + using HeadlessSessionHost first = + CreateSession("first", time, operations); + using HeadlessSessionHost second = + CreateSession("second", time, operations); + Assert.Equal( + RuntimeSessionStartStatus.Connected, + first.Start().Status); + Assert.Equal( + RuntimeSessionStartStatus.Connected, + second.Start().Status); + var scheduler = new HeadlessProcessScheduler( + [first, second], + time); + + Assert.False(scheduler.DispatchDue(time.GetTimestamp())); + time.Advance(TimeSpan.FromMilliseconds(15)); + + Assert.True(scheduler.DispatchDue(time.GetTimestamp())); + Assert.Equal(1UL, first.Runtime.Clock.FrameNumber); + Assert.Equal(1UL, second.Runtime.Clock.FrameNumber); + Assert.Equal( + 0.015d, + first.Runtime.Clock.SimulationTimeSeconds, + precision: 9); + HeadlessSchedulerSnapshot snapshot = + scheduler.CaptureSnapshot(); + Assert.Equal(2, snapshot.TurnCount); + Assert.Equal(0, snapshot.CatchUpCollapseCount); + } + + [Fact] + public void LongPauseUsesBoundedCatchUpAndPreservesElapsedTime() + { + var time = new ManualTimeProvider(); + var operations = new FixtureSessionOperations(); + using HeadlessSessionHost session = + CreateSession("catch-up", time, operations); + Assert.Equal( + RuntimeSessionStartStatus.Connected, + session.Start().Status); + var scheduler = new HeadlessProcessScheduler( + [session], + time, + maximumCatchUpTurns: 4); + + time.Advance(TimeSpan.FromSeconds(1)); + Assert.True(scheduler.DispatchDue(time.GetTimestamp())); + + Assert.Equal(4UL, session.Runtime.Clock.FrameNumber); + Assert.Equal( + 1d, + session.Runtime.Clock.SimulationTimeSeconds, + precision: 9); + Assert.Equal( + 1, + scheduler.CaptureSnapshot().CatchUpCollapseCount); + } + + [Fact] + public void ReconnectQuiescenceIsAMonotonicDeadlineNotABlockingWait() + { + var time = new ManualTimeProvider(); + var operations = new FixtureSessionOperations(); + using HeadlessSessionHost session = CreateSession( + "reconnect", + time, + operations, + TimeSpan.FromMilliseconds(2500)); + Assert.Equal( + RuntimeSessionStartStatus.Connected, + session.Start().Status); + var scheduler = new HeadlessProcessScheduler( + [session], + time); + + RuntimeSessionStartResult reconnect = session.Reconnect(); + + Assert.Equal( + RuntimeSessionStartStatus.Deferred, + reconnect.Status); + Assert.True(session.IsReconnectPending); + Assert.Equal(1, operations.CreatedSessionCount); + Assert.Equal(1, operations.DisposedSessionCount); + + time.Advance(TimeSpan.FromMilliseconds(2499)); + Assert.False(scheduler.DispatchDue(time.GetTimestamp())); + Assert.Equal(1, operations.CreatedSessionCount); + + time.Advance(TimeSpan.FromMilliseconds(1)); + Assert.True(scheduler.DispatchDue(time.GetTimestamp())); + Assert.False(session.IsReconnectPending); + Assert.Equal(2, operations.CreatedSessionCount); + Assert.True(session.Runtime.Session.IsInWorld); + } + + [Fact] + public async Task ProcessHostRunsMultipleSessionsOnOneScheduler() + { + var configuration = new HeadlessConfiguration + { + Version = 1, + Sessions = + [ + Descriptor( + "first", + "first-stdin"), + Descriptor( + "second", + "second-stdin"), + ], + }; + var operations = new FixtureSessionOperations(); + using var diagnostics = new StringWriter(); + using var host = new HeadlessProcessHost( + configuration, + HeadlessPathSet.Resolve(new HeadlessPathOverrides()), + new StringReader( + "first-password" + + Environment.NewLine + + "second-password" + + Environment.NewLine), + diagnostics, + operations); + using var cancellation = new CancellationTokenSource(); + cancellation.Cancel(); + + HeadlessExitCode result = + await host.RunAsync(cancellation.Token); + + Assert.Equal(HeadlessExitCode.Success, result); + Assert.Equal(2, host.Sessions.Count); + Assert.All( + host.Sessions, + session => Assert.True(session.Runtime.Session.IsInWorld)); + Assert.Equal(2, operations.CreatedSessionCount); + Assert.DoesNotContain( + "password", + diagnostics.ToString(), + StringComparison.Ordinal); + } + + private static HeadlessSessionHost CreateSession( + string id, + TimeProvider time, + ILiveSessionOperations operations, + TimeSpan? reconnectQuiescence = null) + { + var credential = new HeadlessCredentialSecret( + $"{id}-credential", + "password"); + try + { + return new HeadlessSessionHost( + Descriptor(id, $"{id}-credential"), + credential, + new HeadlessDiagnosticWriter(TextWriter.Null), + operations, + time, + reconnectQuiescence); + } + catch + { + credential.Dispose(); + throw; + } + } + + private static HeadlessSessionDescriptor Descriptor( + string id, + string credentialReference) => new() + { + Id = id, + Endpoint = new HeadlessEndpointDescriptor + { + Host = "127.0.0.1", + Port = 9000, + }, + Account = $"{id}-account", + Character = new HeadlessCharacterSelector + { + Index = 0, + }, + Policy = new HeadlessBotPolicyDescriptor + { + Id = "idle", + }, + Credential = new HeadlessCredentialReference + { + Provider = HeadlessCredentialProviderKind.StandardInput, + Reference = credentialReference, + }, + }; + + private sealed class ManualTimeProvider : TimeProvider + { + private long _timestamp; + + public override long TimestampFrequency => + TimeSpan.TicksPerSecond; + + public override long GetTimestamp() => _timestamp; + + internal void Advance(TimeSpan duration) => + _timestamp = checked(_timestamp + duration.Ticks); + } + + private sealed class FixtureSessionOperations : ILiveSessionOperations + { + public int CreatedSessionCount { get; private set; } + public int DisposedSessionCount { get; private set; } + + public IPEndPoint ResolveEndpoint(string host, int port) => + new(IPAddress.Loopback, port); + + public WorldSession CreateSession(IPEndPoint endpoint) + { + CreatedSessionCount++; + return new WorldSession(endpoint); + } + + public void Connect( + WorldSession session, + string user, + string password) + { + } + + public CharacterList.Parsed GetCharacters( + WorldSession session) => + new( + 0u, + [ + new CharacterList.Character( + 0x50000001u, + "Headless", + 0u), + ], + [], + 11, + "account", + true, + true); + + public void EnterWorld( + WorldSession session, + int activeCharacterIndex) + { + } + + public void Tick(WorldSession session) + { + } + + public void DisposeSession(WorldSession session) + { + DisposedSessionCount++; + session.Dispose(); + } + } +}