From 9e988580411ca134717e034ba7fbf1d5dc720f43 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 3 Nov 2026 08:06:14 +0000 Subject: [PATCH 1/2] Make compat binary console logging non-blocking The compat binaries log every Halibut event and service call with a synchronous Console.WriteLine on the request path. When the test process stops reading the compat binary's stdout, the pipe buffer fills and the next write blocks, freezing the whole binary (LEV-2253). Route compat binary console output through NonBlockingConsole, which queues lines (bounded) for a background thread and drops and counts lines rather than blocking. Startup lines are flushed before continuing, and output is flushed before exit. FixHungWebSocketsHack keeps its direct console write. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_012CQ12H5eAC2buLoec1L71V --- .../BackwardsCompatProgramBase.cs | 17 ++- .../DelegateCachingService.cs | 13 +- .../DelegateEchoService.cs | 9 +- .../DelegateMultipleParametersTestService.cs | 39 ++--- .../LogUtils/NonBlockingConsole.cs | 28 ++++ .../LogUtils/NonBlockingLineWriter.cs | 139 ++++++++++++++++++ .../LogUtils/TestContextConnectionLog.cs | 2 +- .../ProxyServiceForwardingRequestToClient.cs | 22 +-- .../Services/FixHungWebSocketsHack.cs | 3 + .../Services/StayAliveUntilHelper.cs | 4 +- .../SettingsHelper.cs | 7 +- 11 files changed, 230 insertions(+), 53 deletions(-) create mode 100644 source/Halibut.TestUtils.CompatBinary.Base/LogUtils/NonBlockingConsole.cs create mode 100644 source/Halibut.TestUtils.CompatBinary.Base/LogUtils/NonBlockingLineWriter.cs diff --git a/source/Halibut.TestUtils.CompatBinary.Base/BackwardsCompatProgramBase.cs b/source/Halibut.TestUtils.CompatBinary.Base/BackwardsCompatProgramBase.cs index 80b624fee..f5f4a027c 100644 --- a/source/Halibut.TestUtils.CompatBinary.Base/BackwardsCompatProgramBase.cs +++ b/source/Halibut.TestUtils.CompatBinary.Base/BackwardsCompatProgramBase.cs @@ -17,7 +17,7 @@ public static async Task Main() }); var mode = SettingsHelper.GetSetting("mode") ?? string.Empty; - Console.WriteLine($"Mode is: {mode}"); + NonBlockingConsole.WriteLine($"Mode is: {mode}"); if (mode.Equals("serviceonly")) { @@ -29,7 +29,8 @@ public static async Task Main() } else { - Console.WriteLine($"Unknown mode: {mode}"); + NonBlockingConsole.WriteLine($"Unknown mode: {mode}"); + NonBlockingConsole.FlushBeforeExit(); throw new Exception($"Unknown mode: {mode}"); } @@ -47,7 +48,7 @@ static async Task RunExternalService(CancellationToken cancellationToken) if (serviceConnectionType is ServiceConnectionType.Polling or ServiceConnectionType.PollingOverWebSocket) { addressToPoll = SettingsHelper.GetSetting("octopusservercommsport"); - Console.WriteLine($"Will poll: {addressToPoll}"); + NonBlockingConsole.WriteLine($"Will poll: {addressToPoll}"); } var proxyDetails = SettingsHelper.GetProxyDetails(); @@ -67,22 +68,22 @@ static async Task RunExternalService(CancellationToken cancellationToken) case ServiceConnectionType.PollingOverWebSocket: FixHungWebSocketsHack.EnableHack(); var sslThubprint = SettingsHelper.GetSetting("sslthubprint"); - Console.WriteLine($"Using SSL thumbprint: {sslThubprint}"); + NonBlockingConsole.WriteLine($"Using SSL thumbprint: {sslThubprint}"); tentaclePolling.Poll(new Uri("poll://SQ-TENTAPOLL"), new ServiceEndPoint(new Uri(addressToPoll), sslThubprint, proxyDetails)); break; case ServiceConnectionType.Listening: var port = tentaclePolling.Listen(); - Console.WriteLine($"Listening on port: {port}"); + NonBlockingConsole.WriteLine($"Listening on port: {port}"); tentaclePolling.Trust(octopusThumbprint); break; default: throw new ArgumentOutOfRangeException(); } - Console.WriteLine("RunningAndReady"); - Console.WriteLine("Will Now sleep"); - await Console.Out.FlushAsync(); + NonBlockingConsole.WriteLine("RunningAndReady"); + NonBlockingConsole.WriteLine("Will Now sleep"); + await NonBlockingConsole.FlushAsync(); // Run until the Program is terminated await StayAliveUntilHelper.WaitUntilSignaledToDie(); diff --git a/source/Halibut.TestUtils.CompatBinary.Base/DelegateCachingService.cs b/source/Halibut.TestUtils.CompatBinary.Base/DelegateCachingService.cs index ba23544e7..d2e3151a4 100644 --- a/source/Halibut.TestUtils.CompatBinary.Base/DelegateCachingService.cs +++ b/source/Halibut.TestUtils.CompatBinary.Base/DelegateCachingService.cs @@ -1,5 +1,6 @@ using System; using Halibut.TestUtils.Contracts; +using Halibut.TestUtils.SampleProgram.Base.LogUtils; namespace Halibut.TestUtils.SampleProgram.Base { @@ -14,37 +15,37 @@ public DelegateCachingService(ICachingService cachingService) public Guid NonCachableCall() { - Console.WriteLine("Forwarding NonCachableCall() call to delegate"); + NonBlockingConsole.WriteLine("Forwarding NonCachableCall() call to delegate"); return cachingService.NonCachableCall(); } public Guid CachableCall() { - Console.WriteLine("Forwarding CachableCall() call to delegate"); + NonBlockingConsole.WriteLine("Forwarding CachableCall() call to delegate"); return cachingService.CachableCall(); } public Guid CachableCall(Guid input) { - Console.WriteLine("Forwarding CachableCall(Guid) call to delegate"); + NonBlockingConsole.WriteLine("Forwarding CachableCall(Guid) call to delegate"); return cachingService.CachableCall(input); } public Guid AnotherCachableCall() { - Console.WriteLine("Forwarding AnotherCachableCall() call to delegate"); + NonBlockingConsole.WriteLine("Forwarding AnotherCachableCall() call to delegate"); return cachingService.AnotherCachableCall(); } public Guid CachableCallThatThrowsAnExceptionWithARandomExceptionMessage(string exceptionMessagePrefix) { - Console.WriteLine("Forwarding CachableCallThatThrowsAnExceptionWithARandomExceptionMessage() call to delegate"); + NonBlockingConsole.WriteLine("Forwarding CachableCallThatThrowsAnExceptionWithARandomExceptionMessage() call to delegate"); return cachingService.CachableCallThatThrowsAnExceptionWithARandomExceptionMessage(exceptionMessagePrefix); } public Guid TwoSecondCachableCall() { - Console.WriteLine("Forwarding TwoSecondCachableCall() call to delegate"); + NonBlockingConsole.WriteLine("Forwarding TwoSecondCachableCall() call to delegate"); return cachingService.TwoSecondCachableCall(); } } diff --git a/source/Halibut.TestUtils.CompatBinary.Base/DelegateEchoService.cs b/source/Halibut.TestUtils.CompatBinary.Base/DelegateEchoService.cs index 0856ea080..aff787986 100644 --- a/source/Halibut.TestUtils.CompatBinary.Base/DelegateEchoService.cs +++ b/source/Halibut.TestUtils.CompatBinary.Base/DelegateEchoService.cs @@ -1,6 +1,7 @@ using System; using System.IO; using Halibut.TestUtils.Contracts; +using Halibut.TestUtils.SampleProgram.Base.LogUtils; using Halibut.TestUtils.SampleProgram.Base.Utils; namespace Halibut.TestUtils.SampleProgram.Base @@ -16,19 +17,19 @@ public DelegateEchoService(IEchoService echoService) public int LongRunningOperation() { - Console.WriteLine("Forwarding LongRunningOperation() call to delegate"); + NonBlockingConsole.WriteLine("Forwarding LongRunningOperation() call to delegate"); return echoService.LongRunningOperation(); } public string SayHello(string name) { - Console.WriteLine("Forwarding SayHello() call to delegate"); + NonBlockingConsole.WriteLine("Forwarding SayHello() call to delegate"); return echoService.SayHello(name); } public bool Crash() { - Console.WriteLine("Forwarding Crash() call to delegate"); + NonBlockingConsole.WriteLine("Forwarding Crash() call to delegate"); try { return echoService.Crash(); @@ -44,7 +45,7 @@ public bool Crash() public int CountBytes(DataStream dataStream) { dataStream = dataStream.ConfigureWriterOnReceivedDataStream(); - Console.WriteLine("Forwarding CountBytes() call to delegate"); + NonBlockingConsole.WriteLine("Forwarding CountBytes() call to delegate"); return echoService.CountBytes(dataStream); } } diff --git a/source/Halibut.TestUtils.CompatBinary.Base/DelegateMultipleParametersTestService.cs b/source/Halibut.TestUtils.CompatBinary.Base/DelegateMultipleParametersTestService.cs index 70bec138c..f775a2814 100644 --- a/source/Halibut.TestUtils.CompatBinary.Base/DelegateMultipleParametersTestService.cs +++ b/source/Halibut.TestUtils.CompatBinary.Base/DelegateMultipleParametersTestService.cs @@ -1,5 +1,6 @@ using System; using Halibut.TestUtils.Contracts; +using Halibut.TestUtils.SampleProgram.Base.LogUtils; namespace Halibut.TestUtils.SampleProgram.Base { @@ -14,115 +15,115 @@ public DelegateMultipleParametersTestService(IMultipleParametersTestService mult public void MethodReturningVoid(long a, long b) { - Console.WriteLine("Forwarding MethodReturningVoid() call to delegate"); + NonBlockingConsole.WriteLine("Forwarding MethodReturningVoid() call to delegate"); multipleParametersTestService.MethodReturningVoid(a, b); } public long Add(long a, long b) { - Console.WriteLine("Forwarding Add(long, long) call to delegate"); + NonBlockingConsole.WriteLine("Forwarding Add(long, long) call to delegate"); return multipleParametersTestService.Add(a, b); } public double Add(double a, double b) { - Console.WriteLine("Forwarding Add(double, double) call to delegate"); + NonBlockingConsole.WriteLine("Forwarding Add(double, double) call to delegate"); return multipleParametersTestService.Add(a, b); } public decimal Add(decimal a, decimal b) { - Console.WriteLine("Forwarding Add(decimal, decimal) call to delegate"); + NonBlockingConsole.WriteLine("Forwarding Add(decimal, decimal) call to delegate"); return multipleParametersTestService.Add(a, b); } public string Hello() { - Console.WriteLine("Forwarding Hello() call to delegate"); + NonBlockingConsole.WriteLine("Forwarding Hello() call to delegate"); return multipleParametersTestService.Hello(); } public string Hello(string a) { - Console.WriteLine("Forwarding Hello(1 param) call to delegate"); + NonBlockingConsole.WriteLine("Forwarding Hello(1 param) call to delegate"); return multipleParametersTestService.Hello(a); } public string Hello(string a, string b) { - Console.WriteLine("Forwarding Hello(2 params) call to delegate"); + NonBlockingConsole.WriteLine("Forwarding Hello(2 params) call to delegate"); return multipleParametersTestService.Hello(a, b); } public string Hello(string a, string b, string c) { - Console.WriteLine("Forwarding Hello(3 params) call to delegate"); + NonBlockingConsole.WriteLine("Forwarding Hello(3 params) call to delegate"); return multipleParametersTestService.Hello(a, b, c); } public string Hello(string a, string b, string c, string d) { - Console.WriteLine("Forwarding Hello(4 params) call to delegate"); + NonBlockingConsole.WriteLine("Forwarding Hello(4 params) call to delegate"); return multipleParametersTestService.Hello(a, b, c, d); } public string Hello(string a, string b, string c, string d, string e) { - Console.WriteLine("Forwarding Hello(5 params) call to delegate"); + NonBlockingConsole.WriteLine("Forwarding Hello(5 params) call to delegate"); return multipleParametersTestService.Hello(a, b, c, d, e); } public string Hello(string a, string b, string c, string d, string e, string f) { - Console.WriteLine("Forwarding Hello(6 params) call to delegate"); + NonBlockingConsole.WriteLine("Forwarding Hello(6 params) call to delegate"); return multipleParametersTestService.Hello(a, b, c, d, e, f); } public string Hello(string a, string b, string c, string d, string e, string f, string g) { - Console.WriteLine("Forwarding Hello(7 params) call to delegate"); + NonBlockingConsole.WriteLine("Forwarding Hello(7 params) call to delegate"); return multipleParametersTestService.Hello(a, b, c, d, e, f, g); } public string Hello(string a, string b, string c, string d, string e, string f, string g, string h) { - Console.WriteLine("Forwarding Hello(8 params) call to delegate"); + NonBlockingConsole.WriteLine("Forwarding Hello(8 params) call to delegate"); return multipleParametersTestService.Hello(a, b, c, d, e, f, g, h); } public string Hello(string a, string b, string c, string d, string e, string f, string g, string h, string i) { - Console.WriteLine("Forwarding Hello(9 params) call to delegate"); + NonBlockingConsole.WriteLine("Forwarding Hello(9 params) call to delegate"); return multipleParametersTestService.Hello(a, b, c, d, e, f, g, h, i); } public string Hello(string a, string b, string c, string d, string e, string f, string g, string h, string i, string j) { - Console.WriteLine("Forwarding Hello(10 params) call to delegate"); + NonBlockingConsole.WriteLine("Forwarding Hello(10 params) call to delegate"); return multipleParametersTestService.Hello(a, b, c, d, e, f, g, h, i, j); } public string Hello(string a, string b, string c, string d, string e, string f, string g, string h, string i, string j, string k) { - Console.WriteLine("Forwarding Hello(11 params) call to delegate"); + NonBlockingConsole.WriteLine("Forwarding Hello(11 params) call to delegate"); return multipleParametersTestService.Hello(a, b, c, d, e, f, g, h, i, j, k); } public string Ambiguous(string a, string b) { - Console.WriteLine("Forwarding Ambiguous(string, string) call to delegate"); + NonBlockingConsole.WriteLine("Forwarding Ambiguous(string, string) call to delegate"); return multipleParametersTestService.Ambiguous(a, b); } public string Ambiguous(string a, Tuple b) { - Console.WriteLine("Forwarding Ambiguous(string, Tuple) call to delegate"); + NonBlockingConsole.WriteLine("Forwarding Ambiguous(string, Tuple) call to delegate"); return multipleParametersTestService.Ambiguous(a, b); } public MapLocation GetLocation(MapLocation loc) { - Console.WriteLine("Forwarding GetLocation() call to delegate"); + NonBlockingConsole.WriteLine("Forwarding GetLocation() call to delegate"); return multipleParametersTestService.GetLocation(loc); } } diff --git a/source/Halibut.TestUtils.CompatBinary.Base/LogUtils/NonBlockingConsole.cs b/source/Halibut.TestUtils.CompatBinary.Base/LogUtils/NonBlockingConsole.cs new file mode 100644 index 000000000..5eab161e2 --- /dev/null +++ b/source/Halibut.TestUtils.CompatBinary.Base/LogUtils/NonBlockingConsole.cs @@ -0,0 +1,28 @@ +using System; +using System.Threading.Tasks; + +namespace Halibut.TestUtils.SampleProgram.Base.LogUtils +{ + /// + /// Use this instead of everywhere in the compat binaries. + /// See for why. + /// + public static class NonBlockingConsole + { + static readonly TimeSpan DefaultFlushTimeout = TimeSpan.FromSeconds(10); + + static readonly NonBlockingLineWriter Writer = new(Console.Out, threadName: "CompatBinaryConsoleWriter"); + + public static void WriteLine(string line) => Writer.WriteLine(line); + + /// + /// Waits for queued lines to reach stdout, e.g. so the test process sees "RunningAndReady". + /// + public static Task FlushAsync() => Writer.FlushAsync(DefaultFlushTimeout); + + /// + /// Waits (briefly) for queued lines to reach stdout before the process exits. + /// + public static bool FlushBeforeExit() => Writer.Flush(TimeSpan.FromSeconds(2)); + } +} diff --git a/source/Halibut.TestUtils.CompatBinary.Base/LogUtils/NonBlockingLineWriter.cs b/source/Halibut.TestUtils.CompatBinary.Base/LogUtils/NonBlockingLineWriter.cs new file mode 100644 index 000000000..2041b2524 --- /dev/null +++ b/source/Halibut.TestUtils.CompatBinary.Base/LogUtils/NonBlockingLineWriter.cs @@ -0,0 +1,139 @@ +using System; +using System.Collections.Concurrent; +using System.IO; +using System.Threading; +using System.Threading.Tasks; + +namespace Halibut.TestUtils.SampleProgram.Base.LogUtils +{ + /// + /// Writes lines to a from a dedicated background thread, so callers never block on the write. + /// + /// The compat binaries write their logs to stdout, which the test process reads through a pipe. If the test process + /// stops reading, the pipe buffer (about 4KB on Windows) fills up and a plain Console.WriteLine blocks. Because Halibut + /// logs on the request path, that blocked write freezes the whole compat binary: in-flight requests stop, and new + /// connections are never accepted (see LEV-2253). + /// + /// With this writer, a stalled reader only stalls the background thread. Callers queue lines up to a fixed capacity, + /// after which lines are dropped (and counted) rather than blocking the Halibut protocol. + /// + public class NonBlockingLineWriter : IDisposable + { + public const int DefaultCapacity = 10_000; + + readonly TextWriter target; + readonly BlockingCollection lines; + readonly Thread writerThread; + long droppedLines; + long pendingLines; + + public NonBlockingLineWriter(TextWriter target, int capacity = DefaultCapacity, string threadName = nameof(NonBlockingLineWriter)) + { + if (capacity <= 0) throw new ArgumentOutOfRangeException(nameof(capacity), capacity, "Capacity must be greater than zero."); + + this.target = target ?? throw new ArgumentNullException(nameof(target)); + lines = new BlockingCollection(new ConcurrentQueue(), capacity); + writerThread = new Thread(WriteLoop) + { + IsBackground = true, + Name = threadName + }; + writerThread.Start(); + } + + /// + /// The number of lines dropped because the queue was full, and not yet reported to the target. + /// + public long DroppedLines => Interlocked.Read(ref droppedLines); + + /// + /// Queues a line to be written. Never blocks. If the queue is full the line is dropped, and a count of dropped + /// lines is written to the target once it starts accepting writes again. + /// + public void WriteLine(string line) + { + Interlocked.Increment(ref pendingLines); + + bool added; + try + { + added = lines.TryAdd(line); + } + catch (InvalidOperationException) + { + // Disposed (CompleteAdding was called). + added = false; + } + + if (!added) + { + Interlocked.Decrement(ref pendingLines); + Interlocked.Increment(ref droppedLines); + } + } + + /// + /// Waits until every queued line has been written to the target, or the timeout elapses. + /// Use this where the test process must see a line before continuing, such as "RunningAndReady". + /// + /// True if every queued line was written before the timeout. + public async Task FlushAsync(TimeSpan timeout) + { + var deadline = DateTime.UtcNow + timeout; + while (Interlocked.Read(ref pendingLines) > 0) + { + if (DateTime.UtcNow >= deadline) return false; + await Task.Delay(10).ConfigureAwait(false); + } + + return true; + } + + /// + /// Synchronous version of , for use just before the process exits. + /// + public bool Flush(TimeSpan timeout) + { + var deadline = DateTime.UtcNow + timeout; + while (Interlocked.Read(ref pendingLines) > 0) + { + if (DateTime.UtcNow >= deadline) return false; + Thread.Sleep(10); + } + + return true; + } + + void WriteLoop() + { + foreach (var line in lines.GetConsumingEnumerable()) + { + try + { + var dropped = Interlocked.Exchange(ref droppedLines, 0); + if (dropped > 0) + { + target.WriteLine($"[{nameof(NonBlockingLineWriter)}] {dropped} log line(s) were dropped because the output was not being read."); + } + + target.WriteLine(line); + target.Flush(); + } + catch (Exception) + { + // Nothing sensible can be done if the target is broken (e.g. the pipe was closed), + // and an exception here must not take down the compat binary. + } + finally + { + Interlocked.Decrement(ref pendingLines); + } + } + } + + public void Dispose() + { + lines.CompleteAdding(); + } + } +} diff --git a/source/Halibut.TestUtils.CompatBinary.Base/LogUtils/TestContextConnectionLog.cs b/source/Halibut.TestUtils.CompatBinary.Base/LogUtils/TestContextConnectionLog.cs index 48dc47f6b..dce596355 100644 --- a/source/Halibut.TestUtils.CompatBinary.Base/LogUtils/TestContextConnectionLog.cs +++ b/source/Halibut.TestUtils.CompatBinary.Base/LogUtils/TestContextConnectionLog.cs @@ -41,7 +41,7 @@ void WriteInternal(LogEvent logEvent) if (logEventLogLevel >= logLevel) { var logMessage = string.Format("{5, 16}: {0}:{1} {2} {3} {4}", logEvent.Type, logEvent.Error, endpoint, Thread.CurrentThread.ManagedThreadId, logEvent.FormattedMessage, name); - Console.WriteLine(logMessage); + NonBlockingConsole.WriteLine(logMessage); } } diff --git a/source/Halibut.TestUtils.CompatBinary.Base/ProxyServiceForwardingRequestToClient.cs b/source/Halibut.TestUtils.CompatBinary.Base/ProxyServiceForwardingRequestToClient.cs index daa092a6a..9fd4d01ce 100644 --- a/source/Halibut.TestUtils.CompatBinary.Base/ProxyServiceForwardingRequestToClient.cs +++ b/source/Halibut.TestUtils.CompatBinary.Base/ProxyServiceForwardingRequestToClient.cs @@ -29,14 +29,14 @@ public static async Task Run(CancellationToken cancellationToken) if (serviceConnectionType is ServiceConnectionType.Polling or ServiceConnectionType.PollingOverWebSocket) { proxyClientAddressToPoll = SettingsHelper.GetSetting("octopusservercommsport"); - Console.WriteLine($"Proxy Service will Poll: {proxyClientAddressToPoll}"); + NonBlockingConsole.WriteLine($"Proxy Service will Poll: {proxyClientAddressToPoll}"); } var serviceAddressToConnectTo = string.Empty; if (serviceConnectionType == ServiceConnectionType.Listening) { serviceAddressToConnectTo = SettingsHelper.GetSetting("realServiceListenAddress"); - Console.WriteLine($"Proxy Service will forward request to: {serviceAddressToConnectTo}"); + NonBlockingConsole.WriteLine($"Proxy Service will forward request to: {serviceAddressToConnectTo}"); } // A Halibut Runtime Client which is the previous version of Halibut the Test is trying to test. @@ -46,7 +46,7 @@ public static async Task Run(CancellationToken cancellationToken) .WithLogFactory(new TestContextLogFactory("PreviousVersionOfClient", SettingsHelper.GetHalibutLogLevel())) .Build()) { - Console.WriteLine("Client will trust: " + serviceCert.Thumbprint); + NonBlockingConsole.WriteLine("Client will trust: " + serviceCert.Thumbprint); client.Trust(serviceCert.Thumbprint); ServiceEndPoint realServiceEndpoint; @@ -55,16 +55,16 @@ public static async Task Run(CancellationToken cancellationToken) case ServiceConnectionType.Polling: var clientPollingListeningPort = client.Listen(); // Do not change the log message as it is used by the HalibutTestBinaryRunners - Console.WriteLine("Polling listener is listening on port: " + clientPollingListeningPort); + NonBlockingConsole.WriteLine("Polling listener is listening on port: " + clientPollingListeningPort); realServiceEndpoint = new ServiceEndPoint(new Uri("poll://SQ-TENTAPOLL"), serviceCert.Thumbprint, proxyDetails); break; case ServiceConnectionType.PollingOverWebSocket: using (await TcpPortHelper.WaitForLock(cancellationToken)) { var webSocketListeningPort = WebSocketListeningPort(client, out var webSocketListeningUrl, cancellationToken); - Console.WriteLine($"WebSocket Polling listener is listening on: {webSocketListeningUrl}"); + NonBlockingConsole.WriteLine($"WebSocket Polling listener is listening on: {webSocketListeningUrl}"); // Do not change the log message as it is used by the HalibutTestBinaryRunners - Console.WriteLine("Polling listener is listening on port: " + webSocketListeningPort); + NonBlockingConsole.WriteLine("Polling listener is listening on port: " + webSocketListeningPort); realServiceEndpoint = new ServiceEndPoint(new Uri("poll://SQ-TENTAPOLL"), serviceCert.Thumbprint, proxyDetails); } break; @@ -95,17 +95,17 @@ public static async Task Run(CancellationToken cancellationToken) case ServiceConnectionType.Listening: var proxyServiceListeningPort = proxyService.Listen(); // Do not change the log message as it is used by the HalibutTestBinaryRunners - Console.WriteLine($"Listening on port: {proxyServiceListeningPort}"); + NonBlockingConsole.WriteLine($"Listening on port: {proxyServiceListeningPort}"); proxyService.Trust(clientCert.Thumbprint); break; default: throw new ArgumentOutOfRangeException(); } - Console.WriteLine("RunningAndReady"); - Console.WriteLine("Will Now sleep"); + NonBlockingConsole.WriteLine("RunningAndReady"); + NonBlockingConsole.WriteLine("Will Now sleep"); - await Console.Out.FlushAsync(); + await NonBlockingConsole.FlushAsync(); // Run until the Program is terminated await StayAliveUntilHelper.WaitUntilSignaledToDie(); @@ -125,7 +125,7 @@ public static int WebSocketListeningPort(HalibutRuntime client, out string webSo } catch (HttpListenerException e) { - Console.WriteLine($"Failed to listen for websocket, trying again. {e}"); + NonBlockingConsole.WriteLine($"Failed to listen for websocket, trying again. {e}"); } } diff --git a/source/Halibut.TestUtils.CompatBinary.Base/Services/FixHungWebSocketsHack.cs b/source/Halibut.TestUtils.CompatBinary.Base/Services/FixHungWebSocketsHack.cs index 783da8e25..7f48f1db8 100644 --- a/source/Halibut.TestUtils.CompatBinary.Base/Services/FixHungWebSocketsHack.cs +++ b/source/Halibut.TestUtils.CompatBinary.Base/Services/FixHungWebSocketsHack.cs @@ -25,6 +25,9 @@ public static void EnableHack() { Thread.Sleep(10000); // Print anything to console. + // This deliberately writes to the console directly rather than through NonBlockingConsole, + // since it is the write itself (on this thread) that appears to unstick the CLR. It is on its + // own background thread, so blocking here cannot freeze the Halibut protocol. Console.WriteLine("Compat binary is still running"); } }); diff --git a/source/Halibut.TestUtils.CompatBinary.Base/Services/StayAliveUntilHelper.cs b/source/Halibut.TestUtils.CompatBinary.Base/Services/StayAliveUntilHelper.cs index 6661201bd..f76bff658 100644 --- a/source/Halibut.TestUtils.CompatBinary.Base/Services/StayAliveUntilHelper.cs +++ b/source/Halibut.TestUtils.CompatBinary.Base/Services/StayAliveUntilHelper.cs @@ -2,6 +2,7 @@ using System.IO; using System.Threading; using System.Threading.Tasks; +using Halibut.TestUtils.SampleProgram.Base.LogUtils; namespace Halibut.TestUtils.SampleProgram.Base.Services { @@ -19,7 +20,7 @@ public static async Task WaitUntilSignaledToDie() #endif using (var fileStreamLock = new FileStream(stayAliveFile, FileMode.Open, FileAccess.Read, FileShare.None)) { - Console.WriteLine("Got a lock, going to die now"); + NonBlockingConsole.WriteLine("Got a lock, going to die now"); fileStreamLock.Dispose(); } @@ -29,6 +30,7 @@ public static async Task WaitUntilSignaledToDie() } finally { + NonBlockingConsole.FlushBeforeExit(); Environment.Exit(0); } } diff --git a/source/Halibut.TestUtils.CompatBinary.Base/SettingsHelper.cs b/source/Halibut.TestUtils.CompatBinary.Base/SettingsHelper.cs index 5deec8ffb..180d3f023 100644 --- a/source/Halibut.TestUtils.CompatBinary.Base/SettingsHelper.cs +++ b/source/Halibut.TestUtils.CompatBinary.Base/SettingsHelper.cs @@ -1,6 +1,7 @@ using System; using System.Security.Cryptography.X509Certificates; using Halibut.Logging; +using Halibut.TestUtils.SampleProgram.Base.LogUtils; using Halibut.Transport.Proxy; namespace Halibut.TestUtils.SampleProgram.Base @@ -18,7 +19,7 @@ static class SettingsHelper var port = GetSetting("proxydetails_port"); var type = GetSetting("proxydetails_type"); - Console.WriteLine($"Using Proxy Details Hot:{host} Port:{port} Type:{type} Username:{userName} Password:{password}"); + NonBlockingConsole.WriteLine($"Using Proxy Details Hot:{host} Port:{port} Type:{type} Username:{userName} Password:{password}"); return new ProxyDetails(host, AsInt(port), AsPortType(type), userName, password); @@ -33,7 +34,7 @@ int AsInt(string s) } } - Console.WriteLine("Not using an Http Proxy"); + NonBlockingConsole.WriteLine("Not using an Http Proxy"); return null; } @@ -122,7 +123,7 @@ public static string CompatBinaryStayAliveLockFile() throw new Exception("Env var CompatBinaryStayAliveFilePath must be set"); } - Console.WriteLine($"Will die when the following file can be locked or is deleted '{stayAliveFilePath}'."); + NonBlockingConsole.WriteLine($"Will die when the following file can be locked or is deleted '{stayAliveFilePath}'."); return stayAliveFilePath; } From cad632052d25a9532190a53a69532058d38ee443 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 3 Nov 2026 08:11:15 +0000 Subject: [PATCH 2/2] Add tests for NonBlockingLineWriter Links the compat binary's NonBlockingLineWriter into Halibut.Tests and checks it keeps order, never blocks callers when the target is blocked, reports dropped lines once output resumes, times out flushes on a blocked target, and survives a throwing target. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_012CQ12H5eAC2buLoec1L71V --- source/Halibut.Tests/Halibut.Tests.csproj | 4 + .../NonBlockingLineWriterFixture.cs | 144 ++++++++++++++++++ 2 files changed, 148 insertions(+) create mode 100644 source/Halibut.Tests/Support/BackwardsCompatibility/NonBlockingLineWriterFixture.cs diff --git a/source/Halibut.Tests/Halibut.Tests.csproj b/source/Halibut.Tests/Halibut.Tests.csproj index 526854c7d..e59695d8f 100644 --- a/source/Halibut.Tests/Halibut.Tests.csproj +++ b/source/Halibut.Tests/Halibut.Tests.csproj @@ -17,6 +17,10 @@ net10.0 + + + + PreserveNewest diff --git a/source/Halibut.Tests/Support/BackwardsCompatibility/NonBlockingLineWriterFixture.cs b/source/Halibut.Tests/Support/BackwardsCompatibility/NonBlockingLineWriterFixture.cs new file mode 100644 index 000000000..b40ed816a --- /dev/null +++ b/source/Halibut.Tests/Support/BackwardsCompatibility/NonBlockingLineWriterFixture.cs @@ -0,0 +1,144 @@ +using System; +using System.Diagnostics; +using System.IO; +using System.Linq; +using System.Text; +using System.Threading; +using System.Threading.Tasks; +using FluentAssertions; +using Halibut.TestUtils.SampleProgram.Base.LogUtils; +using NUnit.Framework; + +namespace Halibut.Tests.Support.BackwardsCompatibility +{ + public class NonBlockingLineWriterFixture : BaseTest + { + [Test] + public async Task WritesLinesInOrder() + { + var target = new StringWriter(); + using var writer = new NonBlockingLineWriter(target); + + for (var i = 0; i < 100; i++) writer.WriteLine($"line {i}"); + + (await writer.FlushAsync(TimeSpan.FromSeconds(10))).Should().BeTrue(); + + ReadLines(target).Should().Equal(Enumerable.Range(0, 100).Select(i => $"line {i}")); + } + + [Test] + public async Task DoesNotBlockWhenTheTargetIsBlocked_AndReportsDroppedLinesOnceUnblocked() + { + const int capacity = 10; + using var target = new BlockingTextWriter(); + using var writer = new NonBlockingLineWriter(target, capacity); + + // The first line is taken by the writer thread, which then blocks on the target. + writer.WriteLine("first"); + target.WaitUntilAWriteIsBlocked(TimeSpan.FromSeconds(10)).Should().BeTrue(); + + var stopwatch = Stopwatch.StartNew(); + for (var i = 0; i < capacity + 50; i++) writer.WriteLine($"queued {i}"); + stopwatch.Stop(); + + stopwatch.Elapsed.Should().BeLessThan(TimeSpan.FromSeconds(5), "writing must never wait on a blocked target"); + writer.DroppedLines.Should().Be(50); + + target.Unblock(); + (await writer.FlushAsync(TimeSpan.FromSeconds(10))).Should().BeTrue(); + writer.WriteLine("after unblocking"); + (await writer.FlushAsync(TimeSpan.FromSeconds(10))).Should().BeTrue(); + + var lines = ReadLines(target.Written); + lines.Should().StartWith(new[] { "first" }); + lines.Should().ContainInOrder(Enumerable.Range(0, capacity).Select(i => $"queued {i}")); + lines.Should().Contain(l => l.Contains("50 log line(s) were dropped")); + lines.Last().Should().Be("after unblocking"); + writer.DroppedLines.Should().Be(0); + } + + [Test] + public async Task FlushReturnsFalseWhenTheTargetStaysBlocked() + { + using var target = new BlockingTextWriter(); + using var writer = new NonBlockingLineWriter(target); + + writer.WriteLine("stuck"); + target.WaitUntilAWriteIsBlocked(TimeSpan.FromSeconds(10)).Should().BeTrue(); + + (await writer.FlushAsync(TimeSpan.FromMilliseconds(200))).Should().BeFalse(); + writer.Flush(TimeSpan.FromMilliseconds(200)).Should().BeFalse(); + + target.Unblock(); + (await writer.FlushAsync(TimeSpan.FromSeconds(10))).Should().BeTrue(); + } + + [Test] + public async Task KeepsWritingWhenTheTargetThrows() + { + var target = new ThrowOnceTextWriter(); + using var writer = new NonBlockingLineWriter(target); + + writer.WriteLine("throws"); + writer.WriteLine("written"); + (await writer.FlushAsync(TimeSpan.FromSeconds(10))).Should().BeTrue(); + + ReadLines(target.Written).Should().Equal("written"); + } + + static string[] ReadLines(StringWriter writer) => + writer.ToString().Split(new[] { Environment.NewLine }, StringSplitOptions.RemoveEmptyEntries); + + /// + /// Simulates a stdout pipe that nobody is reading: writes block until is called. + /// + class BlockingTextWriter : TextWriter + { + readonly ManualResetEventSlim unblocked = new(false); + readonly ManualResetEventSlim aWriteIsBlocked = new(false); + + public StringWriter Written { get; } = new(); + public override Encoding Encoding => Encoding.UTF8; + + public bool WaitUntilAWriteIsBlocked(TimeSpan timeout) => aWriteIsBlocked.Wait(timeout); + public void Unblock() => unblocked.Set(); + + public override void WriteLine(string? value) + { + if (!unblocked.IsSet) + { + aWriteIsBlocked.Set(); + unblocked.Wait(); + } + + lock (Written) Written.WriteLine(value); + } + + protected override void Dispose(bool disposing) + { + // Never leave the writer thread blocked after a test. + unblocked.Set(); + base.Dispose(disposing); + } + } + + class ThrowOnceTextWriter : TextWriter + { + bool hasThrown; + + public StringWriter Written { get; } = new(); + public override Encoding Encoding => Encoding.UTF8; + + public override void WriteLine(string? value) + { + if (!hasThrown) + { + hasThrown = true; + throw new IOException("The pipe has been ended."); + } + + Written.WriteLine(value); + } + } + } +}