From e4a951d8b953c37b9078a48bce43cac6fd09edac Mon Sep 17 00:00:00 2001 From: tanaka takashi Date: Mon, 10 Aug 2026 00:07:27 +0900 Subject: fix(http): decouple request processing from log I/O HTTP observability callbacks previously called RollingFileLog synchronously. RollingFileLog writes to Console.Error before the application file, so a paused console or an undrained redirected stderr pipe could block Kestrel request threads for minutes and cause otherwise valid MCP clients to time out. Route ready, connection, request-started, request-completed, and HTTP JSONL work through a bounded 4,096-entry Channel. Request callbacks use TryWrite and never wait for console or file I/O. A dedicated single-reader background task preserves log ordering, performs rotation, reports dropped entries after queue pressure, and contains non-fatal sink failures. Keep the queue bounded so a stalled console cannot cause unbounded memory growth. When capacity is exhausted, request availability takes precedence over observability and the dropped count is emitted as an application warning after the writer resumes. Add a regression test with a deliberately blocked console writer. The background writer remains blocked while the HTTP completion callback returns within 250 ms. Release build passed with zero warnings and errors; Core tests passed 48/48 with skip 0. Regenerate the win-x64 self-contained 1.0.0 Bundle and ZIP. ZIP SHA-256: 62E28C903880A51614EC0C3CE23409580F803C7C124764CA7E1CB0EB2CCAA7A9. --- README.md | 2 + doc/Design.md | 3 +- src/Agent.App/Logging/McpHostLogObserver.cs | 74 +++++++++++++++++++++++++--- src/Agent.App/Logging/RollingFileLog.cs | 6 ++- tests/Agent.Core.Tests/ObservabilityTests.cs | 52 +++++++++++++++++++ 5 files changed, 126 insertions(+), 11 deletions(-) diff --git a/README.md b/README.md index e66f2c9..d83b5bb 100644 --- a/README.md +++ b/README.md @@ -133,6 +133,8 @@ log directoryは起動ユーザーだけがアクセスできるprotected DACL HTTP logは`X-Forwarded-For`を接続元として信用せず、Authorization、Cookie、query、body、prompt、画像、Base64を保存しません。stateless HTTPでは永続接続を主張せず、request started/completedとして記録します。 +HTTP request処理はconsole/file出力と同期しません。運用logはbounded queue経由でbackground出力し、consoleが一時停止してもHTTP requestを停止させません。queue飽和時はrequest継続を優先し、drop件数をapplication logへ警告します。 + 正常停止後、`agent-mcp.exe` descendant、追加ACL、WFP filter、loopback exemption、`state/backups`、`state/journals` が残っていないことを確認してください。loopback exemptionは `CheckNetIsolation.exe LoopbackExempt -s` で確認できます。 ## よくあるエラー diff --git a/doc/Design.md b/doc/Design.md index 1ca8cbe..1bcbc99 100644 --- a/doc/Design.md +++ b/doc/Design.md @@ -218,7 +218,8 @@ result: - Appはtransport構築前に`WindowsPlatformCheck`を実行する。administrator/BFE/WFP probeが1つでも失敗した場合はhostをbuild/startせずfatalとする。 - `StartAsync`完了後にだけ`IMcpHostObserver.Ready`を通知するため、「起動中」と「接続可能」を区別できる。 -- HTTP middlewareはsocket endpointとrouting metadataだけを`logs/http.jsonl`へJSON Linesで保存する。forwarded header、credential header、query、payloadは保存しない。 +- HTTP middlewareはsocket endpointとrouting metadataだけをbounded in-memory queueへ`TryWrite`し、専用background writerがstderr、application log、`logs/http.jsonl`へ出力する。request threadはconsole/file I/Oを待たず、queue飽和時はrequestを止めずlog entryをdropして件数を警告する。 +- forwarded header、credential header、query、payloadは保存しない。 - `RollingFileWriter`をapplication/http logで共用し、同じsize/retention policyでrotationする。stdioの運用出力はstderrだけへ送る。 --- ## 6. Ollama統合設計 diff --git a/src/Agent.App/Logging/McpHostLogObserver.cs b/src/Agent.App/Logging/McpHostLogObserver.cs index b1410e8..37f32ec 100644 --- a/src/Agent.App/Logging/McpHostLogObserver.cs +++ b/src/Agent.App/Logging/McpHostLogObserver.cs @@ -3,16 +3,22 @@ using System.Text.Json; using Agent.Core.Localization; using Agent.Mcp; using System.Collections.Concurrent; +using System.Threading.Channels; namespace Agent.App.Logging; public sealed class McpHostLogObserver : IMcpHostObserver, IDisposable { + private const int QueueCapacity = 4096; private readonly RollingFileLog _application; private readonly RollingFileWriter? _http; private readonly LocaleCatalog _locale; private readonly string _logDirectory; private readonly ConcurrentDictionary _observedConnections = new(StringComparer.Ordinal); + private readonly Channel _queue; + private readonly Task _writerTask; + private long _droppedEntries; + private int _disposed; public McpHostLogObserver(RollingFileLog application, LocaleCatalog locale, string logDirectory, bool httpRequests, long maxBytes, int retainedFiles) @@ -21,6 +27,14 @@ public sealed class McpHostLogObserver : IMcpHostObserver, IDisposable _locale = locale; _logDirectory = logDirectory; if (httpRequests) _http = new RollingFileWriter(Path.Combine(logDirectory, "http.jsonl"), maxBytes, retainedFiles); + _queue = Channel.CreateBounded(new BoundedChannelOptions(QueueCapacity) + { + SingleReader = true, + SingleWriter = false, + FullMode = BoundedChannelFullMode.Wait, + AllowSynchronousContinuations = false + }); + _writerTask = Task.Run(WriteLoopAsync); } public void Ready(McpReadyInfo info) @@ -35,21 +49,21 @@ public sealed class McpHostLogObserver : IMcpHostObserver, IDisposable { ["protocol"] = info.ProtocolVersion, ["pid"] = info.ProcessId, ["logs"] = _logDirectory }); - _application.Operational(message); + Enqueue(() => _application.Operational(message)); } public void HttpRequestStarted(McpHttpRequestStartInfo info) { if (_observedConnections.TryAdd(info.ConnectionId, 0)) - _application.Information($"HTTP connection observed connection={info.ConnectionId} remote={info.RemoteIp}:{info.RemotePort}"); - _application.Information($"HTTP request started connection={info.ConnectionId} trace={info.TraceId} remote={info.RemoteIp}:{info.RemotePort} method={info.Method} path={info.Path}"); + Enqueue(() => _application.Information($"HTTP connection observed connection={info.ConnectionId} remote={info.RemoteIp}:{info.RemotePort}")); + Enqueue(() => _application.Information($"HTTP request started connection={info.ConnectionId} trace={info.TraceId} remote={info.RemoteIp}:{info.RemotePort} method={info.Method} path={info.Path}")); } public void HttpRequest(McpHttpRequestInfo info) { - _application.Information(string.Create(CultureInfo.InvariantCulture, - $"HTTP request completed connection={info.ConnectionId} remote={info.RemoteIp}:{info.RemotePort} method={info.Method} path={info.Path} status={info.Status} duration_ms={info.DurationMs:F3}")); - _http?.WriteLine(JsonSerializer.Serialize(new + string applicationLine = string.Create(CultureInfo.InvariantCulture, + $"HTTP request completed connection={info.ConnectionId} remote={info.RemoteIp}:{info.RemotePort} method={info.Method} path={info.Path} status={info.Status} duration_ms={info.DurationMs:F3}"); + string? httpLine = _http is null ? null : JsonSerializer.Serialize(new { timestamp = info.Timestamp, connection_id = info.ConnectionId, @@ -68,9 +82,53 @@ public sealed class McpHostLogObserver : IMcpHostObserver, IDisposable mcp_protocol_version = NullIfEmpty(info.ProtocolVersion), mcp_method = NullIfEmpty(info.McpMethod), mcp_name = NullIfEmpty(info.McpName) - })); + }); + Enqueue(() => + { + if (httpLine is not null) _http!.WriteLine(httpLine); + _application.Information(applicationLine); + }); + } + + private void Enqueue(Action write) + { + if (Volatile.Read(ref _disposed) != 0) return; + if (!_queue.Writer.TryWrite(write)) Interlocked.Increment(ref _droppedEntries); + } + + private async Task WriteLoopAsync() + { + try + { + await foreach (Action write in _queue.Reader.ReadAllAsync().ConfigureAwait(false)) + { + try { write(); } + catch (Exception ex) when (ex is not (OutOfMemoryException or StackOverflowException or AccessViolationException)) + { + System.Diagnostics.Debug.WriteLine($"MCP observability write failed: {ex.Message}"); + } + + long dropped = Interlocked.Exchange(ref _droppedEntries, 0); + if (dropped > 0) + { + try { _application.Warning($"MCP observability queue dropped {dropped} log entries."); } + catch (Exception ex) when (ex is not (OutOfMemoryException or StackOverflowException or AccessViolationException)) + { + System.Diagnostics.Debug.WriteLine($"MCP observability drop report failed: {ex.Message}"); + } + } + } + } + finally { _http?.Dispose(); } } private static string? NullIfEmpty(string? value) => string.IsNullOrWhiteSpace(value) ? null : value; - public void Dispose() => _http?.Dispose(); + + public void Dispose() + { + if (Interlocked.Exchange(ref _disposed, 1) != 0) return; + _queue.Writer.TryComplete(); + try { _writerTask.Wait(TimeSpan.FromSeconds(5)); } + catch (AggregateException) { } + } } diff --git a/src/Agent.App/Logging/RollingFileLog.cs b/src/Agent.App/Logging/RollingFileLog.cs index 8bdbbb4..f471a9e 100644 --- a/src/Agent.App/Logging/RollingFileLog.cs +++ b/src/Agent.App/Logging/RollingFileLog.cs @@ -5,12 +5,14 @@ namespace Agent.App.Logging; public sealed class RollingFileLog : IDisposable { private readonly RollingFileWriter _writer; + private readonly TextWriter _consoleWriter; private readonly int _minimumLevel; - public RollingFileLog(string directory, long maxBytes, int retainedFiles, string minimumLevel = "information") + public RollingFileLog(string directory, long maxBytes, int retainedFiles, string minimumLevel = "information", TextWriter? consoleWriter = null) { Directory.CreateDirectory(directory); _writer = new RollingFileWriter(Path.Combine(directory, "application.log"), maxBytes, retainedFiles); + _consoleWriter = consoleWriter ?? Console.Error; _minimumLevel = ParseLevel(minimumLevel); } @@ -24,7 +26,7 @@ public sealed class RollingFileLog : IDisposable { if (!force && Level(level) < _minimumLevel) return; string line = string.Create(CultureInfo.InvariantCulture, $"{DateTimeOffset.UtcNow:O} [{level}] {message}"); - Console.Error.WriteLine(line); + _consoleWriter.WriteLine(line); _writer.WriteLine(line); } diff --git a/tests/Agent.Core.Tests/ObservabilityTests.cs b/tests/Agent.Core.Tests/ObservabilityTests.cs index cd86bc2..c259eb0 100644 --- a/tests/Agent.Core.Tests/ObservabilityTests.cs +++ b/tests/Agent.Core.Tests/ObservabilityTests.cs @@ -1,6 +1,8 @@ using Agent.App.Logging; using Agent.Core.Localization; using Agent.Mcp; +using System.Diagnostics; +using System.Text; namespace Agent.Core.Tests; @@ -43,4 +45,54 @@ public sealed class ObservabilityTests Assert.DoesNotContain("Cookie", httpText, StringComparison.OrdinalIgnoreCase); Assert.DoesNotContain("prompt", httpText, StringComparison.OrdinalIgnoreCase); } + + [Fact] + public void Http_request_callbacks_do_not_wait_for_a_blocked_console_writer() + { + using var directory = new TestDirectory(); + using var console = new BlockingTextWriter(); + LocaleCatalog locale = LocaleCatalog.Load(Path.GetFullPath("config/locales"), "en"); + using var application = new RollingFileLog(directory.Path, 1024 * 1024, 2, consoleWriter: console); + using var observer = new McpHostLogObserver(application, locale, directory.Path, true, 1024 * 1024, 2); + + observer.HttpRequestStarted(new McpHttpRequestStartInfo("connection", "trace-1", "127.0.0.1", 50123, "POST", "/mcp")); + Assert.True(console.WaitUntilBlocked(TimeSpan.FromSeconds(5)), "Background writer did not reach the blocking console."); + + var stopwatch = Stopwatch.StartNew(); + observer.HttpRequest(new McpHttpRequestInfo(DateTimeOffset.UtcNow, "connection", "trace-2", "127.0.0.1", 50124, + "127.0.0.1", 8080, "POST", "/mcp", 200, 12.5, "application/json", 100, "test-client", + "2026-07-28", "tools/list", null)); + stopwatch.Stop(); + + Assert.True(stopwatch.Elapsed < TimeSpan.FromMilliseconds(250), $"HTTP callback blocked for {stopwatch.Elapsed}."); + console.Release(); + } + + private sealed class BlockingTextWriter : TextWriter + { + private readonly ManualResetEventSlim _entered = new(false); + private readonly ManualResetEventSlim _release = new(false); + + public override Encoding Encoding => Encoding.UTF8; + + public override void WriteLine(string? value) + { + _entered.Set(); + _release.Wait(TimeSpan.FromSeconds(30)); + } + + public bool WaitUntilBlocked(TimeSpan timeout) => _entered.Wait(timeout); + public void Release() => _release.Set(); + + protected override void Dispose(bool disposing) + { + if (disposing) + { + _release.Set(); + _entered.Dispose(); + _release.Dispose(); + } + base.Dispose(disposing); + } + } } -- cgit v1.2.3