fix(inbound): in-memory request buffering; relaying standby skips audit capture (review 2026-10-21 05/M4)
This commit is contained in:
@@ -3,6 +3,7 @@ using System.Diagnostics;
|
|||||||
using System.Text;
|
using System.Text;
|
||||||
using System.Text.Json;
|
using System.Text.Json;
|
||||||
using Microsoft.AspNetCore.Http;
|
using Microsoft.AspNetCore.Http;
|
||||||
|
using Microsoft.Extensions.DependencyInjection;
|
||||||
using Microsoft.Extensions.Logging;
|
using Microsoft.Extensions.Logging;
|
||||||
using Microsoft.Extensions.Options;
|
using Microsoft.Extensions.Options;
|
||||||
using ZB.MOM.WW.Audit;
|
using ZB.MOM.WW.Audit;
|
||||||
@@ -221,11 +222,21 @@ public sealed class AuditWriteMiddleware
|
|||||||
&& opts.PerTargetOverrides.TryGetValue(methodNameForOverride, out var perTarget)
|
&& opts.PerTargetOverrides.TryGetValue(methodNameForOverride, out var perTarget)
|
||||||
&& perTarget.SkipBodyCapture;
|
&& perTarget.SkipBodyCapture;
|
||||||
|
|
||||||
|
// A standby that will relay this request to the active peer neither executes nor
|
||||||
|
// audits it (the serving node owns the single row, see `relayed` below). Skip the
|
||||||
|
// request/response capture entirely so the body is read once — by the forwarder —
|
||||||
|
// instead of buffered here, buffered again by the forwarder, and a response copy
|
||||||
|
// captured only to be discarded (review 2026-10-21 05/M4). If every peer refuses
|
||||||
|
// and the standby does emit its own failure row, that row simply carries no bodies.
|
||||||
|
var willRelay = WillRelayToActivePeer(ctx);
|
||||||
|
|
||||||
var requestBody = (string?)null;
|
var requestBody = (string?)null;
|
||||||
var requestTruncated = false;
|
var requestTruncated = false;
|
||||||
if (!skipBody && RequestHasBody(ctx.Request))
|
if (!skipBody && !willRelay && RequestHasBody(ctx.Request))
|
||||||
{
|
{
|
||||||
ctx.Request.EnableBuffering();
|
// Bodies are capped at MaxRequestBodyBytes (1 MiB default): keep them in memory
|
||||||
|
// rather than spilling everything above the 30 KB default threshold to a temp file.
|
||||||
|
ctx.Request.EnableBuffering((int)Math.Min(maxBytes, int.MaxValue));
|
||||||
try
|
try
|
||||||
{
|
{
|
||||||
(requestBody, requestTruncated) = await ReadBufferedRequestBodyAsync(ctx.Request, cap).ConfigureAwait(false);
|
(requestBody, requestTruncated) = await ReadBufferedRequestBodyAsync(ctx.Request, cap).ConfigureAwait(false);
|
||||||
@@ -243,7 +254,8 @@ public sealed class AuditWriteMiddleware
|
|||||||
// original Response.Body is restored in the finally block.
|
// original Response.Body is restored in the finally block.
|
||||||
var originalResponseBody = ctx.Response.Body;
|
var originalResponseBody = ctx.Response.Body;
|
||||||
using var captureStream = new CapturedResponseStream(originalResponseBody, cap);
|
using var captureStream = new CapturedResponseStream(originalResponseBody, cap);
|
||||||
ctx.Response.Body = captureStream;
|
if (!willRelay)
|
||||||
|
ctx.Response.Body = captureStream;
|
||||||
|
|
||||||
Exception? thrown = null;
|
Exception? thrown = null;
|
||||||
try
|
try
|
||||||
@@ -294,6 +306,19 @@ public sealed class AuditWriteMiddleware
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// True when this node is a standby with a configured relay and the request has not
|
||||||
|
/// already been relayed once — i.e. <see cref="InboundApiEndpointFilter"/> will forward it.
|
||||||
|
/// Mirrors the filter's own relay decision.
|
||||||
|
/// </summary>
|
||||||
|
private static bool WillRelayToActivePeer(HttpContext ctx)
|
||||||
|
{
|
||||||
|
var gate = ctx.RequestServices?.GetService<IActiveNodeGate>();
|
||||||
|
if (gate is not { IsActiveNode: false }) return false;
|
||||||
|
if (ctx.Request.Headers.ContainsKey(HttpStandbyRequestForwarder.ForwardedRequestHeader)) return false;
|
||||||
|
return ctx.RequestServices!.GetService<IStandbyRequestForwarder>() is { IsConfigured: true };
|
||||||
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Builds and writes the <see cref="AuditChannel.ApiInbound"/> row for the
|
/// Builds and writes the <see cref="AuditChannel.ApiInbound"/> row for the
|
||||||
/// request. Wrapped in try/catch so a thrown writer or any other emission
|
/// request. Wrapped in try/catch so a thrown writer or any other emission
|
||||||
|
|||||||
@@ -64,7 +64,7 @@ internal static class RelayProvenance
|
|||||||
catch (Exception ex) when (ex is FormatException or JsonException or ArgumentException) { return; }
|
catch (Exception ex) when (ex is FormatException or JsonException or ArgumentException) { return; }
|
||||||
|
|
||||||
// Authenticate small metadata before reading; bound even non-Kestrel request streams.
|
// Authenticate small metadata before reading; bound even non-Kestrel request streams.
|
||||||
context.Request.EnableBuffering();
|
context.Request.EnableBuffering((int)Math.Min(options.MaxRequestBodyBytes, int.MaxValue));
|
||||||
var body = await ReadBoundedAsync(context.Request.Body, options.MaxRequestBodyBytes, context.RequestAborted);
|
var body = await ReadBoundedAsync(context.Request.Body, options.MaxRequestBodyBytes, context.RequestAborted);
|
||||||
context.Request.Body.Position = 0;
|
context.Request.Body.Position = 0;
|
||||||
if (!string.Equals(proof.Binding, Binding(context.Request, body), StringComparison.Ordinal)) return;
|
if (!string.Equals(proof.Binding, Binding(context.Request, body), StringComparison.Ordinal)) return;
|
||||||
|
|||||||
@@ -62,6 +62,36 @@ public class AuditWriteMiddlewareTests
|
|||||||
if (localRows > 0) Assert.NotNull(writer.Events[0].ExecutionId);
|
if (localRows > 0) Assert.NotNull(writer.Events[0].ExecutionId);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// <summary>Verifies that a relaying standby skips request buffering and response capture.</summary>
|
||||||
|
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||||
|
[Fact]
|
||||||
|
public async Task M4_RelayingStandby_DoesNotBufferRequestOrCaptureResponse()
|
||||||
|
{
|
||||||
|
// Review 2026-10-21 05/M4: the standby neither executes nor audits a relayed call, so
|
||||||
|
// it must not buffer the body (the forwarder reads it once) or wrap the response.
|
||||||
|
var services = new ServiceCollection().AddLogging();
|
||||||
|
services.AddSingleton<IActiveNodeGate>(new StandbyGate());
|
||||||
|
services.AddSingleton<IStandbyRequestForwarder>(new ResultForwarder(Results.Ok()));
|
||||||
|
using var provider = services.BuildServiceProvider();
|
||||||
|
var context = BuildContext("echo", "{}");
|
||||||
|
context.RequestServices = provider;
|
||||||
|
var originalRequest = context.Request.Body;
|
||||||
|
var originalResponse = new MemoryStream();
|
||||||
|
context.Response.Body = originalResponse;
|
||||||
|
Stream? seenRequest = null, seenResponse = null;
|
||||||
|
var middleware = CreateMiddleware(ctx =>
|
||||||
|
{
|
||||||
|
seenRequest = ctx.Request.Body;
|
||||||
|
seenResponse = ctx.Response.Body;
|
||||||
|
return Task.CompletedTask;
|
||||||
|
}, new RecordingAuditWriter());
|
||||||
|
|
||||||
|
await middleware.InvokeAsync(context);
|
||||||
|
|
||||||
|
Assert.Same(originalRequest, seenRequest);
|
||||||
|
Assert.Same(originalResponse, seenResponse);
|
||||||
|
}
|
||||||
|
|
||||||
private sealed class StandbyGate : IActiveNodeGate
|
private sealed class StandbyGate : IActiveNodeGate
|
||||||
{
|
{
|
||||||
/// <inheritdoc />
|
/// <inheritdoc />
|
||||||
|
|||||||
Reference in New Issue
Block a user