Refactor online player count detection in Agent

Caused by 72ddaf91.

Agent/Phantom.Agent.Minecraft/Properties/MinecraftServerProperties.cs

@@ -18,4 +18,5 @@ static class MinecraftServerProperties {
 	public static readonly MinecraftServerProperty<ushort> ServerPort = new UnsignedShort("server-port");
 	public static readonly MinecraftServerProperty<ushort> RconPort = new UnsignedShort("rcon.port");
 	public static readonly MinecraftServerProperty<bool> EnableRcon = new Boolean("enable-rcon");
+	public static readonly MinecraftServerProperty<bool> SyncChunkWrites = new Boolean("sync-chunk-writes");
 }

Agent/Phantom.Agent.Minecraft/Properties/ServerProperties.cs

@@ -5,11 +5,13 @@ namespace Phantom.Agent.Minecraft.Properties;
 public sealed record ServerProperties(
 	ushort ServerPort,
 	ushort RconPort,
-	bool EnableRcon = true
+	bool EnableRcon = true,
+	bool SyncChunkWrites = false
 ) {
 	internal void SetTo(JavaPropertiesFileEditor properties) {
 		MinecraftServerProperties.ServerPort.Set(properties, ServerPort);
 		MinecraftServerProperties.RconPort.Set(properties, RconPort);
 		MinecraftServerProperties.EnableRcon.Set(properties, EnableRcon);
+		MinecraftServerProperties.SyncChunkWrites.Set(properties, SyncChunkWrites);
 	}
 }

Agent/Phantom.Agent.Minecraft/Server/ServerStatusProtocol.cs

@@ -3,28 +3,11 @@ using System.Buffers.Binary;
 using System.Net;
 using System.Net.Sockets;
 using System.Text;
-using Phantom.Utils.Logging;
-using Serilog;
 
 namespace Phantom.Agent.Minecraft.Server;
 
