Implement and test bridge request chunking

- Minor cleanups included

src/DMAPI/tgs/includes.dm

@@ -13,6 +13,7 @@
 
 #include "v5\_defines.dm"
 #include "v5\api.dm"
+#include "v5\bridge.dm"
 #include "v5\commands.dm"
 #include "v5\serializers.dm"
 #include "v5\undefs.dm"

src/DMAPI/tgs/v5/README.md

@@ -5,6 +5,7 @@ This DMAPI implements bridge requests using HTTP GET requests to TGS. It has no
 - [__interop_version.dm](./__interop_version.dm) contains the version of the API used between the DMAPI and TGS.
 - [_defines.dm](./_defines.dm) contains constant definitions.
 - [api.dm](./api.dm) contains the bulk of the API code.
+- [bridge.dm](./bridge.dm) contains the functions related to making bridge requests.
 - [commands.dm](./commands.dm) contains functions relating to `/datum/tgs_chat_command`s.
 - [serializers.dm](./serializers.dm) contains function to help convert interop `/datum`s into a JSON encodable `list()` format.
 - [undefs.dm](./undefs.dm) Undoes the work of `_defines.dm`.

src/DMAPI/tgs/v5/_defines.dm

@@ -12,6 +12,7 @@
 #define DMAPI5_BRIDGE_COMMAND_REBOOT 3
 #define DMAPI5_BRIDGE_COMMAND_KILL 4
 #define DMAPI5_BRIDGE_COMMAND_CHAT_SEND 5
+#define DMAPI5_BRIDGE_COMMAND_CHUNK 6
 
 #define DMAPI5_PARAMETER_ACCESS_IDENTIFIER "accessIdentifier"
 #define DMAPI5_PARAMETER_CUSTOM_COMMANDS "customCommands"
@@ -26,6 +27,7 @@
 
 #define DMAPI5_BRIDGE_RESPONSE_NEW_PORT "newPort"
 #define DMAPI5_BRIDGE_RESPONSE_RUNTIME_INFORMATION "runtimeInformation"
+#define DMAPI5_BRIDGE_RESPONSE_MISSING_CHUNKS "missingChunks"
 
 #define DMAPI5_CHAT_MESSAGE_CHANNEL_IDS "channelIds"
 

src/DMAPI/tgs/v5/api.dm

@@ -16,6 +16,7 @@
 	var/list/chat_channels
 
 	var/initialized = FALSE
+	var/chunked_requests = 0
 
 /datum/tgs_api/v5/ApiVersion()
 	return new /datum/tgs_version(
@@ -244,39 +245,6 @@
 
 	return TopicResponse("Unknown command: [command]")
 
-/datum/tgs_api/v5/proc/Bridge(command, list/data)
-	if(!data)
-		data = list()
-
-	data[DMAPI5_BRIDGE_PARAMETER_COMMAND_TYPE] = command
-	data[DMAPI5_PARAMETER_ACCESS_IDENTIFIER] = access_identifier
-
-	var/json = json_encode(data)
-	var/encoded_json = url_encode(json)
-
-	// This is an infinite sleep until we get a response
-	var/export_response = world.Export("http://127.0.0.1:[server_port]/Bridge?[DMAPI5_BRIDGE_DATA]=[encoded_json]")
-	if(!export_response)
-		TGS_ERROR_LOG("Failed export request: [json]")
-		return
-
-	var/response_json = file2text(export_response["CONTENT"])
-	if(!response_json)
-		TGS_ERROR_LOG("Failed export request, missing content!")
-		return
-
-	var/list/bridge_response = json_decode(response_json)
-	if(!bridge_response)
-		TGS_ERROR_LOG("Failed export request, bad json: [response_json]")
-		return
-
-	var/error = bridge_response[DMAPI5_RESPONSE_ERROR_MESSAGE]
-	if(error)
-		TGS_ERROR_LOG("Failed export request, bad request: [error]")
-		return
-
-	return bridge_response
-
 /datum/tgs_api/v5/OnReboot()
 	var/list/result = Bridge(DMAPI5_BRIDGE_COMMAND_REBOOT)
 	if(!result)

src/DMAPI/tgs/v5/bridge.dm

@@ -0,0 +1,105 @@
+/datum/tgs_api/v5/proc/Bridge(command, list/data)
+	if(!data)
+		data = list()
+
+	var/single_bridge_request = CreateBridgeRequest(command, data)
+	if(length(single_bridge_request) <= DMAPI5_BRIDGE_REQUEST_LIMIT)
+		return PerformBridgeRequest(single_bridge_request)
+
+	// chunking required
+	var/payload_id = ++chunked_requests
+
+	var/raw_data = CreateBridgeData(command, data, FALSE)
+	var/data_length = length(raw_data)
+
+	var/chunk_count
+	var/list/chunk_requests
+	for(chunk_count = 2; !chunk_requests; ++chunk_count);
+		var/max_chunk_size = -round(-(data_length / chunk_count))
+		if(max_chunk_size > DMAPI5_BRIDGE_REQUEST_LIMIT)
+			continue
+
+		chunk_requests = list()
+		for(var/i in 1 to chunk_count)
+			var/startIndex = 1 + ((i - 1) * max_chunk_size)
+			var/endIndex = min(1 + (i * max_chunk_size), data_length + 1)
+			var/chunk_payload = copytext(raw_data, startIndex, endIndex)
+			var/list/chunk = list("payloadId" = payload_id, "sequenceId" = (i - 1), "totalChunks" = chunk_count, payload = chunk_payload)
+
+			var/chunk_request = CreateBridgeRequest(DMAPI5_BRIDGE_COMMAND_CHUNK, list("chunk" = chunk))
+			if(length(chunk_request) > DMAPI5_BRIDGE_REQUEST_LIMIT)
+				// Screwed by url encoding, no way to preempt it though
+				chunk_requests = null
+				break
+
+			chunk_requests += chunk_request
+
+	var/list/response
+	for(var/bridge_request in chunk_requests)
+		response = PerformBridgeRequest(bridge_request)
+		if(!response)
+			// Abort
+			return
+
+	var/list/missing_sequence_ids = response[DMAPI5_BRIDGE_RESPONSE_MISSING_CHUNKS]
+	if(length(missing_sequence_ids))
+		do
+			TGS_WARNING_LOG("Server is missing some chunks of payload [payload_id]! Sending missing chunks...")
+			if(!istype(missing_sequence_ids))
+				TGS_ERROR_LOG("Did not receive a list() for [DMAPI5_BRIDGE_RESPONSE_MISSING_CHUNKS]!")
+				return
+
+			for(var/missing_sequence_id in missing_sequence_ids)
+				if(!isnum(missing_sequence_id))
+					TGS_ERROR_LOG("Did not receive a num in [DMAPI5_BRIDGE_RESPONSE_MISSING_CHUNKS]!")
+					return
+
+				var/missing_chunk_request = chunk_requests[missing_sequence_id + 1]
+				response = PerformBridgeRequest(missing_chunk_request)
+				if(!response)
+					// Abort
+					return
+
+			missing_sequence_ids = response[DMAPI5_BRIDGE_RESPONSE_MISSING_CHUNKS]
+		while(length(missing_sequence_ids))
+
+	return response
+
+/datum/tgs_api/v5/proc/CreateBridgeRequest(command, list/data)
+	var/json = CreateBridgeData(command, data, TRUE)
+	var/encoded_json = url_encode(json)
+
+	var/url = "http://127.0.0.1:[server_port]/Bridge?[DMAPI5_BRIDGE_DATA]=[encoded_json]"
+	return url
+
+/datum/tgs_api/v5/proc/CreateBridgeData(command, list/data, needs_auth)
+	data[DMAPI5_BRIDGE_PARAMETER_COMMAND_TYPE] = command
+	if(needs_auth)
+		data[DMAPI5_PARAMETER_ACCESS_IDENTIFIER] = access_identifier
+
+	var/json = json_encode(data)
+	return json
+
+/datum/tgs_api/v5/proc/PerformBridgeRequest(bridge_request)
+	// This is an infinite sleep until we get a response
+	var/export_response = world.Export(bridge_request)
+	if(!export_response)
+		TGS_ERROR_LOG("Failed bridge request: [bridge_request]")
+		return
+
+	var/response_json = file2text(export_response["CONTENT"])
+	if(!response_json)
+		TGS_ERROR_LOG("Failed bridge request, missing content!")
+		return
+
+	var/list/bridge_response = json_decode(response_json)
+	if(!bridge_response)
+		TGS_ERROR_LOG("Failed bridge request, bad json: [response_json]")
+		return
+
+	var/error = bridge_response[DMAPI5_RESPONSE_ERROR_MESSAGE]
+	if(error)
+		TGS_ERROR_LOG("Failed bridge request, bad request: [error]")
+		return
+
+	return bridge_response

src/Tgstation.Server.Host/Components/Interop/Bridge/BridgeCommandType.cs

@@ -34,5 +34,10 @@
 		/// DreamDaemon requesting a <see cref="ChatMessage"/> be sent.
 		/// </summary>
 		ChatSend,
+
+		/// <summary>
+		/// DreamDaemon attempting to send a longer bridge message.
+		/// </summary>
+		Chunk,
 	}
 }

