From 7cee137e43977c489b5791d5b8546bbb05101bb2 Mon Sep 17 00:00:00 2001 From: Dominion Date: Sat, 15 Apr 2023 19:14:57 -0400 Subject: [PATCH] Implement and test bridge request chunking - Minor cleanups included --- src/DMAPI/tgs/includes.dm | 1 + src/DMAPI/tgs/v5/README.md | 1 + src/DMAPI/tgs/v5/_defines.dm | 2 + src/DMAPI/tgs/v5/api.dm | 34 +- src/DMAPI/tgs/v5/bridge.dm | 105 ++++++ .../Interop/Bridge/BridgeCommandType.cs | 5 + .../Interop/Bridge/BridgeParameters.cs | 5 + .../Interop/Bridge/BridgeRequestChunker.cs | 131 +++++++ .../Interop/Bridge/BridgeResponse.cs | 9 +- .../Components/Interop/Bridge/ChunkData.cs | 18 + .../Interop/Bridge/ChunkedRequestInfo.cs | 18 + .../Components/Interop/DMApiConstants.cs | 2 +- .../Components/Session/SessionController.cs | 337 +++++++++--------- tests/DMAPI/LongRunning/Test.dm | 10 +- .../Instance/TestBridgeHandler.cs | 85 +++++ .../Instance/WatchdogTest.cs | 87 +---- tgstation-server.sln | 1 + 17 files changed, 579 insertions(+), 272 deletions(-) create mode 100644 src/DMAPI/tgs/v5/bridge.dm create mode 100644 src/Tgstation.Server.Host/Components/Interop/Bridge/BridgeRequestChunker.cs create mode 100644 src/Tgstation.Server.Host/Components/Interop/Bridge/ChunkData.cs create mode 100644 src/Tgstation.Server.Host/Components/Interop/Bridge/ChunkedRequestInfo.cs create mode 100644 tests/Tgstation.Server.Tests/Instance/TestBridgeHandler.cs diff --git a/src/DMAPI/tgs/includes.dm b/src/DMAPI/tgs/includes.dm index 25e1b8421a..bf91aa6006 100644 --- a/src/DMAPI/tgs/includes.dm +++ b/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" diff --git a/src/DMAPI/tgs/v5/README.md b/src/DMAPI/tgs/v5/README.md index 619b58cd72..7c65ecf776 100644 --- a/src/DMAPI/tgs/v5/README.md +++ b/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`. diff --git a/src/DMAPI/tgs/v5/_defines.dm b/src/DMAPI/tgs/v5/_defines.dm index daf1da050a..580c40bca0 100644 --- a/src/DMAPI/tgs/v5/_defines.dm +++ b/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" diff --git a/src/DMAPI/tgs/v5/api.dm b/src/DMAPI/tgs/v5/api.dm index 4d08ec194d..15d5bb03e2 100644 --- a/src/DMAPI/tgs/v5/api.dm +++ b/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) diff --git a/src/DMAPI/tgs/v5/bridge.dm b/src/DMAPI/tgs/v5/bridge.dm new file mode 100644 index 0000000000..d2e7bbc395 --- /dev/null +++ b/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 diff --git a/src/Tgstation.Server.Host/Components/Interop/Bridge/BridgeCommandType.cs b/src/Tgstation.Server.Host/Components/Interop/Bridge/BridgeCommandType.cs index 2de0e824b8..ad3c227356 100644 --- a/src/Tgstation.Server.Host/Components/Interop/Bridge/BridgeCommandType.cs +++ b/src/Tgstation.Server.Host/Components/Interop/Bridge/BridgeCommandType.cs @@ -34,5 +34,10 @@ /// DreamDaemon requesting a be sent. /// ChatSend, + + /// + /// DreamDaemon attempting to send a longer bridge message. + /// + Chunk, } } diff --git a/src/Tgstation.Server.Host/Components/Interop/Bridge/BridgeParameters.cs b/src/Tgstation.Server.Host/Components/Interop/Bridge/BridgeParameters.cs index f174ffec1f..6bd13cb72d 100644 --- a/src/Tgstation.Server.Host/Components/Interop/Bridge/BridgeParameters.cs +++ b/src/Tgstation.Server.Host/Components/Interop/Bridge/BridgeParameters.cs @@ -40,5 +40,10 @@ namespace Tgstation.Server.Host.Components.Interop.Bridge /// The for requests. /// public ChatMessage ChatMessage { get; set; } + + /// + /// The for requests. + /// + public ChunkData Chunk { get; set; } } } diff --git a/src/Tgstation.Server.Host/Components/Interop/Bridge/BridgeRequestChunker.cs b/src/Tgstation.Server.Host/Components/Interop/Bridge/BridgeRequestChunker.cs new file mode 100644 index 0000000000..9ce6d7b443 --- /dev/null +++ b/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 +{ + /// + /// Processes chunked bridge requests. + /// + abstract class BridgeRequestChunker : IBridgeDispatcher + { + /// + /// The cache of chunked bridge requests. + /// + /// If the DMAPI is erroring, this can present a memory leak. Worth expiring entries if they aren't completed after some minutes. + readonly Dictionary> bridgeChunks; + + /// + /// The for the . + /// + protected ILogger Logger { get; } + + /// + /// Initializes a new instance of the class. + /// + /// The value of . + protected BridgeRequestChunker(ILogger logger) + { + Logger = logger ?? throw new ArgumentNullException(nameof(logger)); + bridgeChunks = new Dictionary>(); + } + + /// + public abstract Task ProcessBridgeRequest(BridgeParameters parameters, CancellationToken cancellationToken); + + /// + /// Process a given . + /// + /// The . + /// The for the operation. + /// A resulting in the for the chunked request. + protected async Task 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(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(); + 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(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); + } + + /// + /// Create and logs an errored . + /// + /// The error message. + /// If should be written to the . + /// A new errored . + protected BridgeResponse BridgeError(string message, bool log = true) + { + if (log) + Logger.LogWarning("Bridge processing error: {errorMessage}", message); + + return new BridgeResponse + { + ErrorMessage = message, + }; + } + } +} diff --git a/src/Tgstation.Server.Host/Components/Interop/Bridge/BridgeResponse.cs b/src/Tgstation.Server.Host/Components/Interop/Bridge/BridgeResponse.cs index 4ce5adab21..68c1adbea7 100644 --- a/src/Tgstation.Server.Host/Components/Interop/Bridge/BridgeResponse.cs +++ b/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 { /// /// A response to a bridge request. @@ -14,5 +16,10 @@ /// The for requests. /// public RuntimeInformation RuntimeInformation { get; set; } + + /// + /// The s missing from a chunked request. + /// + public IReadOnlyCollection MissingChunks { get; set; } } } diff --git a/src/Tgstation.Server.Host/Components/Interop/Bridge/ChunkData.cs b/src/Tgstation.Server.Host/Components/Interop/Bridge/ChunkData.cs new file mode 100644 index 0000000000..e0c0204d2c --- /dev/null +++ b/src/Tgstation.Server.Host/Components/Interop/Bridge/ChunkData.cs @@ -0,0 +1,18 @@ +namespace Tgstation.Server.Host.Components.Interop.Bridge +{ + /// + /// A packet of a split serialized set of . + /// + public sealed class ChunkData : ChunkedRequestInfo + { + /// + /// The sequence ID of the chunk. + /// + public uint SequenceId { get; set; } + + /// + /// The partial JSON payload of the chunk. + /// + public string Payload { get; set; } + } +} diff --git a/src/Tgstation.Server.Host/Components/Interop/Bridge/ChunkedRequestInfo.cs b/src/Tgstation.Server.Host/Components/Interop/Bridge/ChunkedRequestInfo.cs new file mode 100644 index 0000000000..1246f37a19 --- /dev/null +++ b/src/Tgstation.Server.Host/Components/Interop/Bridge/ChunkedRequestInfo.cs @@ -0,0 +1,18 @@ +namespace Tgstation.Server.Host.Components.Interop.Bridge +{ + /// + /// Information about a chunked bridge request. + /// + public abstract class ChunkedRequestInfo + { + /// + /// The ID of the full request to differentiate different chunkings. + /// + public uint PayloadId { get; set; } + + /// + /// The total number of chunks in the request. + /// + public uint TotalChunks { get; set; } + } +} diff --git a/src/Tgstation.Server.Host/Components/Interop/DMApiConstants.cs b/src/Tgstation.Server.Host/Components/Interop/DMApiConstants.cs index f5c9ffc21b..3a2c6fe9fb 100644 --- a/src/Tgstation.Server.Host/Components/Interop/DMApiConstants.cs +++ b/src/Tgstation.Server.Host/Components/Interop/DMApiConstants.cs @@ -57,7 +57,7 @@ namespace Tgstation.Server.Host.Components.Interop /// /// for use when communicating with the DMAPI. /// - public static readonly JsonSerializerSettings SerializerSettings = new JsonSerializerSettings + public static readonly JsonSerializerSettings SerializerSettings = new () { ContractResolver = new DefaultContractResolver { diff --git a/src/Tgstation.Server.Host/Components/Session/SessionController.cs b/src/Tgstation.Server.Host/Components/Session/SessionController.cs index 8e88a1c03a..723c8626ac 100644 --- a/src/Tgstation.Server.Host/Components/Session/SessionController.cs +++ b/src/Tgstation.Server.Host/Components/Session/SessionController.cs @@ -24,7 +24,7 @@ using Tgstation.Server.Host.System; namespace Tgstation.Server.Host.Components.Session { /// - sealed class SessionController : ISessionController, IBridgeHandler, IChannelSink + sealed class SessionController : BridgeRequestChunker, ISessionController, IBridgeHandler, IChannelSink { /// public DMApiParameters DMApiParameters => ReattachInformation; @@ -120,11 +120,6 @@ namespace Tgstation.Server.Host.Components.Session /// readonly IChatManager chat; - /// - /// The for the . - /// - readonly ILogger logger; - /// /// for port updates and . /// @@ -182,7 +177,7 @@ namespace Tgstation.Server.Host.Components.Session /// The value of . /// The value of . /// The for the . - /// The value of . + /// The value of . /// The returning a to be run after the ends. /// The optional time to wait before failing the . /// If this is a reattached session. @@ -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 } /// - public Task ProcessBridgeRequest(BridgeParameters parameters, CancellationToken cancellationToken) + public override async Task ProcessBridgeRequest(BridgeParameters parameters, CancellationToken cancellationToken) { if (parameters == null) throw new ArgumentNullException(nameof(parameters)); - static Task 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(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)); } + + /// + /// Handle a set of bridge . + /// + /// The to handle. + /// The for the operation. + /// A resulting in the for the request or if the request could not be dispatched. + async Task 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; + } } } diff --git a/tests/DMAPI/LongRunning/Test.dm b/tests/DMAPI/LongRunning/Test.dm index bbd06bfd27..5425908258 100644 --- a/tests/DMAPI/LongRunning/Test.dm +++ b/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) diff --git a/tests/Tgstation.Server.Tests/Instance/TestBridgeHandler.cs b/tests/Tgstation.Server.Tests/Instance/TestBridgeHandler.cs new file mode 100644 index 0000000000..b2c07e33cf --- /dev/null +++ b/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 logger, ushort serverPort) + : base(logger) + { + bridgeTestsTcs = tcs; + this.serverPort = serverPort; + } + + public override async Task 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; + } + } + } +} diff --git a/tests/Tgstation.Server.Tests/Instance/WatchdogTest.cs b/tests/Tgstation.Server.Tests/Instance/WatchdogTest.cs index 1d4a15cb7e..6df7614dce 100644 --- a/tests/Tgstation.Server.Tests/Instance/WatchdogTest.cs +++ b/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 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( - 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( - new BridgeResponseHack - { - IntegrationHack = "ok" - }); - } - catch (Exception ex) - { - bridgeTestsTcs.SetException(ex); - return Task.FromResult(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(), 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), diff --git a/tgstation-server.sln b/tgstation-server.sln index 7057a6eab0..8d53fc25c0 100644 --- a/tgstation-server.sln +++ b/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