From b0c863c6a11734d0c20411d8615691fc4fe99f86 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 09:22:22 -0400 Subject: [PATCH] fix(management): single-flight large /management bodies at the HTTP edge (409 IMPORT_IN_PROGRESS) and parse the command off the request stream (review 2026-10-21 05/M2) --- .../Component-ManagementService.md | 3 + .../ManagementEndpoints.cs | 217 +++++++++++++++--- .../ManagementEndpointsLargeRequestTests.cs | 153 ++++++++++++ 3 files changed, 335 insertions(+), 38 deletions(-) create mode 100644 tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/ManagementEndpointsLargeRequestTests.cs diff --git a/docs/requirements/Component-ManagementService.md b/docs/requirements/Component-ManagementService.md index 61bf7699..67fd6c80 100644 --- a/docs/requirements/Component-ManagementService.md +++ b/docs/requirements/Component-ManagementService.md @@ -67,6 +67,9 @@ Content-Type: application/json - `ManagementUnauthorized` → HTTP 403 with `{ "error": "...", "code": "UNAUTHORIZED" }` - Authentication failure → HTTP 401 - Actor timeout → HTTP 504 +- Large request while another is in flight → HTTP 409 with `{ "error": "...", "code": "IMPORT_IN_PROGRESS" }` (see below) + +**Large-request single-flight and streaming parse (review 2026-10-21 05/M2).** The body cap is raised to 200 MB for bundle import/preview. A request whose `Content-Length` exceeds 16 MB — or is unknown (chunked) — must take a per-node `LargeRequestGate` with `Wait(0)` *after authentication and before the body is read*, and holds it until the actor replies; a second large request is refused with **409 `IMPORT_IN_PROGRESS`** (retry when the first completes) rather than materialising another copy and then waiting at the actor's `TransportGate`. Ordinary (small) commands never touch the gate. The command is parsed straight off the request stream (`ParseCommandAsync`: one UTF-8 buffer, the payload deserialised directly into the command type once `command` has been read — the CLI sends `command` first; a payload that precedes it is buffered as a `JsonElement`), replacing the old `ReadToEnd` string → `JsonDocument` → `GetRawText` → `Deserialize` chain (~4 copies, ~1.1 GB peak for a 200 MB body). Error messages (`Invalid JSON body.`, `Missing 'command' field.`, `Unknown command: '…'.`, `Failed to deserialize payload: …`) are unchanged. The endpoint performs LDAP authentication and role resolution server-side, collapsing the CLI's previous two-step flow (ResolveRoles + actual command) into a single HTTP round-trip. diff --git a/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementEndpoints.cs b/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementEndpoints.cs index f96e0398..fe8f3860 100644 --- a/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementEndpoints.cs +++ b/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementEndpoints.cs @@ -1,4 +1,3 @@ -using System.Text; using System.Text.Json; using Akka.Actor; using Microsoft.AspNetCore.Builder; @@ -94,7 +93,25 @@ public static class ManagementEndpoints /// private const long MaxManagementRequestBodyBytes = 200L * 1024 * 1024; - private static async Task HandleRequest(HttpContext context) + /// + /// Bodies above this size (or of unknown size) must hold . + /// Ordinary management commands are a few KB; only Transport bundle import/preview + /// carries a multi-megabyte base64 payload. + /// + internal const long LargeRequestThresholdBytes = 16L * 1024 * 1024; + + /// + /// Single-flight gate for LARGE management requests at the HTTP edge (review 2026-10-21 + /// 05/M2). The actor's TransportGate serialises the bundle handlers, but it is + /// reached only after the endpoint has already parsed the body — so concurrent imports + /// each materialised their payload here and then pinned it while queued at the actor + /// gate. Acquired with Wait(0) BEFORE the body is read and held until the actor + /// replies; a second large request is refused with 409 IMPORT_IN_PROGRESS instead + /// of queueing another copy. Per node (static), like TransportGate. + /// + internal static readonly SemaphoreSlim LargeRequestGate = new(1, 1); + + internal static async Task HandleRequest(HttpContext context) { var logger = context.RequestServices.GetRequiredService>(); @@ -118,14 +135,36 @@ public static class ManagementEndpoints var authenticatedUser = auth.User!; - // 4. Parse command from request body - string body; - using (var reader = new StreamReader(context.Request.Body, Encoding.UTF8)) + // 4. Large bodies (bundle import/preview) are single-flight per node, decided + // BEFORE the body is read so a refused request never materialises its payload + // (review 2026-10-21 05/M2). Unknown length (chunked) counts as large. + var isLarge = context.Request.ContentLength is not { } length || length > LargeRequestThresholdBytes; + if (isLarge && !LargeRequestGate.Wait(0)) { - body = await reader.ReadToEndAsync(); + return Results.Json(new + { + error = "Another large management request (a bundle import or preview) is already in progress on this node. Retry when it completes.", + code = "IMPORT_IN_PROGRESS", + }, statusCode: 409); } - var parse = ParseCommand(body); + try + { + return await ParseAndDispatchAsync(context, authenticatedUser, logger); + } + finally + { + if (isLarge) + LargeRequestGate.Release(); + } + } + + private static async Task ParseAndDispatchAsync( + HttpContext context, AuthenticatedUser authenticatedUser, ILogger logger) + { + // Parse the command straight off the request stream: one UTF-8 pass into the typed + // command, no ReadToEnd string / JsonDocument / GetRawText copies (review 2026-10-21 05/M2). + var parse = await ParseCommandAsync(context.Request.Body, context.RequestAborted); if (!parse.Success) { return Results.Json(new { error = parse.ErrorMessage, code = parse.ErrorCode }, statusCode: 400); @@ -193,39 +232,72 @@ public static class ManagementEndpoints /// /// Parses a management request body — a JSON object with a command name and an - /// optional payload — into the strongly-typed command record. The parsed - /// is disposed deterministically and the missing-payload - /// case does not allocate a throwaway document. + /// optional payload — into the strongly-typed command record. Shares + /// 's single-pass reader, so string and stream callers get + /// identical results. /// /// The raw JSON request body string. /// A with the deserialized command on success, or an error on failure. public static CommandParseResult ParseCommand(string body) { - using JsonDocument doc = ParseDocument(body, out var parseError); - if (parseError != null) - return CommandParseResult.Fail(parseError); + try + { + return Complete(JsonSerializer.Deserialize(body, EnvelopeJsonOptions)); + } + catch (Exception ex) + { + return Failure(ex); + } + } - if (!doc.RootElement.TryGetProperty("command", out var commandNameElement)) + /// + /// Stream variant of used by the HTTP endpoint: the body + /// is read once as UTF-8 and the payload is deserialised directly into the command type as + /// soon as command has been seen (the CLI sends command first). A payload that + /// precedes command is buffered as a and deserialised after. + /// + /// The request body stream. + /// Cancels the read. + /// A with the deserialized command on success, or an error on failure. + public static async Task ParseCommandAsync(Stream body, CancellationToken cancellationToken = default) + { + try + { + return Complete(await JsonSerializer.DeserializeAsync(body, EnvelopeJsonOptions, cancellationToken)); + } + catch (OperationCanceledException) + { + throw; + } + catch (Exception ex) + { + return Failure(ex); + } + } + + private static readonly JsonSerializerOptions EnvelopeJsonOptions = new() + { + Converters = { new CommandRequestBodyConverter() }, + }; + + private static CommandParseResult Complete(CommandRequestBody? parsed) + { + if (parsed is null || !parsed.HasCommand) return CommandParseResult.Fail("Missing 'command' field."); - - var commandName = commandNameElement.GetString(); - if (string.IsNullOrWhiteSpace(commandName)) + if (string.IsNullOrWhiteSpace(parsed.CommandName)) return CommandParseResult.Fail("Empty 'command' field."); - - var commandType = ManagementCommandRegistry.Resolve(commandName); - if (commandType == null) - return CommandParseResult.Fail($"Unknown command: '{commandName}'."); + if (parsed.CommandType is null) + return CommandParseResult.Fail($"Unknown command: '{parsed.CommandName}'."); try { - // Missing payload: deserialize from the empty-object literal rather than - // allocating (and leaking) a throwaway JsonDocument. - var payloadJson = doc.RootElement.TryGetProperty("payload", out var p) - ? p.GetRawText() - : "{}"; - var command = JsonSerializer.Deserialize(payloadJson, commandType, - CommandPayloadJsonOptions)!; - return CommandParseResult.Ok(command); + // Payload was either deserialised in-stream, buffered because it preceded the + // command name, or absent (deserialise the empty object). + var command = parsed.Command + ?? (parsed.DeferredPayload is { } deferred + ? deferred.Deserialize(parsed.CommandType, CommandPayloadJsonOptions) + : JsonSerializer.Deserialize("{}", parsed.CommandType, CommandPayloadJsonOptions)); + return CommandParseResult.Ok(command!); } catch (Exception ex) { @@ -233,17 +305,86 @@ public static class ManagementEndpoints } } - private static JsonDocument ParseDocument(string body, out string? error) + private static CommandParseResult Failure(Exception ex) => ex is PayloadDeserializationException payload + ? CommandParseResult.Fail($"Failed to deserialize payload: {payload.InnerException?.Message ?? payload.Message}") + : CommandParseResult.Fail("Invalid JSON body."); + + /// The top-level request shape as read by . + private sealed class CommandRequestBody { - try + public bool HasCommand { get; set; } + public string? CommandName { get; set; } + public Type? CommandType { get; set; } + public object? Command { get; set; } + public JsonElement? DeferredPayload { get; set; } + } + + /// Distinguishes a bad payload from malformed JSON in the parse error. + private sealed class PayloadDeserializationException(Exception inner) + : Exception(inner.Message, inner); + + /// + /// Reads {"command": "...", "payload": {...}} in one pass. Property names are + /// matched case-sensitively, as the previous JsonDocument.TryGetProperty parse did; + /// unknown top-level properties are skipped. + /// + private sealed class CommandRequestBodyConverter : System.Text.Json.Serialization.JsonConverter + { + public override CommandRequestBody Read(ref Utf8JsonReader reader, Type typeToConvert, JsonSerializerOptions options) { - error = null; - return JsonDocument.Parse(body); - } - catch - { - error = "Invalid JSON body."; - return JsonDocument.Parse("{}"); + if (reader.TokenType != JsonTokenType.StartObject) + throw new JsonException("Expected a JSON object."); + + var result = new CommandRequestBody(); + while (reader.Read()) + { + if (reader.TokenType == JsonTokenType.EndObject) + return result; + if (reader.TokenType != JsonTokenType.PropertyName) + throw new JsonException("Expected a property name."); + + var isCommand = reader.ValueTextEquals("command"); + var isPayload = !isCommand && reader.ValueTextEquals("payload"); + reader.Read(); + + if (isCommand) + { + result.HasCommand = true; + result.CommandName = reader.TokenType == JsonTokenType.String + ? reader.GetString() + : reader.TokenType == JsonTokenType.Null ? null + : throw new JsonException("'command' must be a string."); + result.CommandType = string.IsNullOrWhiteSpace(result.CommandName) + ? null + : ManagementCommandRegistry.Resolve(result.CommandName); + } + else if (isPayload && result.CommandType is not null) + { + try + { + result.Command = JsonSerializer.Deserialize(ref reader, result.CommandType, CommandPayloadJsonOptions); + } + catch (Exception ex) + { + throw new PayloadDeserializationException(ex); + } + + result.DeferredPayload = null; + } + else if (isPayload) + { + result.DeferredPayload = JsonElement.ParseValue(ref reader); + } + else + { + reader.Skip(); + } + } + + throw new JsonException("Unterminated JSON object."); } + + public override void Write(Utf8JsonWriter writer, CommandRequestBody value, JsonSerializerOptions options) + => throw new NotSupportedException(); } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/ManagementEndpointsLargeRequestTests.cs b/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/ManagementEndpointsLargeRequestTests.cs new file mode 100644 index 00000000..aab4ce24 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/ManagementEndpointsLargeRequestTests.cs @@ -0,0 +1,153 @@ +using System.Text; +using Microsoft.AspNetCore.Http; +using Microsoft.AspNetCore.Http.HttpResults; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Options; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Management; +using ZB.MOM.WW.ScadaBridge.Security.Auth; + +namespace ZB.MOM.WW.ScadaBridge.ManagementService.Tests; + +/// +/// Review 2026-10-21 05/M2: bundle import/preview bodies used to be materialised as ~4 +/// copies at the HTTP edge before the actor's TransportGate was reached. Large +/// requests are now single-flight at the edge (decided before the body is read), and the +/// command is parsed straight off the request stream. +/// +public class ManagementEndpointsLargeRequestTests +{ + private static DefaultHttpContext Context(Stream body, long? contentLength) + { + var services = new ServiceCollection(); + services.AddLogging(); + services.AddSingleton>( + Options.Create(new AuthDisableLoginOptions { DisableLogin = true })); + services.AddSingleton(new ManagementActorHolder()); // no actor → 503 after a successful parse + var context = new DefaultHttpContext { RequestServices = services.BuildServiceProvider() }; + context.Request.Method = "POST"; + context.Request.Body = body; + context.Request.ContentLength = contentLength; + return context; + } + + private static MemoryStream Json(string json) => new(Encoding.UTF8.GetBytes(json)); + + private static int? StatusOf(IResult result) => (result as IStatusCodeHttpResult)?.StatusCode; + + /// A second large request is refused with 409 without reading its body. + /// A task that represents the asynchronous operation. + [Fact] + public async Task LargeRequest_WhileAnotherIsInFlight_Returns409_WithoutReadingTheBody() + { + Assert.True(ManagementEndpoints.LargeRequestGate.Wait(0)); + try + { + var body = new UnreadableStream(); + var result = await ManagementEndpoints.HandleRequest( + Context(body, ManagementEndpoints.LargeRequestThresholdBytes + 1)); + + Assert.Equal(409, StatusOf(result)); + var json = Assert.IsAssignableFrom(result).Value!.ToString()!; + Assert.Contains("IMPORT_IN_PROGRESS", json); + Assert.False(body.WasRead); + } + finally + { + ManagementEndpoints.LargeRequestGate.Release(); + } + } + + /// A request of unknown length (chunked) counts as large. + /// A task that represents the asynchronous operation. + [Fact] + public async Task UnknownLengthRequest_WhileAnotherIsInFlight_Returns409() + { + Assert.True(ManagementEndpoints.LargeRequestGate.Wait(0)); + try + { + var result = await ManagementEndpoints.HandleRequest(Context(new UnreadableStream(), null)); + Assert.Equal(409, StatusOf(result)); + } + finally + { + ManagementEndpoints.LargeRequestGate.Release(); + } + } + + /// Ordinary small commands are never blocked by an in-flight import. + /// A task that represents the asynchronous operation. + [Fact] + public async Task SmallRequest_WhileLargeIsInFlight_IsNotGated() + { + Assert.True(ManagementEndpoints.LargeRequestGate.Wait(0)); + try + { + var body = Json("""{"command":"ListSites","payload":{}}"""); + var result = await ManagementEndpoints.HandleRequest(Context(body, body.Length)); + Assert.Equal(503, StatusOf(result)); // parsed and reached dispatch + } + finally + { + ManagementEndpoints.LargeRequestGate.Release(); + } + } + + /// The gate is released once a large request completes. + /// A task that represents the asynchronous operation. + [Fact] + public async Task LargeRequest_ReleasesTheGate_WhenDone() + { + var body = Json("""{"command":"ListSites"}"""); + var result = await ManagementEndpoints.HandleRequest( + Context(body, ManagementEndpoints.LargeRequestThresholdBytes + 1)); + + Assert.Equal(503, StatusOf(result)); + Assert.True(ManagementEndpoints.LargeRequestGate.Wait(0)); + ManagementEndpoints.LargeRequestGate.Release(); + } + + /// The stream parse handles a payload that precedes the command name. + /// A task that represents the asynchronous operation. + [Fact] + public async Task ParseCommandAsync_PayloadBeforeCommand_StillDeserializes() + { + var result = await ManagementEndpoints.ParseCommandAsync( + Json("""{"payload":{"siteId":7},"command":"GetSite"}""")); + + Assert.True(result.Success, result.ErrorMessage); + Assert.Equal(7, Assert.IsType(result.Command).SiteId); + } + + /// The stream parse reports invalid JSON and bad payloads with the existing messages. + /// A task that represents the asynchronous operation. + [Fact] + public async Task ParseCommandAsync_Errors_MatchTheStringParse() + { + var invalid = await ManagementEndpoints.ParseCommandAsync(Json("{ not json")); + Assert.Equal("Invalid JSON body.", invalid.ErrorMessage); + + var badPayload = await ManagementEndpoints.ParseCommandAsync( + Json("""{"command":"GetSite","payload":{"siteId":"not-a-number"}}""")); + Assert.False(badPayload.Success); + Assert.StartsWith("Failed to deserialize payload:", badPayload.ErrorMessage); + + var empty = await ManagementEndpoints.ParseCommandAsync(Json("""{"command":" "}""")); + Assert.Equal("Empty 'command' field.", empty.ErrorMessage); + } + + private sealed class UnreadableStream : Stream + { + public bool WasRead { get; private set; } + public override bool CanRead => true; + public override bool CanSeek => false; + public override bool CanWrite => false; + public override long Length => throw new NotSupportedException(); + public override long Position { get => 0; set => throw new NotSupportedException(); } + public override void Flush() { } + public override int Read(byte[] buffer, int offset, int count) { WasRead = true; throw new InvalidOperationException("body read"); } + public override ValueTask ReadAsync(Memory buffer, CancellationToken cancellationToken = default) { WasRead = true; throw new InvalidOperationException("body read"); } + public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); + public override void SetLength(long value) => throw new NotSupportedException(); + public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException(); + } +}