src/Tgstation.Server.Host/Components/Interop/Bridge/BridgeParameters.cs

@@ -40,5 +40,10 @@ namespace Tgstation.Server.Host.Components.Interop.Bridge
 		/// The <see cref="Interop.ChatMessage"/> for <see cref="BridgeCommandType.ChatSend"/> requests.
 		/// </summary>
 		public ChatMessage ChatMessage { get; set; }
+
+		/// <summary>
+		/// The <see cref="ChunkData"/> for <see cref="BridgeCommandType.Chunk"/> requests.
+		/// </summary>
+		public ChunkData Chunk { get; set; }
 	}
 }

src/Tgstation.Server.Host/Components/Interop/Bridge/BridgeRequestChunker.cs

@@ -0,0 +1,131 @@
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+
+using Microsoft.Extensions.Logging;
+
+using Newtonsoft.Json;
+
+namespace Tgstation.Server.Host.Components.Interop.Bridge
+{
+	/// <summary>
+	/// Processes chunked bridge requests.
+	/// </summary>
+	abstract class BridgeRequestChunker : IBridgeDispatcher
+	{
+		/// <summary>
+		/// The cache of chunked bridge requests.
+		/// </summary>
+		/// <remarks>If the DMAPI is erroring, this can present a memory leak. Worth expiring entries if they aren't completed after some minutes.</remarks>
+		readonly Dictionary<uint, Tuple<ChunkedRequestInfo, string[]>> bridgeChunks;
+
+		/// <summary>
+		/// The <see cref="ILogger"/> for the <see cref="BridgeRequestChunker"/>.
+		/// </summary>
+		protected ILogger<BridgeRequestChunker> Logger { get; }
+
+		/// <summary>
+		/// Initializes a new instance of the <see cref="BridgeRequestChunker"/> class.
+		/// </summary>
+		/// <param name="logger">The value of <see cref="Logger"/>.</param>
+		protected BridgeRequestChunker(ILogger<BridgeRequestChunker> logger)
+		{
+			Logger = logger ?? throw new ArgumentNullException(nameof(logger));
+			bridgeChunks = new Dictionary<uint, Tuple<ChunkedRequestInfo, string[]>>();
+		}
+
+		/// <inheritdoc />
+		public abstract Task<BridgeResponse> ProcessBridgeRequest(BridgeParameters parameters, CancellationToken cancellationToken);
+
+		/// <summary>
+		/// Process a given <paramref name="chunk"/>.
+		/// </summary>
+		/// <param name="chunk">The <see cref="ChunkData"/>.</param>
+		/// <param name="cancellationToken">The <see cref="CancellationToken"/> for the operation.</param>
+		/// <returns>A <see cref="Task{TResult}"/> resulting in the <see cref="BridgeResponse"/> for the chunked request.</returns>
+		protected async Task<BridgeResponse> ProcessBridgeChunk(ChunkData chunk, CancellationToken cancellationToken)
+		{
+			if (chunk == null)
+				return BridgeError("Missing chunk!");
+
+			ChunkedRequestInfo requestInfo;
+			string[] payloads;
+			lock (bridgeChunks)
+			{
+				if (!bridgeChunks.TryGetValue(chunk.PayloadId, out var tuple))
+				{
+					// first time seeing this payload
+					if (chunk.TotalChunks == 0)
+						return BridgeError("Receieved chunked request with 0 totalChunks!");
+
+					tuple = Tuple.Create<ChunkedRequestInfo, string[]>(chunk, new string[chunk.TotalChunks]);
+					bridgeChunks.Add(chunk.PayloadId, tuple);
+				}
+
+				requestInfo = tuple.Item1;
+				payloads = tuple.Item2;
+
+				Logger.LogTrace("Received bridge payload chunk P{payloadId}: {sequenceId}/{totalChunks}", requestInfo.PayloadId, chunk.SequenceId + 1, requestInfo.TotalChunks);
+
+				if (chunk.TotalChunks != requestInfo.TotalChunks)
+				{
+					bridgeChunks.Remove(requestInfo.PayloadId);
+					return BridgeError("Received differing total chunks for same payloadId! Invalidating payloadId!");
+				}
+
+				if (payloads[chunk.SequenceId] != null && payloads[chunk.SequenceId] != chunk.Payload)
+				{
+					bridgeChunks.Remove(requestInfo.PayloadId);
+					return BridgeError("Received differing payload for same sequenceId! Invalidating payloadId!");
+				}
+
+				payloads[chunk.SequenceId] = chunk.Payload;
+				var missingPayloads = new List<uint>();
+				for (uint i = 0; i < payloads.Length; ++i)
+					if (payloads[i] == null)
+						missingPayloads.Add(i);
+
+				if (missingPayloads.Count > 0)
+					return new BridgeResponse
+					{
+						MissingChunks = missingPayloads,
+					};
+
+				Logger.LogTrace("Received all bridge chunks for P{payloadId}, processing request...", requestInfo.PayloadId);
+				bridgeChunks.Remove(requestInfo.PayloadId);
+			}
+
+			BridgeParameters completedRequest;
+			var fullRequestJson = String.Concat(payloads);
+			try
+			{
+				completedRequest = JsonConvert.DeserializeObject<BridgeParameters>(fullRequestJson, DMApiConstants.SerializerSettings);
+			}
+			catch (Exception ex)
+			{
+				Logger.LogWarning(ex, "Bad chunked bridge request for payload {payloadId}!", requestInfo.PayloadId);
+				return BridgeError("Chunked request completed with bad JSON!", false);
+			}
+
+			return await ProcessBridgeRequest(completedRequest, cancellationToken);
+		}
+
+		/// <summary>
+		/// Create and logs an errored <see cref="BridgeResponse"/>.
+		/// </summary>
+		/// <param name="message">The error message.</param>
+		/// <param name="log">If <paramref name="message"/> should be written to the <see cref="Logger"/>.</param>
+		/// <returns>A new errored <see cref="BridgeResponse"/>.</returns>
+		protected BridgeResponse BridgeError(string message, bool log = true)
+		{
+			if (log)
+				Logger.LogWarning("Bridge processing error: {errorMessage}", message);
+
+			return new BridgeResponse
+			{
+				ErrorMessage = message,
+			};
+		}
+	}
+}

