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)

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.
 

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
     /// </summary>
     private const long MaxManagementRequestBodyBytes = 200L * 1024 * 1024;
 
-    private static async Task<IResult> HandleRequest(HttpContext context)
+    /// <summary>
+    /// Bodies above this size (or of unknown size) must hold <see cref="LargeRequestGate"/>.
+    /// Ordinary management commands are a few KB; only Transport bundle import/preview
+    /// carries a multi-megabyte base64 payload.
+    /// </summary>
+    internal const long LargeRequestThresholdBytes = 16L * 1024 * 1024;
+
+    /// <summary>
+    /// Single-flight gate for LARGE management requests at the HTTP edge (review 2026-10-21
+    /// 05/M2). The actor's <c>TransportGate</c> 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 <c>Wait(0)</c> BEFORE the body is read and held until the actor
+    /// replies; a second large request is refused with <c>409 IMPORT_IN_PROGRESS</c> instead
+    /// of queueing another copy. Per node (static), like <c>TransportGate</c>.
+    /// </summary>
+    internal static readonly SemaphoreSlim LargeRequestGate = new(1, 1);
+
+    internal static async Task<IResult> HandleRequest(HttpContext context)
     {
         var logger = context.RequestServices.GetRequiredService<ILogger<ManagementActorHolder>>();
 
@@ -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<IResult> 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
 
     /// <summary>
     /// Parses a management request body — a JSON object with a <c>command</c> name and an
-    /// optional <c>payload</c> — into the strongly-typed command record. The parsed
-    /// <see cref="JsonDocument"/> is disposed deterministically and the missing-payload
-    /// case does not allocate a throwaway document.
+    /// optional <c>payload</c> — into the strongly-typed command record. Shares
+    /// <see cref="ParseCommandAsync"/>'s single-pass reader, so string and stream callers get
+    /// identical results.
     /// </summary>
     /// <param name="body">The raw JSON request body string.</param>
     /// <returns>A <see cref="CommandParseResult"/> with the deserialized command on success, or an error on failure.</returns>
     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<CommandRequestBody>(body, EnvelopeJsonOptions));
+        }
+        catch (Exception ex)
+        {
+            return Failure(ex);
+        }
+    }
 
-        if (!doc.RootElement.TryGetProperty("command", out var commandNameElement))
+    /// <summary>
+    /// Stream variant of <see cref="ParseCommand(string)"/> 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 <c>command</c> has been seen (the CLI sends <c>command</c> first). A payload that
+    /// precedes <c>command</c> is buffered as a <see cref="JsonElement"/> and deserialised after.
+    /// </summary>
+    /// <param name="body">The request body stream.</param>
+    /// <param name="cancellationToken">Cancels the read.</param>
+    /// <returns>A <see cref="CommandParseResult"/> with the deserialized command on success, or an error on failure.</returns>
+    public static async Task<CommandParseResult> ParseCommandAsync(Stream body, CancellationToken cancellationToken = default)
+    {
+        try
+        {
+            return Complete(await JsonSerializer.DeserializeAsync<CommandRequestBody>(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.");
+
+    /// <summary>The top-level request shape as read by <see cref="CommandRequestBodyConverter"/>.</summary>
+    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; }
+    }
+
+    /// <summary>Distinguishes a bad payload from malformed JSON in the parse error.</summary>
+    private sealed class PayloadDeserializationException(Exception inner)
+        : Exception(inner.Message, inner);
+
+    /// <summary>
+    /// Reads <c>{"command": "...", "payload": {...}}</c> in one pass. Property names are
+    /// matched case-sensitively, as the previous <c>JsonDocument.TryGetProperty</c> parse did;
+    /// unknown top-level properties are skipped.
+    /// </summary>
+    private sealed class CommandRequestBodyConverter : System.Text.Json.Serialization.JsonConverter<CommandRequestBody>
+    {
+        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();
     }
 }

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;
+
+/// <summary>
+/// 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 <c>TransportGate</c> 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.
+/// </summary>
+public class ManagementEndpointsLargeRequestTests
+{
+    private static DefaultHttpContext Context(Stream body, long? contentLength)
+    {
+        var services = new ServiceCollection();
+        services.AddLogging();
+        services.AddSingleton<IOptions<AuthDisableLoginOptions>>(
+            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;
+
+    /// <summary>A second large request is refused with 409 without reading its body.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [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<IValueHttpResult>(result).Value!.ToString()!;
+            Assert.Contains("IMPORT_IN_PROGRESS", json);
+            Assert.False(body.WasRead);
+        }
+        finally
+        {
+            ManagementEndpoints.LargeRequestGate.Release();
+        }
+    }
+
+    /// <summary>A request of unknown length (chunked) counts as large.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [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();
+        }
+    }
+
+    /// <summary>Ordinary small commands are never blocked by an in-flight import.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [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();
+        }
+    }
+
+    /// <summary>The gate is released once a large request completes.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [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();
+    }
+
+    /// <summary>The stream parse handles a payload that precedes the command name.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [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<GetSiteCommand>(result.Command).SiteId);
+    }
+
+    /// <summary>The stream parse reports invalid JSON and bad payloads with the existing messages.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [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<int> ReadAsync(Memory<byte> 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();
+    }
+}