-public sealed class ServerStatusProtocol {
-	private readonly ILogger logger;
-
-	public ServerStatusProtocol(string loggerName) {
-		this.logger = PhantomLogger.Create<ServerStatusProtocol>(loggerName);
-	}
-
-	public async Task<int?> GetOnlinePlayerCount(int serverPort, CancellationToken cancellationToken) {
-		try {
-			return await GetOnlinePlayerCountOrThrow(serverPort, cancellationToken);
-		} catch (Exception e) {
-			logger.Error(e, "Caught exception while checking if players are online.");
-			return null;
-		}
-	}
-
-	private async Task<int?> GetOnlinePlayerCountOrThrow(int serverPort, CancellationToken cancellationToken) {
+public static class ServerStatusProtocol {
+	public static async Task<int> GetOnlinePlayerCount(ushort serverPort, CancellationToken cancellationToken) {
 		using var tcpClient = new TcpClient();
 		await tcpClient.ConnectAsync(IPAddress.Loopback, serverPort, cancellationToken);
 		var tcpStream = tcpClient.GetStream();
@@ -33,24 +16,22 @@ public sealed class ServerStatusProtocol {
 		tcpStream.WriteByte(0xFE);
 		await tcpStream.FlushAsync(cancellationToken);
 
-		short? messageLength = await ReadStreamHeader(tcpStream, cancellationToken);
-		return messageLength == null ? null : await ReadOnlinePlayerCount(tcpStream, messageLength.Value * 2, cancellationToken);
+		short messageLength = await ReadStreamHeader(tcpStream, cancellationToken);
+		return await ReadOnlinePlayerCount(tcpStream, messageLength * 2, cancellationToken);
 	}
 
-	private async Task<short?> ReadStreamHeader(NetworkStream tcpStream, CancellationToken cancellationToken) {
+	private static async Task<short> ReadStreamHeader(NetworkStream tcpStream, CancellationToken cancellationToken) {
 		var headerBuffer = ArrayPool<byte>.Shared.Rent(3);
 		try {
 			await tcpStream.ReadExactlyAsync(headerBuffer, 0, 3, cancellationToken);
 
 			if (headerBuffer[0] != 0xFF) {
-				logger.Error("Unexpected first byte in response from server: {FirstByte}.", headerBuffer[0]);
-				return null;
+				throw new ProtocolException("Unexpected first byte in response from server: " + headerBuffer[0]);
 			}
 
 			short messageLength = BinaryPrimitives.ReadInt16BigEndian(headerBuffer.AsSpan(1));
 			if (messageLength <= 0) {
-				logger.Error("Unexpected message length in response from server: {MessageLength}.", messageLength);
-				return null;
+				throw new ProtocolException("Unexpected message length in response from server: " + messageLength);
 			}
 			
 			return messageLength;
@@ -59,7 +40,7 @@ public sealed class ServerStatusProtocol {
 		}
 	}
 
-	private async Task<int?> ReadOnlinePlayerCount(NetworkStream tcpStream, int messageLength, CancellationToken cancellationToken) {
+	private static async Task<int> ReadOnlinePlayerCount(NetworkStream tcpStream, int messageLength, CancellationToken cancellationToken) {
 		var messageBuffer = ArrayPool<byte>.Shared.Rent(messageLength);
 		try {
 			await tcpStream.ReadExactlyAsync(messageBuffer, 0, messageLength, cancellationToken);
@@ -74,20 +55,21 @@ public sealed class ServerStatusProtocol {
 			int separator2 = Array.LastIndexOf(messageBuffer, SeparatorSecondByte);
 			int separator1 = separator2 == -1 ? -1 : Array.LastIndexOf(messageBuffer, SeparatorSecondByte, separator2 - 1);
 			if (!IsValidSeparator(messageBuffer, separator1) || !IsValidSeparator(messageBuffer, separator2)) {
-				logger.Error("Could not find message separators in response from server.");
-				return null;
+				throw new ProtocolException("Could not find message separators in response from server.");
 			}
 
 			string onlinePlayerCountStr = Encoding.BigEndianUnicode.GetString(messageBuffer.AsSpan((separator1 + 1)..(separator2 - 1)));
 			if (!int.TryParse(onlinePlayerCountStr, out int onlinePlayerCount)) {
-				logger.Error("Could not parse online player count in response from server: {OnlinePlayerCount}.", onlinePlayerCountStr);
-				return null;
+				throw new ProtocolException("Could not parse online player count in response from server: " + onlinePlayerCountStr);
 			}
 			
-			logger.Debug("Detected {OnlinePlayerCount} online player(s).", onlinePlayerCount);
 			return onlinePlayerCount;
 		} finally {
 			ArrayPool<byte>.Shared.Return(messageBuffer);
 		}
 	}
+	
+	public sealed class ProtocolException : Exception {
+		internal ProtocolException(string message) : base(message) {}
+	}
 }

Agent/Phantom.Agent.Services/Backups/BackupScheduler.cs

@@ -1,10 +1,8 @@
-using Phantom.Agent.Minecraft.Instance;
-using Phantom.Agent.Minecraft.Server;
-using Phantom.Agent.Services.Instances;
+using Phantom.Agent.Services.Instances;
+using Phantom.Agent.Services.Instances.State;
 using Phantom.Common.Data.Backups;
 using Phantom.Utils.Logging;
 using Phantom.Utils.Tasks;
-using Phantom.Utils.Threading;
 
 namespace Phantom.Agent.Services.Backups;
 
@@ -16,27 +14,23 @@ sealed class BackupScheduler : CancellableBackgroundTask {
 
 	private readonly BackupManager backupManager;
 	private readonly InstanceContext context;
-	private readonly InstanceProcess process;
 	private readonly SemaphoreSlim backupSemaphore = new (1, 1);
-	private readonly int serverPort;
-	private readonly ServerStatusProtocol serverStatusProtocol;
 	private readonly ManualResetEventSlim serverOutputWhileWaitingForOnlinePlayers = new ();
+	private readonly InstancePlayerCountTracker playerCountTracker;
 	
 	public event EventHandler<BackupCreationResult>? BackupCompleted;
 
-	public BackupScheduler(InstanceContext context, InstanceProcess process, int serverPort) : base(PhantomLogger.Create<BackupScheduler>(context.ShortName)) {
+	public BackupScheduler(InstanceContext context, InstancePlayerCountTracker playerCountTracker) : base(PhantomLogger.Create<BackupScheduler>(context.ShortName)) {
 		this.backupManager = context.Services.BackupManager;
 		this.context = context;
-		this.process = process;
-		this.serverPort = serverPort;
-		this.serverStatusProtocol = new ServerStatusProtocol(context.ShortName);
+		this.playerCountTracker = playerCountTracker;
 		Start();
 	}
 
 	protected override async Task RunTask() {
 		await Task.Delay(InitialDelay, CancellationToken);
 		Logger.Information("Starting a new backup after server launched.");
-			
+		
 		while (!CancellationToken.IsCancellationRequested) {
 			var result = await CreateBackup();
 			BackupCompleted?.Invoke(this, result);
@@ -69,43 +63,18 @@ sealed class BackupScheduler : CancellableBackgroundTask {
 	}
 
 	private async Task WaitForOnlinePlayers() {
-		bool needsToLogOfflinePlayersMessage = true;
-		
-		process.AddOutputListener(ServerOutputListener, maxLinesToReadFromHistory: 0);
-		try {
-			while (!CancellationToken.IsCancellationRequested) {
-				serverOutputWhileWaitingForOnlinePlayers.Reset();
-				
-				var onlinePlayerCount = await serverStatusProtocol.GetOnlinePlayerCount(serverPort, CancellationToken);
-				if (onlinePlayerCount == null) {
-					Logger.Warning("Could not detect whether any players are online, starting a new backup.");
-					break;
-				}
-
-				if (onlinePlayerCount > 0) {
-					Logger.Information("Players are online, starting a new backup.");
-					break;
-				}
-
-				if (needsToLogOfflinePlayersMessage) {
-					needsToLogOfflinePlayersMessage = false;
-					Logger.Information("No players are online, waiting for someone to join before starting a new backup.");
-				}
-
-				await Task.Delay(TimeSpan.FromSeconds(10), CancellationToken);
-				
-				Logger.Debug("Waiting for server output before checking for online players again...");
-				await serverOutputWhileWaitingForOnlinePlayers.WaitHandle.WaitOneAsync(CancellationToken);
-			}
-		} finally {
-			process.RemoveOutputListener(ServerOutputListener);
+		var task = playerCountTracker.WaitForOnlinePlayers(CancellationToken);
+		if (!task.IsCompleted) {
+			Logger.Information("Waiting for someone to join before starting a new backup.");
 		}
-	}
-
-	private void ServerOutputListener(object? sender, string line) {
-		if (!serverOutputWhileWaitingForOnlinePlayers.IsSet) {
-			serverOutputWhileWaitingForOnlinePlayers.Set();
-			Logger.Debug("Detected server output, signalling to check for online players again.");
+		
+		try {
+			await task;
+			Logger.Information("Players are online, starting a new backup.");
+		} catch (OperationCanceledException) {
+			throw;
+		} catch (Exception) {
+			Logger.Warning("Could not detect whether any players are online, starting a new backup.");
 		}
 	}
 

Agent/Phantom.Agent.Services/Backups/BackupServerCommandDispatcher.cs

@@ -55,7 +55,6 @@ sealed partial class BackupServerCommandDispatcher : IDisposable {
 	}
 
 	public async Task SaveAllChunks() {
-		// TODO Try if not flushing and waiting a few seconds before flushing reduces lag.
 		await process.SendCommand(MinecraftCommand.SaveAll(flush: true), cancellationToken);
 		await savedTheGame.Task.WaitAsync(TimeSpan.FromMinutes(1), cancellationToken);
 	}

Agent/Phantom.Agent.Services/Instances/State/InstancePlayerCountTracker.cs

@@ -0,0 +1,132 @@
+using Phantom.Agent.Minecraft.Instance;
+using Phantom.Agent.Minecraft.Server;
+using Phantom.Utils.Logging;
+using Phantom.Utils.Tasks;
+using Phantom.Utils.Threading;
+
+namespace Phantom.Agent.Services.Instances.State;
+
+sealed class InstancePlayerCountTracker : CancellableBackgroundTask {
+	private readonly InstanceProcess process;
+	private readonly ushort serverPort;
+
+	private readonly TaskCompletionSource firstDetection = AsyncTasks.CreateCompletionSource();
+	private readonly ManualResetEventSlim serverOutputEvent = new ();
+
+	private int? onlinePlayerCount;
+
+	public int? OnlinePlayerCount {
+		get {
+			lock (this) {
+				return onlinePlayerCount;
+			}
+		}
+		private set {
+			EventHandler<int?>? onlinePlayerCountChanged;
+			lock (this) {
+				if (onlinePlayerCount == value) {
+					return;
+				}
+				
+				onlinePlayerCount = value;
+				onlinePlayerCountChanged = OnlinePlayerCountChanged;
+			}
+
+			onlinePlayerCountChanged?.Invoke(this, value);
+		}
+	}
+
+	private event EventHandler<int?>? OnlinePlayerCountChanged;
+
+	private bool isDisposed = false;
+
+	public InstancePlayerCountTracker(InstanceContext context, InstanceProcess process, ushort serverPort) : base(PhantomLogger.Create<InstancePlayerCountTracker>(context.ShortName)) {
+		this.process = process;
+		this.serverPort = serverPort;
+		Start();
+	}
+
+	protected override async Task RunTask() {
+		// Give the server time to start accepting connections.
+		await Task.Delay(TimeSpan.FromSeconds(10), CancellationToken);
+
+		serverOutputEvent.Set();
+		process.AddOutputListener(OnOutput, maxLinesToReadFromHistory: 0);
+		
+		while (!CancellationToken.IsCancellationRequested) {
+			serverOutputEvent.Reset();
+
+			OnlinePlayerCount = await TryGetOnlinePlayerCount();
+			
+			if (!firstDetection.Task.IsCompleted) {
+				firstDetection.SetResult();
+			}
+
+			await Task.Delay(TimeSpan.FromSeconds(10), CancellationToken);
+			await serverOutputEvent.WaitHandle.WaitOneAsync(CancellationToken);
+			await Task.Delay(TimeSpan.FromSeconds(1), CancellationToken);
+		}
+	}
+
+	private async Task<int?> TryGetOnlinePlayerCount() {
+		try {
+			int newOnlinePlayerCount = await ServerStatusProtocol.GetOnlinePlayerCount(serverPort, CancellationToken);
+			Logger.Debug("Detected {OnlinePlayerCount} online player(s).", newOnlinePlayerCount);
+			return newOnlinePlayerCount;
+		} catch (ServerStatusProtocol.ProtocolException e) {
+			Logger.Error(e.Message);
+			return null;
+		} catch (Exception e) {
+			Logger.Error(e, "Caught exception while checking online player count.");
+			return null;
+		}
+	}
+
+	public async Task WaitForOnlinePlayers(CancellationToken cancellationToken) {
+		await firstDetection.Task.WaitAsync(cancellationToken);
+
+		var onlinePlayersDetected = AsyncTasks.CreateCompletionSource();
+
+		lock (this) {
+			if (onlinePlayerCount == null) {
+				throw new InvalidOperationException();
+			}
+			else if (onlinePlayerCount > 0) {
+				return;
+			}
+
+			OnlinePlayerCountChanged += OnOnlinePlayerCountChanged;
+
+			void OnOnlinePlayerCountChanged(object? sender, int? newPlayerCount) {
+				if (newPlayerCount == null) {
+					onlinePlayersDetected.TrySetException(new InvalidOperationException());
+					OnlinePlayerCountChanged -= OnOnlinePlayerCountChanged;
+				}
+				else if (newPlayerCount > 0) {
+					onlinePlayersDetected.TrySetResult();
+					OnlinePlayerCountChanged -= OnOnlinePlayerCountChanged;
+				}
+			}
+		}
+
+		await onlinePlayersDetected.Task;
+	}
+
+	private void OnOutput(object? sender, string? line) {
+		lock (this) {
+			if (!isDisposed) {
+				serverOutputEvent.Set();
+			}
+		}
+	}
+
+	protected override void Dispose() {
+		lock (this) {
+			isDisposed = true;
+			onlinePlayerCount = null;
+		}
+
+		process.RemoveOutputListener(OnOutput);
+		serverOutputEvent.Dispose();
+	}
+}

Agent/Phantom.Agent.Services/Instances/State/InstanceRunningState.cs

@@ -19,6 +19,7 @@ sealed class InstanceRunningState : IDisposable {
 	private readonly CancellationToken cancellationToken;
 
 	private readonly InstanceLogSender logSender;
+	private readonly InstancePlayerCountTracker playerCountTracker;
 	private readonly BackupScheduler backupScheduler;
 
 	private bool isDisposed;
@@ -32,8 +33,9 @@ sealed class InstanceRunningState : IDisposable {
 		this.cancellationToken = cancellationToken;
 
 		this.logSender = new InstanceLogSender(context.Services.ControllerConnection, context.InstanceGuid, context.ShortName);
+		this.playerCountTracker = new InstancePlayerCountTracker(context, process, configuration.ServerPort);
 
-		this.backupScheduler = new BackupScheduler(context, process, configuration.ServerPort);
+		this.backupScheduler = new BackupScheduler(context, playerCountTracker);
 		this.backupScheduler.BackupCompleted += OnScheduledBackupCompleted;
 	}
 
@@ -93,6 +95,11 @@ sealed class InstanceRunningState : IDisposable {
 		}
 	}
 
+	public void OnStopInitiated() {
+		backupScheduler.Stop();
+		playerCountTracker.Stop();
+	}
+	
 	private bool TryDispose() {
 		lock (this) {
 			if (isDisposed) {
@@ -102,8 +109,8 @@ sealed class InstanceRunningState : IDisposable {
 			isDisposed = true;
 		}
 
+		OnStopInitiated();
 		logSender.Stop();
-		backupScheduler.Stop();
 		
 		Process.Dispose();
 		

Agent/Phantom.Agent.Services/Instances/State/InstanceStopProcedure.cs

@@ -25,6 +25,8 @@ static class InstanceStopProcedure {
 
 		try {
 			// Too late to cancel the stop procedure now.
+			runningState.OnStopInitiated();
+			
 			if (!process.HasEnded) {
 				context.Logger.Information("Session stopping now.");
 				await DoStop(context, process);
@@ -85,7 +87,7 @@ static class InstanceStopProcedure {
 	private static async Task WaitForSessionToEnd(InstanceContext context, InstanceProcess process) {
 		try {
 			await process.WaitForExit(TimeSpan.FromSeconds(55));
-		} catch (OperationCanceledException) {
+		} catch (TimeoutException) {
 			try {
 				context.Logger.Warning("Waiting timed out, killing session...");
 				process.Kill();