src/Tgstation.Server.Host/Components/Interop/Bridge/BridgeResponse.cs

@@ -1,4 +1,6 @@
-namespace Tgstation.Server.Host.Components.Interop.Bridge
+using System.Collections.Generic;
+
+namespace Tgstation.Server.Host.Components.Interop.Bridge
 {
 	/// <summary>
 	/// A response to a bridge request.
@@ -14,5 +16,10 @@
 		/// The <see cref="Bridge.RuntimeInformation"/> for <see cref="BridgeCommandType.Startup"/> requests.
 		/// </summary>
 		public RuntimeInformation RuntimeInformation { get; set; }
+
+		/// <summary>
+		/// The <see cref="ChunkData.SequenceId"/>s missing from a chunked request.
+		/// </summary>
+		public IReadOnlyCollection<uint> MissingChunks { get; set; }
 	}
 }

src/Tgstation.Server.Host/Components/Interop/Bridge/ChunkData.cs

@@ -0,0 +1,18 @@
+namespace Tgstation.Server.Host.Components.Interop.Bridge
+{
+	/// <summary>
+	/// A packet of a split serialized set of <see cref="BridgeParameters"/>.
+	/// </summary>
+	public sealed class ChunkData : ChunkedRequestInfo
+	{
+		/// <summary>
+		/// The sequence ID of the chunk.
+		/// </summary>
+		public uint SequenceId { get; set; }
+
+		/// <summary>
+		/// The partial JSON payload of the chunk.
+		/// </summary>
+		public string Payload { get; set; }
+	}
+}

src/Tgstation.Server.Host/Components/Interop/Bridge/ChunkedRequestInfo.cs

@@ -0,0 +1,18 @@
+namespace Tgstation.Server.Host.Components.Interop.Bridge
+{
+	/// <summary>
+	/// Information about a chunked bridge request.
+	/// </summary>
+	public abstract class ChunkedRequestInfo
+	{
+		/// <summary>
+		/// The ID of the full request to differentiate different chunkings.
+		/// </summary>
+		public uint PayloadId { get; set; }
+
+		/// <summary>
+		/// The total number of chunks in the request.
+		/// </summary>
+		public uint TotalChunks { get; set; }
+	}
+}

src/Tgstation.Server.Host/Components/Interop/DMApiConstants.cs

@@ -57,7 +57,7 @@ namespace Tgstation.Server.Host.Components.Interop
 		/// <summary>
 		/// <see cref="JsonSerializerSettings"/> for use when communicating with the DMAPI.
 		/// </summary>
