From e992680b51c6c5fb918b2f96cf1eb4e93a2fc3e5 Mon Sep 17 00:00:00 2001 From: "E. Lewis" Date: Fri, 23 Oct 2026 13:43:09 -0400 Subject: [PATCH 1/3] Open the named pipe on first write and reopen it after a failure A client built while the agent's named pipe didn't exist failed to build, and a client whose pipe broke, for example after an agent restart, failed every write until the application restarted. UDP and Unix sockets don't fail this way. NamedPipeClientChannel now opens the pipe on the first write. When a write fails, it closes the pipe, and the next write opens it again. This matches UnixStreamClientChannel and datadog-go's Windows pipe writer. A missing pipe is now reported to the error handler for each dropped write instead of failing build(). SVLS-8246 --- .../statsd/NamedPipeClientChannel.java | 60 +++++++++++++----- .../statsd/NamedPipeDummyStatsDServer.java | 5 +- .../com/timgroup/statsd/NamedPipeTest.java | 63 +++++++++++++++++++ .../NonBlockingStatsDClientBuilderTest.java | 28 +++++++++ 4 files changed, 141 insertions(+), 15 deletions(-) diff --git a/src/main/java/com/timgroup/statsd/NamedPipeClientChannel.java b/src/main/java/com/timgroup/statsd/NamedPipeClientChannel.java index 2dbcf832..5d101e3a 100644 --- a/src/main/java/com/timgroup/statsd/NamedPipeClientChannel.java +++ b/src/main/java/com/timgroup/statsd/NamedPipeClientChannel.java @@ -1,42 +1,74 @@ package com.timgroup.statsd; -import java.io.FileNotFoundException; import java.io.IOException; import java.io.RandomAccessFile; import java.nio.ByteBuffer; import java.nio.channels.FileChannel; +/** + * A ClientChannel for Windows named pipes. + * + *

The pipe is opened on the first write, not when the client is built, so the client can start + * before the server creates the pipe. After a failed write the pipe is closed and the next write + * opens it again, so the client recovers when the server restarts. + */ class NamedPipeClientChannel implements ClientChannel { - private final RandomAccessFile randomAccessFile; - private final FileChannel fileChannel; private final String pipe; + private RandomAccessFile randomAccessFile; + private FileChannel fileChannel; /** * Creates a new NamedPipeClientChannel with the given address. * * @param address Location of named pipe - * @throws FileNotFoundException if pipe does not exist */ - NamedPipeClientChannel(NamedPipeSocketAddress address) throws FileNotFoundException { + NamedPipeClientChannel(NamedPipeSocketAddress address) { pipe = address.getPipe(); - randomAccessFile = new RandomAccessFile(pipe, "rw"); - fileChannel = randomAccessFile.getChannel(); } @Override - public boolean isOpen() { - return fileChannel.isOpen(); + public synchronized boolean isOpen() { + return fileChannel != null && fileChannel.isOpen(); } @Override - public int write(ByteBuffer src) throws IOException { - return fileChannel.write(src); + public synchronized int write(ByteBuffer src) throws IOException { + connectIfNeeded(); + try { + return fileChannel.write(src); + } catch (IOException e) { + // The pipe is broken, for example because the server restarted. Reopen on next write. + try { + disconnect(); + } catch (IOException ignored) { + // Report the write error, not the close error. + } + throw e; + } + } + + private void connectIfNeeded() throws IOException { + if (fileChannel == null) { + randomAccessFile = new RandomAccessFile(pipe, "rw"); + fileChannel = randomAccessFile.getChannel(); + } + } + + private void disconnect() throws IOException { + if (randomAccessFile != null) { + try { + // closing the file also closes the channel + randomAccessFile.close(); + } finally { + randomAccessFile = null; + fileChannel = null; + } + } } @Override - public void close() throws IOException { - // closing the file also closes the channel - randomAccessFile.close(); + public synchronized void close() throws IOException { + disconnect(); } @Override diff --git a/src/test/java/com/timgroup/statsd/NamedPipeDummyStatsDServer.java b/src/test/java/com/timgroup/statsd/NamedPipeDummyStatsDServer.java index 19cf7dae..8181fc85 100644 --- a/src/test/java/com/timgroup/statsd/NamedPipeDummyStatsDServer.java +++ b/src/test/java/com/timgroup/statsd/NamedPipeDummyStatsDServer.java @@ -80,7 +80,10 @@ protected void receive(ByteBuffer packet) throws IOException { } @Override - public void close() throws IOException { + public synchronized void close() throws IOException { + if (!isOpen) { + return; + } isOpen = false; Kernel32.INSTANCE.CloseHandle(hNamedPipe); } diff --git a/src/test/java/com/timgroup/statsd/NamedPipeTest.java b/src/test/java/com/timgroup/statsd/NamedPipeTest.java index 65b0e1de..7ba79a71 100644 --- a/src/test/java/com/timgroup/statsd/NamedPipeTest.java +++ b/src/test/java/com/timgroup/statsd/NamedPipeTest.java @@ -73,4 +73,67 @@ public void sends_to_statsd() { } assertThat(lastException.getMessage(), nullValue()); } + + @Test(timeout = 5000L) + public void sends_to_pipe_created_after_build() throws Exception { + String pipeName = "testPipe-late-" + random.nextInt(10000); + NonBlockingStatsDClient lateClient = newClient(pipeName, new RecordingErrorHandler()); + DummyStatsDServer lateServer = null; + try { + lateServer = new NamedPipeDummyStatsDServer(pipeName); + lateClient.gauge("late", 1); + lateServer.waitForMessage(); + assertThat(lateServer.messagesReceived(), contains("my.prefix.late:1|g")); + } finally { + lateClient.stop(); + if (lateServer != null) { + lateServer.close(); + } + } + } + + @Test(timeout = 10000L) + public void reconnects_after_server_restart() throws Exception { + String pipeName = "testPipe-restart-" + random.nextInt(10000); + RecordingErrorHandler errorHandler = new RecordingErrorHandler(); + DummyStatsDServer firstServer = new NamedPipeDummyStatsDServer(pipeName); + NonBlockingStatsDClient restartClient = newClient(pipeName, errorHandler); + DummyStatsDServer secondServer = null; + try { + restartClient.gauge("before", 1); + firstServer.waitForMessage(); + firstServer.close(); + + // Send until the client sees the broken pipe and closes its end. + while (errorHandler.getExceptions().isEmpty()) { + restartClient.gauge("during", 1); + Thread.sleep(10); + } + + secondServer = new NamedPipeDummyStatsDServer(pipeName); + // The queue holds one message, so keep sending until one arrives. + while (!secondServer.messagesReceived().contains("my.prefix.after:1|g")) { + restartClient.gauge("after", 1); + Thread.sleep(10); + } + } finally { + restartClient.stop(); + firstServer.close(); + if (secondServer != null) { + secondServer.close(); + } + } + } + + private static NonBlockingStatsDClient newClient( + String pipeName, StatsDClientErrorHandler errorHandler) { + return new NonBlockingStatsDClientBuilder() + .withEnvironmentVariables(new HashMap()) + .prefix("my.prefix") + .namedPipe(pipeName) + .queueSize(1) + .enableAggregation(false) + .errorHandler(errorHandler) + .build(); + } } diff --git a/src/test/java/com/timgroup/statsd/NonBlockingStatsDClientBuilderTest.java b/src/test/java/com/timgroup/statsd/NonBlockingStatsDClientBuilderTest.java index bc349081..abb169ca 100644 --- a/src/test/java/com/timgroup/statsd/NonBlockingStatsDClientBuilderTest.java +++ b/src/test/java/com/timgroup/statsd/NonBlockingStatsDClientBuilderTest.java @@ -5,13 +5,41 @@ import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; +import java.io.FileNotFoundException; import java.util.HashMap; import java.util.Map; +import java.util.Random; import org.junit.Test; import org.junit.function.ThrowingRunnable; public class NonBlockingStatsDClientBuilderTest { + @Test(timeout = 5000L) + public void missing_named_pipe_does_not_fail_build() throws Exception { + // A pipe inside a missing directory fails to open on every platform without creating a + // file. + final Map env = new HashMap<>(); + env.put( + NonBlockingStatsDClient.DD_NAMED_PIPE_ENV_VAR, + "missing-dir-" + new Random().nextInt(1000000) + "/pipe"); + final RecordingErrorHandler errorHandler = new RecordingErrorHandler(); + final NonBlockingStatsDClient client = + new NonBlockingStatsDClientBuilder() + .withEnvironmentVariables(env) + .enableAggregation(false) + .errorHandler(errorHandler) + .build(); + try { + client.gauge("value", 42); + while (errorHandler.getExceptions().isEmpty()) { + Thread.sleep(10); + } + assertTrue(errorHandler.getExceptions().get(0) instanceof FileNotFoundException); + } finally { + client.stop(); + } + } + @Test(timeout = 5000L) public void origin_detection_env_false() throws Exception { final Map env = new HashMap<>(); From 64cc96b51399b84460f9ef1343b1e1e7efcfef2a Mon Sep 17 00:00:00 2001 From: "E. Lewis" Date: Fri, 23 Oct 2026 14:58:02 -0400 Subject: [PATCH 2/3] Close the test pipe server's handle on its listening thread CloseHandle on a synchronous pipe can wait for the listening thread's pending ReadFile. When a client is still connected and idle, as in reconnects_after_server_restart, close() could then block until the client sends again. close() now only marks the server closed, and the listening thread closes the handle after its current read returns. --- .../timgroup/statsd/DummyStatsDServer.java | 50 +++++++++++-------- .../statsd/NamedPipeDummyStatsDServer.java | 17 +++++-- .../com/timgroup/statsd/NamedPipeTest.java | 3 +- 3 files changed, 44 insertions(+), 26 deletions(-) diff --git a/src/test/java/com/timgroup/statsd/DummyStatsDServer.java b/src/test/java/com/timgroup/statsd/DummyStatsDServer.java index 5d87168e..e80d1fae 100644 --- a/src/test/java/com/timgroup/statsd/DummyStatsDServer.java +++ b/src/test/java/com/timgroup/statsd/DummyStatsDServer.java @@ -24,27 +24,10 @@ protected void listen() { new Runnable() { @Override public void run() { - final ByteBuffer packet = - ByteBuffer.allocate(DEFAULT_UDS_MAX_PACKET_SIZE_BYTES); - - while (isOpen() && !Thread.interrupted()) { - if (freeze) { - try { - Thread.sleep(10); - } catch (InterruptedException e) { - } - } else { - try { - ((Buffer) packet) - .clear(); // Cast necessary to handle Java9 - // covariant return types - // see: https://jira.mongodb.org/browse/JAVA-2559 for - // ref. - receive(packet); - handlePacket(packet); - } catch (IOException e) { - } - } + try { + readUntilClosed(); + } finally { + onListenStopped(); } } }); @@ -52,6 +35,31 @@ public void run() { thread.start(); } + private void readUntilClosed() { + final ByteBuffer packet = ByteBuffer.allocate(DEFAULT_UDS_MAX_PACKET_SIZE_BYTES); + + while (isOpen() && !Thread.interrupted()) { + if (freeze) { + try { + Thread.sleep(10); + } catch (InterruptedException e) { + } + } else { + try { + // Cast necessary to handle Java9 covariant return types + // see: https://jira.mongodb.org/browse/JAVA-2559 for ref. + ((Buffer) packet).clear(); + receive(packet); + handlePacket(packet); + } catch (IOException e) { + } + } + } + } + + /** Called on the listening thread after it stops reading. */ + protected void onListenStopped() {} + protected boolean sleepIfFrozen() { if (freeze) { try { diff --git a/src/test/java/com/timgroup/statsd/NamedPipeDummyStatsDServer.java b/src/test/java/com/timgroup/statsd/NamedPipeDummyStatsDServer.java index 8181fc85..78307592 100644 --- a/src/test/java/com/timgroup/statsd/NamedPipeDummyStatsDServer.java +++ b/src/test/java/com/timgroup/statsd/NamedPipeDummyStatsDServer.java @@ -79,12 +79,21 @@ protected void receive(ByteBuffer packet) throws IOException { packet.position(bytesRead.getValue()); } + /** + * Stops the server. The listening thread closes the pipe once its current read returns, so a + * client that is still connected must send something before the pipe closes. + * + *

The handle isn't closed here because the pipe uses synchronous I/O, so {@code CloseHandle} + * from this thread can wait for the listening thread's pending {@code ReadFile}, which returns + * only when the client writes or disconnects. + */ @Override - public synchronized void close() throws IOException { - if (!isOpen) { - return; - } + public void close() throws IOException { isOpen = false; + } + + @Override + protected void onListenStopped() { Kernel32.INSTANCE.CloseHandle(hNamedPipe); } } diff --git a/src/test/java/com/timgroup/statsd/NamedPipeTest.java b/src/test/java/com/timgroup/statsd/NamedPipeTest.java index 7ba79a71..a1119162 100644 --- a/src/test/java/com/timgroup/statsd/NamedPipeTest.java +++ b/src/test/java/com/timgroup/statsd/NamedPipeTest.java @@ -104,7 +104,8 @@ public void reconnects_after_server_restart() throws Exception { firstServer.waitForMessage(); firstServer.close(); - // Send until the client sees the broken pipe and closes its end. + // The server closes its pipe after its next read, so send until the client sees the + // broken pipe and closes its end. while (errorHandler.getExceptions().isEmpty()) { restartClient.gauge("during", 1); Thread.sleep(10); From 9ff975a1ed5f2b29d832539ff4e30f954a28cf8f Mon Sep 17 00:00:00 2001 From: "E. Lewis" Date: Fri, 23 Oct 2026 15:21:06 -0400 Subject: [PATCH 3/3] Keep a closed named pipe channel closed close() only disconnected, so a later write reopened the pipe. A late write during a non-blocking stop() could reopen it after the client closed the channel. close() now marks the channel closed: writes throw ClosedChannelException without opening the pipe, and isOpen() reports false only after close(). A failed write still disconnects, and the next write reopens the pipe. --- .../statsd/NamedPipeClientChannel.java | 11 ++++-- .../statsd/NamedPipeClientChannelTest.java | 34 +++++++++++++++++++ 2 files changed, 43 insertions(+), 2 deletions(-) create mode 100644 src/test/java/com/timgroup/statsd/NamedPipeClientChannelTest.java diff --git a/src/main/java/com/timgroup/statsd/NamedPipeClientChannel.java b/src/main/java/com/timgroup/statsd/NamedPipeClientChannel.java index 5d101e3a..5ac850b5 100644 --- a/src/main/java/com/timgroup/statsd/NamedPipeClientChannel.java +++ b/src/main/java/com/timgroup/statsd/NamedPipeClientChannel.java @@ -3,6 +3,7 @@ import java.io.IOException; import java.io.RandomAccessFile; import java.nio.ByteBuffer; +import java.nio.channels.ClosedChannelException; import java.nio.channels.FileChannel; /** @@ -10,12 +11,14 @@ * *

The pipe is opened on the first write, not when the client is built, so the client can start * before the server creates the pipe. After a failed write the pipe is closed and the next write - * opens it again, so the client recovers when the server restarts. + * opens it again, so the client recovers when the server restarts. Once {@link #close()} is called, + * the channel stays closed and writes throw {@link ClosedChannelException}. */ class NamedPipeClientChannel implements ClientChannel { private final String pipe; private RandomAccessFile randomAccessFile; private FileChannel fileChannel; + private boolean closed; /** * Creates a new NamedPipeClientChannel with the given address. @@ -28,11 +31,14 @@ class NamedPipeClientChannel implements ClientChannel { @Override public synchronized boolean isOpen() { - return fileChannel != null && fileChannel.isOpen(); + return !closed; } @Override public synchronized int write(ByteBuffer src) throws IOException { + if (closed) { + throw new ClosedChannelException(); + } connectIfNeeded(); try { return fileChannel.write(src); @@ -68,6 +74,7 @@ private void disconnect() throws IOException { @Override public synchronized void close() throws IOException { + closed = true; disconnect(); } diff --git a/src/test/java/com/timgroup/statsd/NamedPipeClientChannelTest.java b/src/test/java/com/timgroup/statsd/NamedPipeClientChannelTest.java new file mode 100644 index 00000000..83a909b8 --- /dev/null +++ b/src/test/java/com/timgroup/statsd/NamedPipeClientChannelTest.java @@ -0,0 +1,34 @@ +package com.timgroup.statsd; + +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; + +import java.nio.ByteBuffer; +import java.nio.channels.ClosedChannelException; +import org.junit.Test; +import org.junit.function.ThrowingRunnable; + +public class NamedPipeClientChannelTest { + + @Test + public void write_after_close_throws_without_opening_the_pipe() throws Exception { + // The pipe doesn't exist, so a write that tried to open it would throw + // FileNotFoundException instead. + final NamedPipeClientChannel channel = + new NamedPipeClientChannel(new NamedPipeSocketAddress("missing-dir/pipe")); + assertTrue(channel.isOpen()); + + channel.close(); + + assertFalse(channel.isOpen()); + assertThrows( + ClosedChannelException.class, + new ThrowingRunnable() { + @Override + public void run() throws Exception { + channel.write(ByteBuffer.wrap(new byte[] {1})); + } + }); + } +}