-		public static readonly JsonSerializerSettings SerializerSettings = new JsonSerializerSettings
+		public static readonly JsonSerializerSettings SerializerSettings = new ()
 		{
 			ContractResolver = new DefaultContractResolver
 			{

src/Tgstation.Server.Host/Components/Session/SessionController.cs

@@ -24,7 +24,7 @@ using Tgstation.Server.Host.System;
 namespace Tgstation.Server.Host.Components.Session
 {
 	/// <inheritdoc />
-	sealed class SessionController : ISessionController, IBridgeHandler, IChannelSink
+	sealed class SessionController : BridgeRequestChunker, ISessionController, IBridgeHandler, IChannelSink
 	{
 		/// <inheritdoc />
 		public DMApiParameters DMApiParameters => ReattachInformation;
@@ -120,11 +120,6 @@ namespace Tgstation.Server.Host.Components.Session
 		/// </summary>
 		readonly IChatManager chat;
 
-		/// <summary>
-		/// The <see cref="ILogger"/> for the <see cref="SessionController"/>.
-		/// </summary>
-		readonly ILogger<SessionController> logger;
-
 		/// <summary>
 		/// <see langword="lock"/> <see cref="object"/> for port updates and <see cref="disposed"/>.
 		/// </summary>
@@ -182,7 +177,7 @@ namespace Tgstation.Server.Host.Components.Session
 		/// <param name="chat">The value of <see cref="chat"/>.</param>
 		/// <param name="chatTrackingContext">The value of <see cref="chatTrackingContext"/>.</param>
 		/// <param name="assemblyInformationProvider">The <see cref="IAssemblyInformationProvider"/> for the <see cref="SessionController"/>.</param>
-		/// <param name="logger">The value of <see cref="logger"/>.</param>
+		/// <param name="logger">The value of <see cref="BridgeRequestChunker.Logger"/>.</param>
 		/// <param name="postLifetimeCallback">The <see cref="Func{TResult}"/> returning a <see cref="Task"/> to be run after the <paramref name="process"/> ends.</param>
 		/// <param name="startupTimeout">The optional time to wait before failing the <see cref="LaunchResult"/>.</param>
 		/// <param name="reattached">If this is a reattached session.</param>
@@ -202,6 +197,7 @@ namespace Tgstation.Server.Host.Components.Session
 			uint? startupTimeout,
 			bool reattached,
 			bool apiValidate)
+			: base(logger)
 		{
 			ReattachInformation = reattachInformation ?? throw new ArgumentNullException(nameof(reattachInformation));
 			this.metadata = metadata ?? throw new ArgumentNullException(nameof(metadata));
@@ -212,7 +208,6 @@ namespace Tgstation.Server.Host.Components.Session
 			if (bridgeRegistrar == null)
 				throw new ArgumentNullException(nameof(bridgeRegistrar));
 			this.chat = chat ?? throw new ArgumentNullException(nameof(chat));
-			this.logger = logger ?? throw new ArgumentNullException(nameof(logger));
 
 			portClosedForReboot = false;
 			disposed = false;
@@ -268,7 +263,7 @@ namespace Tgstation.Server.Host.Components.Session
 				disposed = true;
 			}
 
-			logger.LogTrace("Disposing...");
+			Logger.LogTrace("Disposing...");
 			if (!released)
 			{
 				process.Terminate();
@@ -288,158 +283,23 @@ namespace Tgstation.Server.Host.Components.Session
 		}
 
 		/// <inheritdoc />
-		public Task<BridgeResponse> ProcessBridgeRequest(BridgeParameters parameters, CancellationToken cancellationToken)
+		public override async Task<BridgeResponse> ProcessBridgeRequest(BridgeParameters parameters, CancellationToken cancellationToken)
 		{
 			if (parameters == null)
 				throw new ArgumentNullException(nameof(parameters));
 
-			static Task<BridgeResponse> Error(string message) => Task.FromResult(new BridgeResponse
-			{
-				ErrorMessage = message,
-			});
-
 			using (LogContext.PushProperty("Instance", metadata.Id))
 			{
-				logger.LogTrace("Handling bridge request...");
-				initialBridgeRequestTcs.TrySetResult();
+				Logger.LogTrace("Handling bridge request...");
 
-				var response = new BridgeResponse();
-				switch (parameters.CommandType)
+				try
 				{
-					case BridgeCommandType.ChatSend:
-						if (parameters.ChatMessage == null)
-							return Error("Missing chatMessage field!");
-
-						if (parameters.ChatMessage.ChannelIds == null)
-							return Error("Missing channelIds field in chatMessage!");
-
-						if (parameters.ChatMessage.ChannelIds.Any(channelIdString => !UInt64.TryParse(channelIdString, out var _)))
-							return Error("Invalid channelIds in chatMessage!");
-
-						if (parameters.ChatMessage.Text == null)
-							return Error("Missing message field in chatMessage!");
-
-						var anyFailed = false;
-						var parsedChannels = parameters.ChatMessage.ChannelIds.Select(
-							channelString =>
-							{
-								anyFailed |= !UInt64.TryParse(channelString, out var channelId);
-								return channelId;
-							});
-
-						if (anyFailed)
-							return Error("Failed to parse channelIds as U64!");
-
-						chat.QueueMessage(
-							parameters.ChatMessage,
-							parsedChannels);
-						break;
-					case BridgeCommandType.Prime:
-						var oldPrimeTcs = primeTcs;
-						primeTcs = new TaskCompletionSource();
-						oldPrimeTcs.SetResult();
-						break;
-					case BridgeCommandType.Kill:
-						logger.LogInformation("Bridge requested process termination!");
-						TerminationWasRequested = true;
-						process.Terminate();
-						break;
-					case BridgeCommandType.PortUpdate:
-						lock (synchronizationLock)
-						{
-							if (!parameters.CurrentPort.HasValue)
-							{
-								/////UHHHH
-								logger.LogWarning("DreamDaemon sent new port command without providing it's own!");
-								return Error("Missing stringified port as data parameter!");
-							}
-
-							var currentPort = parameters.CurrentPort.Value;
-							if (!nextPort.HasValue)
-								ReattachInformation.Port = parameters.CurrentPort.Value; // not ready yet, so what we'll do is accept the random port DD opened on for now and change it later when we decide to
-							else
-							{
-								// nextPort is ready, tell DD to switch to that
-								// if it fails it'll kill itself
-								response.NewPort = nextPort.Value;
-								ReattachInformation.Port = nextPort.Value;
-								nextPort = null;
-
-								// we'll also get here from SetPort so complete that task
-								var tmpTcs = portAssignmentTcs;
-								portAssignmentTcs = null;
-								tmpTcs.SetResult(true);
-							}
-
-							portClosedForReboot = false;
-						}
-
-						break;
-					case BridgeCommandType.Startup:
-						apiValidationStatus = ApiValidationStatus.BadValidationRequest;
-						if (parameters.Version == null)
-							return Error("Missing dmApiVersion field!");
-
-						DMApiVersion = parameters.Version;
-						if (DMApiVersion.Major != DMApiConstants.InteropVersion.Major)
-						{
-							apiValidationStatus = ApiValidationStatus.Incompatible;
-							return Error("Incompatible dmApiVersion!");
-						}
-
-						switch (parameters.MinimumSecurityLevel)
-						{
-							case DreamDaemonSecurity.Ultrasafe:
-								apiValidationStatus = ApiValidationStatus.RequiresUltrasafe;
-								break;
-							case DreamDaemonSecurity.Safe:
-								apiValidationStatus = ApiValidationStatus.RequiresSafe;
-								break;
-							case DreamDaemonSecurity.Trusted:
-								apiValidationStatus = ApiValidationStatus.RequiresTrusted;
-								break;
-							case null:
-								return Error("Missing minimumSecurityLevel field!");
-							default:
-								return Error("Invalid minimumSecurityLevel!");
-						}
-
-						logger.LogTrace("ApiValidationStatus set to {0}", apiValidationStatus);
-
-						response.RuntimeInformation = new RuntimeInformation(
-							chatTrackingContext,
-							ReattachInformation.Dmb,
-							ReattachInformation.RuntimeInformation.ServerVersion,
-							ReattachInformation.RuntimeInformation.InstanceName,
-							ReattachInformation.RuntimeInformation.SecurityLevel,
-							ReattachInformation.RuntimeInformation.Visibility,
-							ReattachInformation.RuntimeInformation.ServerPort,
-							ReattachInformation.RuntimeInformation.ApiValidateOnly);
-
-						// Load custom commands
-						chatTrackingContext.CustomCommands = parameters.CustomCommands;
-						break;
-					case BridgeCommandType.Reboot:
-						if (ClosePortOnReboot)
-						{
-							chatTrackingContext.Active = false;
-							response.NewPort = 0;
-							portClosedForReboot = true;
-						}
-
-						var oldRebootTcs = rebootTcs;
-						rebootTcs = new TaskCompletionSource();
-						oldRebootTcs.SetResult();
-						break;
-					case null:
-						response.ErrorMessage = "Missing commandType!";
-						break;
-					default:
-						response.ErrorMessage = "Requested commandType not supported!";
-						break;
+					return await ProcessBridgeCommand(parameters, cancellationToken);
+				}
+				finally
+				{
+					initialBridgeRequestTcs.TrySetResult();
 				}
-
-				return Task.FromResult(response);
 			}
 		}
 
@@ -466,7 +326,7 @@ namespace Tgstation.Server.Host.Components.Session
 
 			if (Lifetime.IsCompleted)
 			{
-				logger.LogWarning(
+				Logger.LogWarning(
 					"Attempted to send a command to an inactive SessionController: {0}",
 					parameters.CommandType);
 				return null;
@@ -474,14 +334,14 @@ namespace Tgstation.Server.Host.Components.Session
 
 			if (!DMApiAvailable)
 			{
-				logger.LogTrace("Not sending topic request {0} to server without/with incompatible DMAPI!", parameters.CommandType);
+				Logger.LogTrace("Not sending topic request {0} to server without/with incompatible DMAPI!", parameters.CommandType);
 				return null;
 			}
 
 			parameters.AccessIdentifier = ReattachInformation.AccessIdentifier;
 
 			var json = JsonConvert.SerializeObject(parameters, DMApiConstants.SerializerSettings);
-			logger.LogTrace("Topic request: {0}", json);
+			Logger.LogTrace("Topic request: {0}", json);
 			try
 			{
 				var commandString = String.Format(
@@ -506,21 +366,21 @@ namespace Tgstation.Server.Host.Components.Session
 						interopResponse = JsonConvert.DeserializeObject<Interop.Topic.TopicResponse>(topicReturn, DMApiConstants.SerializerSettings);
 						if (interopResponse.ErrorMessage != null)
 						{
-							logger.LogWarning("Errored topic response for command {0}: {1}", parameters.CommandType, interopResponse.ErrorMessage);
+							Logger.LogWarning("Errored topic response for command {0}: {1}", parameters.CommandType, interopResponse.ErrorMessage);
 						}
 
-						logger.LogTrace("Interop response: {0}", topicReturn);
+						Logger.LogTrace("Interop response: {0}", topicReturn);
 					}
 					catch (Exception ex)
 					{
-						logger.LogWarning(ex, "Invalid interop response: {0}", topicReturn);
+						Logger.LogWarning(ex, "Invalid interop response: {0}", topicReturn);
 					}
 
 				return new CombinedTopicResponse(topicResponse, interopResponse);
 			}
 			catch (OperationCanceledException ex)
 			{
-				logger.LogTrace(
+				Logger.LogTrace(
 					ex,
 					"Topic request {0}!",
 					cancellationToken.IsCancellationRequested
@@ -530,7 +390,7 @@ namespace Tgstation.Server.Host.Components.Session
 			}
 			catch (Exception e)
 			{
-				logger.LogWarning(e, "Send command exception!");
+				Logger.LogWarning(e, "Send command exception!");
 			}
 
 			return null;
@@ -577,7 +437,7 @@ namespace Tgstation.Server.Host.Components.Session
 			if (RebootState == newRebootState)
 				return true;
 
-			logger.LogTrace("Changing reboot state to {0}", newRebootState);
+			Logger.LogTrace("Changing reboot state to {0}", newRebootState);
 
 			ReattachInformation.RebootState = newRebootState;
 			var result = await SendCommand(
@@ -592,7 +452,7 @@ namespace Tgstation.Server.Host.Components.Session
 		public void ResetRebootState()
 		{
 			CheckDisposed();
-			logger.LogTrace("Resetting reboot state...");
+			Logger.LogTrace("Resetting reboot state...");
 			ReattachInformation.RebootState = RebootState.Normal;
 		}
 
@@ -654,7 +514,7 @@ namespace Tgstation.Server.Host.Components.Session
 			if (startupTimeout.HasValue)
 				toAwait = Task.WhenAny(toAwait, Task.Delay(TimeSpan.FromSeconds(startupTimeout.Value)));
 
-			logger.LogTrace(
+			Logger.LogTrace(
 				"Waiting for LaunchResult based on {0}{1}...",
 				useBridgeRequestForLaunchResult ? "initial bridge request" : "process startup",
 				startupTimeout.HasValue ? $" with a timeout of {startupTimeout.Value}s" : String.Empty);
@@ -667,7 +527,7 @@ namespace Tgstation.Server.Host.Components.Session
 				StartupTime = startupTask.IsCompleted ? (DateTimeOffset.UtcNow - startTime) : null,
 			};
 
-			logger.LogTrace("Launch result: {0}", result);
+			Logger.LogTrace("Launch result: {0}", result);
 
 			if (!result.ExitCode.HasValue && reattached && !disposed)
 			{
@@ -683,7 +543,7 @@ namespace Tgstation.Server.Host.Components.Session
 					if (reattachResponse.InteropResponse?.CustomCommands != null)
 						chatTrackingContext.CustomCommands = reattachResponse.InteropResponse.CustomCommands;
 					else if (reattachResponse.InteropResponse != null)
-						logger.LogWarning(
+						Logger.LogWarning(
 							"DMAPI v{0} isn't returning the TGS custom commands list. Functionality added in v5.2.0.",
 							CompileJob.DMApiVersion.Semver());
 				}
@@ -700,5 +560,152 @@ namespace Tgstation.Server.Host.Components.Session
 			if (disposed)
 				throw new ObjectDisposedException(nameof(SessionController));
 		}
+
+		/// <summary>
+		/// Handle a set of bridge <paramref name="parameters"/>.
+		/// </summary>
+		/// <param name="parameters">The <see cref="BridgeParameters"/> to handle.</param>
+		/// <param name="cancellationToken">The <see cref="CancellationToken"/> for the operation.</param>
+		/// <returns>A <see cref="Task{TResult}"/> resulting in the <see cref="BridgeResponse"/> for the request or <see langword="null"/> if the request could not be dispatched.</returns>
+		async Task<BridgeResponse> ProcessBridgeCommand(BridgeParameters parameters, CancellationToken cancellationToken)
+		{
+			var response = new BridgeResponse();
+			switch (parameters.CommandType)
+			{
+				case BridgeCommandType.ChatSend:
+					if (parameters.ChatMessage == null)
+						return BridgeError("Missing chatMessage field!");
+
+					if (parameters.ChatMessage.ChannelIds == null)
+						return BridgeError("Missing channelIds field in chatMessage!");
+
+					if (parameters.ChatMessage.ChannelIds.Any(channelIdString => !UInt64.TryParse(channelIdString, out var _)))
+						return BridgeError("Invalid channelIds in chatMessage!");
+
+					if (parameters.ChatMessage.Text == null)
+						return BridgeError("Missing message field in chatMessage!");
+
+					var anyFailed = false;
+					var parsedChannels = parameters.ChatMessage.ChannelIds.Select(
+						channelString =>
+						{
+							anyFailed |= !UInt64.TryParse(channelString, out var channelId);
+							return channelId;
+						});
+
+					if (anyFailed)
+						return BridgeError("Failed to parse channelIds as U64!");
+
+					chat.QueueMessage(
+						parameters.ChatMessage,
+						parsedChannels);
+					break;
+				case BridgeCommandType.Prime:
+					var oldPrimeTcs = primeTcs;
+					primeTcs = new TaskCompletionSource();
+					oldPrimeTcs.SetResult();
+					break;
+				case BridgeCommandType.Kill:
+					Logger.LogInformation("Bridge requested process termination!");
+					TerminationWasRequested = true;
+					process.Terminate();
+					break;
+				case BridgeCommandType.PortUpdate:
+					lock (synchronizationLock)
+					{
+						if (!parameters.CurrentPort.HasValue)
+						{
+							/////UHHHH
+							Logger.LogWarning("DreamDaemon sent new port command without providing it's own!");
+							return BridgeError("Missing stringified port as data parameter!");
+						}
+
+						var currentPort = parameters.CurrentPort.Value;
+						if (!nextPort.HasValue)
+							ReattachInformation.Port = parameters.CurrentPort.Value; // not ready yet, so what we'll do is accept the random port DD opened on for now and change it later when we decide to
+						else
+						{
+							// nextPort is ready, tell DD to switch to that
+							// if it fails it'll kill itself
+							response.NewPort = nextPort.Value;
+							ReattachInformation.Port = nextPort.Value;
+							nextPort = null;
+
+							// we'll also get here from SetPort so complete that task
+							var tmpTcs = portAssignmentTcs;
+							portAssignmentTcs = null;
+							tmpTcs.SetResult(true);
+						}
+
+						portClosedForReboot = false;
+					}
+
+					break;
+				case BridgeCommandType.Startup:
+					apiValidationStatus = ApiValidationStatus.BadValidationRequest;
+					if (parameters.Version == null)
+						return BridgeError("Missing dmApiVersion field!");
+
+					DMApiVersion = parameters.Version;
+					if (DMApiVersion.Major != DMApiConstants.InteropVersion.Major)
+					{
+						apiValidationStatus = ApiValidationStatus.Incompatible;
+						return BridgeError("Incompatible dmApiVersion!");
+					}
+
+					switch (parameters.MinimumSecurityLevel)
+					{
+						case DreamDaemonSecurity.Ultrasafe:
+							apiValidationStatus = ApiValidationStatus.RequiresUltrasafe;
+							break;
+						case DreamDaemonSecurity.Safe:
+							apiValidationStatus = ApiValidationStatus.RequiresSafe;
+							break;
+						case DreamDaemonSecurity.Trusted:
+							apiValidationStatus = ApiValidationStatus.RequiresTrusted;
+							break;
+						case null:
+							return BridgeError("Missing minimumSecurityLevel field!");
+						default:
+							return BridgeError("Invalid minimumSecurityLevel!");
+					}
+
+					Logger.LogTrace("ApiValidationStatus set to {0}", apiValidationStatus);
+
+					response.RuntimeInformation = new RuntimeInformation(
+						chatTrackingContext,
+						ReattachInformation.Dmb,
+						ReattachInformation.RuntimeInformation.ServerVersion,
+						ReattachInformation.RuntimeInformation.InstanceName,
+						ReattachInformation.RuntimeInformation.SecurityLevel,
+						ReattachInformation.RuntimeInformation.Visibility,
+						ReattachInformation.RuntimeInformation.ServerPort,
+						ReattachInformation.RuntimeInformation.ApiValidateOnly);
+
+					// Load custom commands
+					chatTrackingContext.CustomCommands = parameters.CustomCommands;
+					break;
+				case BridgeCommandType.Reboot:
+					if (ClosePortOnReboot)
+					{
+						chatTrackingContext.Active = false;
+						response.NewPort = 0;
+						portClosedForReboot = true;
+					}
+
+					var oldRebootTcs = rebootTcs;
+					rebootTcs = new TaskCompletionSource();
+					oldRebootTcs.SetResult();
+					break;
+				case BridgeCommandType.Chunk:
+					return await ProcessBridgeChunk(parameters.Chunk, cancellationToken);
+				case null:
+					return BridgeError("Missing commandType!");
+				default:
+					return BridgeError($"commandType {parameters.CommandType} not supported!");
+			}
+
+			return response;
+		}
 	}
 }

tests/DMAPI/LongRunning/Test.dm

@@ -173,6 +173,12 @@ var/lastTgsError
 	set waitfor = FALSE
 	CheckBridgeLimitsImpl()
 
+
+/proc/BridgeWithoutChunking(command, list/data)
+	var/datum/tgs_api/v5/api = TGS_READ_GLOBAL(tgs)
+	var/bridge_request = api.CreateBridgeRequest(command, data)
+	return api.PerformBridgeRequest(bridge_request)
+
 /proc/CheckBridgeLimitsImpl()
 	sleep(30)
 
@@ -192,7 +198,7 @@ var/lastTgsError
 	for(i = 1; ; i = base + (2 ** nextPow))
 		var/payload = create_payload(i)
 		
-		var/list/result = api.Bridge(0, list("chatMessage" = list("text" = "payload:[payload]")))
+		var/list/result = BridgeWithoutChunking(0, list("chatMessage" = list("text" = "payload:[payload]")))
 
 		if(!result || lastTgsError || result["integrationHack"] != "ok")
 			if(i == lastI + 1)
@@ -206,7 +212,7 @@ var/lastTgsError
 		lastI = i
 		++nextPow
 
-	var/finalResult = api.Bridge(0, list("chatMessage" = list("text" = "done")))
+	var/finalResult = api.Bridge(0, list("chatMessage" = list("text" = "done:[create_payload(DMAPI5_BRIDGE_REQUEST_LIMIT * 3)]")))
 	if(!finalResult || lastTgsError || finalResult["integrationHack"] != "ok")
 		text2file("Failed to end bridge limit test!", "test_fail_reason.txt")
 		del(world)

tests/Tgstation.Server.Tests/Instance/TestBridgeHandler.cs

@@ -0,0 +1,85 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using System.Web;
+
+using Microsoft.Extensions.Logging;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+
+using Moq;
+
+using Newtonsoft.Json;
+
+using Tgstation.Server.Host.Components.Interop;
+using Tgstation.Server.Host.Components.Interop.Bridge;
+
+namespace Tgstation.Server.Tests.Instance
+{
+	sealed class TestBridgeHandler : BridgeRequestChunker, IBridgeHandler
+	{
+		class DMApiParametersImpl : DMApiParameters { }
+
+		class BridgeResponseHack : BridgeResponse
+		{
+			public string IntegrationHack { get; set; }
+		}
+
+		public DMApiParameters DMApiParameters => new DMApiParametersImpl
+		{
+			AccessIdentifier = "tgs_integration_test"
+		};
+
+		long lastBridgeRequestSize = 0;
+
+		readonly TaskCompletionSource bridgeTestsTcs;
+		readonly ushort serverPort;
+
+		public TestBridgeHandler(TaskCompletionSource tcs, ILogger<TestBridgeHandler> logger, ushort serverPort)
+			: base(logger)
+		{
+			bridgeTestsTcs = tcs;
+			this.serverPort = serverPort;
+		}
+
+		public override async Task<BridgeResponse> ProcessBridgeRequest(BridgeParameters parameters, CancellationToken cancellationToken)
+		{
+			try
+			{
+				Assert.AreEqual(DMApiParameters.AccessIdentifier, parameters.AccessIdentifier);
+				if (parameters.CommandType == BridgeCommandType.Chunk)
+					return await ProcessBridgeChunk(parameters.Chunk, cancellationToken);
+
+				Assert.AreEqual((BridgeCommandType)0, parameters.CommandType);
+				Assert.IsNotNull(parameters.ChatMessage?.Text);
+				var splits = parameters.ChatMessage.Text.Split(':', StringSplitOptions.RemoveEmptyEntries);
+				Assert.AreEqual(2, splits.Length);
+				var coreMessage = splits[0];
+				Assert.IsFalse(String.IsNullOrWhiteSpace(coreMessage));
+				if (coreMessage == "done")
+				{
+					Assert.AreEqual(DMApiConstants.MaximumBridgeRequestLength, lastBridgeRequestSize);
+					Assert.AreEqual(new string('a', (int)(DMApiConstants.MaximumBridgeRequestLength * 3)), splits[1]);
+
+					bridgeTestsTcs.SetResult();
+					return new BridgeResponseHack
+					{
+						IntegrationHack = "ok"
+					};
+				}
+
+				Assert.AreEqual("payload", coreMessage);
+				lastBridgeRequestSize = $"http://127.0.0.1:{serverPort}/Bridge?data=".Length + HttpUtility.UrlEncode(	
+					JsonConvert.SerializeObject(parameters, DMApiConstants.SerializerSettings)).Length;
+				return new BridgeResponseHack
+				{
+					IntegrationHack = "ok"
+				};
+			}
+			catch (Exception ex)
+			{
+				bridgeTestsTcs.SetException(ex);
+				return null;
+			}
+		}
+	}
+}

tests/Tgstation.Server.Tests/Instance/WatchdogTest.cs

@@ -34,7 +34,7 @@ using Tgstation.Server.Host.System;
 
 namespace Tgstation.Server.Tests.Instance
 {
-	sealed class WatchdogTest : JobsRequiredTest, IBridgeHandler
+	sealed class WatchdogTest : JobsRequiredTest
 	{
 		readonly IInstanceClient instanceClient;
 		readonly InstanceManager instanceManager;
@@ -336,66 +336,7 @@ namespace Tgstation.Server.Tests.Instance
 			return await instanceClient.DreamDaemon.Start(cancellationToken);
 		}
 
-		class DMApiParametersImpl : DMApiParameters { }
-		public DMApiParameters DMApiParameters => new DMApiParametersImpl
-		{
-			AccessIdentifier = "tgs_integration_test"
-		};
-
-		TaskCompletionSource bridgeTestsTcs;
-
-		public class BridgeResponseHack : BridgeResponse
-		{
-			public string IntegrationHack { get; set; }
-		}
-
-		public class ResponseTestData : TestData
-		{
-			public bool Continue { get; set; }
-			public string PayloadId { get; set; }
-		}
-
-		long lastBridgeRequestSize = 0;
-
-		public Task<BridgeResponse> ProcessBridgeRequest(BridgeParameters parameters, CancellationToken cancellationToken)
-		{
-			try
-			{
-				Assert.AreEqual(DMApiParameters.AccessIdentifier, parameters.AccessIdentifier);
-				Assert.AreEqual((BridgeCommandType)0, parameters.CommandType);
-				Assert.IsNotNull(parameters.ChatMessage?.Text);
-				var splits = parameters.ChatMessage.Text.Split(':', StringSplitOptions.RemoveEmptyEntries);
-				var coreMessage = splits[0];
-				Assert.IsFalse(String.IsNullOrWhiteSpace(coreMessage));
-				if (coreMessage == "done")
-				{
-					Assert.AreEqual(DMApiConstants.MaximumBridgeRequestLength, lastBridgeRequestSize);
-
-					bridgeTestsTcs.SetResult();
-					return Task.FromResult<BridgeResponse>(
-						new BridgeResponseHack
-						{
-							IntegrationHack = "ok"
-						});
-				}
-
-				Assert.AreEqual("payload", coreMessage);
-				lastBridgeRequestSize = $"http://127.0.0.1:{serverPort}/Bridge?data=".Length + HttpUtility.UrlEncode(
-					JsonConvert.SerializeObject(parameters, DMApiConstants.SerializerSettings)).Length;
-				return Task.FromResult<BridgeResponse>(
-					new BridgeResponseHack
-					{
-						IntegrationHack = "ok"
-					});
-			}
-			catch (Exception ex)
-			{
-				bridgeTestsTcs.SetException(ex);
-				return Task.FromResult<BridgeResponse>(null);
-			}
-		}
-
-		public class TestData
+		class TestData
 		{
 			public string Size { get; set; }
 			public string Payload { get; set; }
@@ -408,17 +349,23 @@ namespace Tgstation.Server.Tests.Instance
 			await WaitForJob(startJob, 40, false, null, cancellationToken);
 
 			// first check the bridge limits
-			bridgeTestsTcs = new TaskCompletionSource();
+			var bridgeTestsTcs = new TaskCompletionSource();
 			BridgeController.LogContent = false;
-			using (var bridgeRegistration = instanceManager.RegisterHandler(this))
+			using (var loggerFactory = LoggerFactory.Create(builder =>
 			{
-				System.Console.WriteLine("TEST: Sending Bridge tests topic...");
-				var bridgeTestTopicResult = await topicClient.SendTopic(IPAddress.Loopback, "tgs_integration_test_tactics2=1", IntegrationTest.DDPort, cancellationToken);
-				Assert.AreEqual("ack2", bridgeTestTopicResult.StringData);
+				builder.AddConsole();
+				builder.SetMinimumLevel(LogLevel.Trace);
+			}))
+			{
+				var bridgeProcessor = new TestBridgeHandler(bridgeTestsTcs, loggerFactory.CreateLogger<TestBridgeHandler>(), serverPort);
+				using (var bridgeRegistration = instanceManager.RegisterHandler(bridgeProcessor))
+				{
+					System.Console.WriteLine("TEST: Sending Bridge tests topic...");
+					var bridgeTestTopicResult = await topicClient.SendTopic(IPAddress.Loopback, "tgs_integration_test_tactics2=1", IntegrationTest.DDPort, cancellationToken);
+					Assert.AreEqual("ack2", bridgeTestTopicResult.StringData);
 
-				using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
-				// this test doesn't take long
-				await bridgeTestsTcs.Task.WithToken(cancellationToken);
+					await bridgeTestsTcs.Task.WithToken(cancellationToken);
+				}
 			}
 
 			BridgeController.LogContent = true;
@@ -776,7 +723,7 @@ namespace Tgstation.Server.Tests.Instance
 			return ddProc != null;
 		}
 
-		TopicClient topicClient = new (new SocketParameters
+		readonly TopicClient topicClient = new (new SocketParameters
 		{
 			SendTimeout = TimeSpan.FromSeconds(30),
 			ReceiveTimeout = TimeSpan.FromSeconds(30),

tgstation-server.sln

@@ -138,6 +138,7 @@ EndProject
 Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "v5", "v5", "{FAEAD3B5-2EAB-465C-A9C3-E8CB6AAA7131}"
 	ProjectSection(SolutionItems) = preProject
 		src\DMAPI\tgs\v5\api.dm = src\DMAPI\tgs\v5\api.dm
+		src\DMAPI\tgs\v5\bridge.dm = src\DMAPI\tgs\v5\bridge.dm
 		src\DMAPI\tgs\v5\commands.dm = src\DMAPI\tgs\v5\commands.dm
 		src\DMAPI\tgs\v5\README.md = src\DMAPI\tgs\v5\README.md
 		src\DMAPI\tgs\v5\serializers.dm = src\DMAPI\tgs\v5\serializers.dm