From 9322832c95342b9160b0020a894b38db1a098b67 Mon Sep 17 00:00:00 2001 From: slayerjain Date: Sun, 8 Nov 2026 15:03:15 +0530 Subject: [PATCH 1/3] fix(dedup): answer every command on the control socket, say proto=2 The unix control socket answered ACK even when a START's counter reset failed or an END's coverage was never sent, and closed with no answer on a blank line or a command it could not parse. Keploy read each of those as taken. A command that threw (a JaCoCo port out of range) ended the accept loop, so every later test's command was refused. Every command the SDK reads now gets one line before the close: "ACK proto=2" when it was taken, "IGNORED proto=2" when not. A START whose reset failed is ignored and so is its END; an END whose coverage could not be collected or sent is ignored and still ends the test. A command that throws is ignored and the server goes on. Only a connection closed with nothing sent gets no answer. proto=2 says this SDK answers every command it reads, so Keploy can read a close with no answer as a command not taken: on macOS a connection closed with its command unread gives the reader a plain close, not an error. Agents that read only an IGNORED prefix, or no answer at all, read "ACK proto=2" as ACK. The TCP transport is unchanged. The data socket path is a field, as the control socket's is, so the rows can bind their own. The smoke test now requires the exact answer. Signed-off-by: slayerjain --- .../io/keploy/dedup/KeployDedupAgent.java | 115 ++++--- .../io/keploy/dedup/ControlSocketTest.java | 287 +++++++++++++++++- scripts/smoke-javaagent.sh | 6 +- 3 files changed, 348 insertions(+), 60 deletions(-) diff --git a/keploy-sdk/src/main/java/io/keploy/dedup/KeployDedupAgent.java b/keploy-sdk/src/main/java/io/keploy/dedup/KeployDedupAgent.java index d601146..9ad923f 100644 --- a/keploy-sdk/src/main/java/io/keploy/dedup/KeployDedupAgent.java +++ b/keploy-sdk/src/main/java/io/keploy/dedup/KeployDedupAgent.java @@ -87,16 +87,28 @@ static String encodeBytecodeFrame(String buildTag, String manifestJson, byte[] z // The unix control socket's path. Not final only so tests can bind one of // their own; the agent never changes it. static volatile String controlSocketPath = "/tmp/coverage_control.sock"; - private static final String DATA_SOCKET_PATH = "/tmp/coverage_data.sock"; + // The unix data socket each test's coverage is sent on. Not final for the + // same reason. + static volatile String dataSocketPath = "/tmp/coverage_data.sock"; private static final String DEFAULT_JACOCO_HOST = "127.0.0.1"; private static final int DEFAULT_JACOCO_PORT = 36320; private static final int SOCKET_TIMEOUT_MILLIS = 3000; private static final int SOCKET_BACKLOG = 50; // The answer to an END not taken (no START for that test came first, as // after a restart): nothing was published for it, so Keploy must not use - // the test's coverage. On the unix socket it is the whole answer; on TCP - // it is a frame naming the test, and ACK still follows (CoverageTcpClient). + // the test's coverage. On the unix socket it starts the whole answer + // (UNIX_REPLY_IGNORED); on TCP it is a frame naming the test, and ACK + // still follows (CoverageTcpClient). static final String IGNORED_REPLY = "IGNORED"; + // The answers on the unix control socket: one line for every command the + // SDK reads, written before it closes the connection. ACK says it took the + // command: it reset the counters for a START, or sent the test's coverage + // for an END. IGNORED says it read the command and did not take it. + // proto=2 says this SDK answers every command it reads, so Keploy can read + // a close with no answer as a command not taken: on macOS a connection + // closed with its command unread gives Keploy no error, only the close. + static final String UNIX_REPLY_TAKEN = "ACK proto=2"; + static final String UNIX_REPLY_IGNORED = IGNORED_REPLY + " proto=2"; private static final AtomicBoolean STARTED = new AtomicBoolean(false); private static final AtomicBoolean SHUTDOWN_HOOK_REGISTERED = new AtomicBoolean(false); @@ -185,7 +197,7 @@ public static boolean start() { // up after the app answered and Keploy had fired its first tests, // whose START and END then reached no one: their coverage was lost. CommandServer server = new CommandServer(collector, - new CoveragePublisher(new File(DATA_SOCKET_PATH)), new File(controlSocketPath)); + new CoveragePublisher(new File(dataSocketPath)), new File(controlSocketPath)); try { server.bind(); } catch (Throwable t) { @@ -399,71 +411,83 @@ public void run() { } } + // handle reads one command and answers it before the close: ACK if it + // was taken, IGNORED if not. Only a connection closed with nothing sent + // gets no answer, as there is no command to answer. private void handle(Socket socket) { try (Socket commandSocket = socket; BufferedReader reader = new BufferedReader( new InputStreamReader(commandSocket.getInputStream(), StandardCharsets.UTF_8))) { String line = reader.readLine(); - if (line == null || line.trim().isEmpty()) { + if (line == null) { return; } - CoverageCommand command = CoverageCommand.parse(line.trim()); - if (command == null) { - return; + boolean taken; + try { + taken = run(line.trim()); + } catch (RuntimeException e) { + // Not taken, and the server goes on: thrown past here, it + // would end the accept loop and every later test's command. + log(Level.SEVERE, "Failed to run Java dedup coverage command: " + line.trim(), e); + taken = false; } - - dispatch(command, commandSocket.getOutputStream()); + writeReply(commandSocket.getOutputStream(), taken ? UNIX_REPLY_TAKEN : UNIX_REPLY_IGNORED); } catch (IOException e) { log(Level.SEVERE, "Failed to handle Java dedup coverage command", e); } } - private void dispatch(CoverageCommand command, OutputStream outputStream) { + // run runs one command line ("START id", "END id") and says whether it + // was taken. + private boolean run(String line) { + CoverageCommand command = CoverageCommand.parse(line); + if (command == null) { + return false; + } synchronized (testCaseLock) { if (command.action == CommandAction.START) { - activeTestId = command.testId; + // Set only once the counters are reset: counters not reset + // hold the lines of the tests before this one, so this START + // is not taken and its END is ignored too. + activeTestId = ""; // Warm up app classes once before the first measured window so // one-time lines aren't charged to the first test. collector.warmup(); - collector.reset(); - writeAck(outputStream); - return; + if (!collector.reset()) { + return false; + } + activeTestId = command.testId; + return true; } - if (command.action == CommandAction.END) { - if (!command.testId.equals(activeTestId)) { - log(Level.SEVERE, - "Ignoring mismatched END command. expected=" + activeTestId + ", actual=" - + command.testId, - null); - // Not ACK: this END publishes nothing, and an ACK told - // Keploy it had, so it waited for a record that never came. - writeReply(outputStream, IGNORED_REPLY); - return; - } + if (!command.testId.equals(activeTestId)) { + log(Level.SEVERE, + "Ignoring mismatched END command. expected=" + activeTestId + ", actual=" + + command.testId, + null); + // Not ACK: this END publishes nothing, and an ACK told + // Keploy it had, so it waited for a record that never came. + return false; + } - try { - Map> executedProbesByClass = collector.capture(); - if (executedProbesByClass.isEmpty()) { - log(Level.FINE, "No Java coverage collected for " + command.testId, null); - } - publisher.publish(command.testId, executedProbesByClass); - } catch (Exception e) { - log(Level.SEVERE, "Failed to collect Java coverage for " + command.testId, e); - } finally { - activeTestId = ""; - writeAck(outputStream); + try { + Map> executedProbesByClass = collector.capture(); + if (executedProbesByClass.isEmpty()) { + log(Level.FINE, "No Java coverage collected for " + command.testId, null); } - return; + publisher.publish(command.testId, executedProbesByClass); + return true; + } catch (Exception e) { + // Not ACK for the same reason: nothing reached Keploy. + log(Level.SEVERE, "Failed to collect Java coverage for " + command.testId, e); + return false; + } finally { + activeTestId = ""; } } } - private void writeAck(OutputStream outputStream) { - writeReply(outputStream, "ACK"); - } - private void writeReply(OutputStream outputStream, String reply) { try { outputStream.write((reply + "\n").getBytes(StandardCharsets.UTF_8)); @@ -735,11 +759,14 @@ private CoverageCollector(JacocoClient jacocoClient, CoverageIndex coverageIndex this.coverageIndex = coverageIndex; } - private void reset() { + // reset zeroes JaCoCo's counters and says whether it did. + private boolean reset() { try { jacocoClient.dump(false, true); + return true; } catch (IOException e) { - log(Level.FINE, "Failed to reset JaCoCo counters", e); + log(Level.WARNING, "Failed to reset JaCoCo counters", e); + return false; } } diff --git a/keploy-sdk/src/test/java/io/keploy/dedup/ControlSocketTest.java b/keploy-sdk/src/test/java/io/keploy/dedup/ControlSocketTest.java index 70c9a76..4873dd7 100644 --- a/keploy-sdk/src/test/java/io/keploy/dedup/ControlSocketTest.java +++ b/keploy-sdk/src/test/java/io/keploy/dedup/ControlSocketTest.java @@ -2,44 +2,86 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; -import java.io.BufferedReader; +import java.io.ByteArrayOutputStream; +import java.io.Closeable; import java.io.File; import java.io.IOException; -import java.io.InputStreamReader; +import java.io.InputStream; import java.io.OutputStream; +import java.net.InetAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; import java.nio.file.Files; import java.nio.file.attribute.PosixFilePermission; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import org.jacoco.core.runtime.RemoteControlReader; +import org.jacoco.core.runtime.RemoteControlWriter; import org.junit.After; import org.junit.Before; import org.junit.Test; +import org.newsclub.net.unix.AFUNIXServerSocket; import org.newsclub.net.unix.AFUNIXSocket; import org.newsclub.net.unix.AFUNIXSocketAddress; +import org.newsclub.net.unix.AFUNIXSocketChannel; /** * The unix control socket Keploy sends START and END on. Keploy fires its first * test as soon as the app answers its health check, so the socket must listen - * by the time start() (premain, before the app's main) returns; and an END the - * SDK does not take must not be answered as taken. + * by the time start() (premain, before the app's main) returns. Every command + * the SDK reads gets one answer line before the close: "ACK proto=2" if it was + * taken, "IGNORED proto=2" if not. Keploy reads a close with no answer from an + * SDK that says proto=2 as a command not taken, so only a connection closed + * with nothing sent may go unanswered. + * + *

The answers are compared with literals, not the SDK's constants: Keploy + * reads the words, so a change to them must fail here. */ public class ControlSocketTest { - private String savedPath; + private static final String JACOCO_PORT_PROPERTY = "keploy.jacoco.port"; + private static final String WARMUP_DISABLED_PROPERTY = "keploy.java.dedup.warmup.disabled"; + private static final String TAKEN = "ACK proto=2\n"; + private static final String IGNORED = "IGNORED proto=2\n"; + + private String savedControlPath; + private String savedDataPath; private File dir; + private FakeJacoco jacoco; + private DataReceiver data; @Before public void setUp() throws IOException { - savedPath = KeployDedupAgent.controlSocketPath; + savedControlPath = KeployDedupAgent.controlSocketPath; + savedDataPath = KeployDedupAgent.dataSocketPath; dir = Files.createTempDirectory("keploy-dedup-control").toFile(); KeployDedupAgent.controlSocketPath = new File(dir, "control.sock").getAbsolutePath(); + KeployDedupAgent.dataSocketPath = new File(dir, "data.sock").getAbsolutePath(); + // The warmup loads every class on the test classpath; nothing here needs it. + System.setProperty(WARMUP_DISABLED_PROPERTY, "true"); } @After - public void tearDown() { + public void tearDown() throws Exception { KeployDedupAgent.stop(); - KeployDedupAgent.controlSocketPath = savedPath; + KeployDedupAgent.controlSocketPath = savedControlPath; + KeployDedupAgent.dataSocketPath = savedDataPath; + System.clearProperty(JACOCO_PORT_PROPERTY); + System.clearProperty(WARMUP_DISABLED_PROPERTY); + if (jacoco != null) { + jacoco.close(); + } + if (data != null) { + data.close(); + } File[] left = dir.listFiles(); if (left != null) { for (File f : left) { @@ -69,7 +111,7 @@ public void startReplacesAStaleSocketFile() throws Exception { File socket = new File(KeployDedupAgent.controlSocketPath); assertTrue(socket.createNewFile()); assertTrue(KeployDedupAgent.start()); - assertEquals("IGNORED", send("END test-set-0/test-1")); + assertEquals(IGNORED, send("END test-set-0/test-1")); } @Test @@ -82,19 +124,236 @@ public void startReportsFailureWhenTheControlSocketCannotBind() { @Test public void endWithNoStartIsAnsweredIgnoredNotAck() throws Exception { assertTrue(KeployDedupAgent.start()); - assertEquals("IGNORED", send("END test-set-0/test-47")); + assertEquals(IGNORED, send("END test-set-0/test-47")); + } + + @Test + public void aTestStartedAndEndedIsAnsweredTakenAndItsCoverageIsSent() throws Exception { + startWithJacoco(); + data = new DataReceiver(KeployDedupAgent.dataSocketPath); + assertEquals(TAKEN, send("START test-set-0/test-1")); + assertEquals(1, jacoco.served.get()); + assertEquals(TAKEN, send("END test-set-0/test-1")); + String payload = data.next(); + assertNotNull("the END's coverage reached the data socket", payload); + assertTrue(payload, payload.contains("\"id\":\"test-set-0/test-1\"")); + } + + @Test + public void anEndForAnotherTestIsIgnoredAndTheTestRunningGoesOn() throws Exception { + startWithJacoco(); + data = new DataReceiver(KeployDedupAgent.dataSocketPath); + assertEquals(TAKEN, send("START test-set-0/test-1")); + assertEquals(IGNORED, send("END test-set-0/test-2")); + assertNull("an ignored END sends no coverage", data.poll()); + assertEquals(TAKEN, send("END test-set-0/test-1")); + assertTrue(data.next().contains("\"id\":\"test-set-0/test-1\"")); + } + + @Test + public void aStartWhoseResetFailedIsIgnoredAndSoIsItsEnd() throws Exception { + startWithJacoco(); + data = new DataReceiver(KeployDedupAgent.dataSocketPath); + assertEquals(TAKEN, send("START test-set-0/test-1")); + jacoco.dropNext.set(1); + assertEquals(IGNORED, send("START test-set-0/test-2")); + // JaCoCo answers again, so only the failed START can make these ENDs + // not taken: the counters still hold test-1's lines and whatever ran + // after, and test-2's window never began. + assertEquals(IGNORED, send("END test-set-0/test-1")); + assertEquals(IGNORED, send("END test-set-0/test-2")); + assertNull("an ignored END sends no coverage", data.poll()); + assertEquals(TAKEN, send("START test-set-0/test-3")); + assertEquals(TAKEN, send("END test-set-0/test-3")); + } + + @Test + public void anEndWhoseCoverageCouldNotBeSentIsIgnoredAndEndsTheTest() throws Exception { + startWithJacoco(); + assertEquals(TAKEN, send("START test-set-0/test-1")); + // No one listens on the data socket. + assertEquals(IGNORED, send("END test-set-0/test-1")); + data = new DataReceiver(KeployDedupAgent.dataSocketPath); + assertEquals("the failed END ended the test", IGNORED, send("END test-set-0/test-1")); + assertNull(data.poll()); + assertEquals(TAKEN, send("START test-set-0/test-2")); + assertEquals(TAKEN, send("END test-set-0/test-2")); + } + + @Test + public void aLineThatIsNoCommandIsIgnored() throws Exception { + startWithJacoco(); + String[] lines = {"", " ", "START", "START ", "END", "PING test-set-0/test-1", + "start test-set-0/test-1", "STARTtest-set-0/test-1"}; + for (String line : lines) { + assertEquals("[" + line + "]", IGNORED, send(line)); + } + byte[] notUtf8 = {(byte) 0xff, (byte) 0xfe, ' ', 'x', '\n'}; + assertEquals("bytes that are not UTF-8", IGNORED, sendBytes(notUtf8)); + assertEquals("none of them reset the counters", 0, jacoco.served.get()); + } + + @Test(timeout = 60000) + public void aCommandWithNoNewlineBeforeTheCloseIsRunAndAnswered() throws Exception { + startWithJacoco(); + assertEquals(TAKEN, sendThenCloseOutput("START test-set-0/test-1".getBytes(StandardCharsets.UTF_8))); + assertEquals(1, jacoco.served.get()); + } + + @Test(timeout = 60000) + public void aConnectionClosedWithNothingSentGetsNoAnswer() throws Exception { + assertTrue(KeployDedupAgent.start()); + assertEquals("", sendThenCloseOutput(new byte[0])); + assertEquals("the server still answers", IGNORED, send("END test-set-0/test-1")); + } + + @Test + public void aCommandThatThrowsIsIgnoredAndTheServerGoesOn() throws Exception { + // A port no socket can have: the reset throws IllegalArgumentException. + System.setProperty(JACOCO_PORT_PROPERTY, "70000"); + assertTrue(KeployDedupAgent.start()); + assertEquals(IGNORED, send("START test-set-0/test-1")); + assertEquals("the server still answers", IGNORED, send("END test-set-0/test-1")); + assertTrue(KeployDedupAgent.isStarted()); + } + + private void startWithJacoco() throws IOException { + jacoco = new FakeJacoco(); + System.setProperty(JACOCO_PORT_PROPERTY, String.valueOf(jacoco.server.getLocalPort())); + assertTrue(KeployDedupAgent.start()); } private static String send(String command) throws IOException { + return sendBytes((command + "\n").getBytes(StandardCharsets.UTF_8)); + } + + // sendBytes sends raw as one command, as Keploy does, and returns all the + // SDK answered before its close. + private static String sendBytes(byte[] raw) throws IOException { try (AFUNIXSocket client = AFUNIXSocket.newInstance()) { client.connect(AFUNIXSocketAddress.of(new File(KeployDedupAgent.controlSocketPath)), 3000); client.setSoTimeout(30000); OutputStream out = client.getOutputStream(); - out.write((command + "\n").getBytes(StandardCharsets.UTF_8)); + out.write(raw); out.flush(); - BufferedReader in = new BufferedReader( - new InputStreamReader(client.getInputStream(), StandardCharsets.UTF_8)); - return in.readLine(); + return readAll(client.getInputStream()); + } + } + + // sendThenCloseOutput sends raw, closes its own side, and returns all the + // SDK answered before its close. A channel: on JDK 21 java.net.Socket's + // shutdownOutput refuses a junixsocket socket as not connected. + private static String sendThenCloseOutput(byte[] raw) throws IOException { + try (AFUNIXSocketChannel client = AFUNIXSocketChannel.open( + AFUNIXSocketAddress.of(new File(KeployDedupAgent.controlSocketPath)))) { + ByteBuffer out = ByteBuffer.wrap(raw); + while (out.hasRemaining()) { + client.write(out); + } + client.shutdownOutput(); + ByteArrayOutputStream all = new ByteArrayOutputStream(); + ByteBuffer in = ByteBuffer.allocate(512); + while (client.read(in) != -1) { + all.write(in.array(), 0, in.position()); + in.clear(); + } + return new String(all.toByteArray(), StandardCharsets.UTF_8); + } + } + + private static String readAll(InputStream in) throws IOException { + ByteArrayOutputStream all = new ByteArrayOutputStream(); + byte[] buf = new byte[512]; + int n; + while ((n = in.read(buf)) != -1) { + all.write(buf, 0, n); + } + return new String(all.toByteArray(), StandardCharsets.UTF_8); + } + + /** + * A JaCoCo tcpserver: it answers each dump or reset with an empty dump. + * dropNext closes that many connections first with no answer, as a JaCoCo + * that went away does. + */ + private static final class FakeJacoco implements Closeable { + + final ServerSocket server; + final AtomicInteger dropNext = new AtomicInteger(); + final AtomicInteger served = new AtomicInteger(); + + FakeJacoco() throws IOException { + server = new ServerSocket(0, 50, InetAddress.getLoopbackAddress()); + Thread thread = new Thread(this::serve, "fake-jacoco"); + thread.setDaemon(true); + thread.start(); + } + + private void serve() { + while (!server.isClosed()) { + try (Socket socket = server.accept()) { + socket.setSoTimeout(30000); + if (dropNext.getAndUpdate(n -> n > 0 ? n - 1 : 0) > 0) { + continue; + } + RemoteControlWriter writer = new RemoteControlWriter(socket.getOutputStream()); + RemoteControlReader reader = new RemoteControlReader(socket.getInputStream()); + reader.setRemoteCommandVisitor((dump, reset) -> { + served.incrementAndGet(); + writer.sendCmdOk(); + }); + reader.read(); + } catch (IOException e) { + // Closed, or the client went away. + } + } + } + + @Override + public void close() throws IOException { + server.close(); + } + } + + /** Keploy's data socket: it keeps each payload sent to it. */ + private static final class DataReceiver implements Closeable { + + final AFUNIXServerSocket server; + final BlockingQueue payloads = new LinkedBlockingQueue<>(); + + DataReceiver(String path) throws IOException { + server = AFUNIXServerSocket.newInstance(); + server.bind(AFUNIXSocketAddress.of(new File(path)), 50); + Thread thread = new Thread(this::serve, "data-receiver"); + thread.setDaemon(true); + thread.start(); + } + + private void serve() { + while (!server.isClosed()) { + try (Socket socket = server.accept()) { + socket.setSoTimeout(30000); + payloads.add(readAll(socket.getInputStream())); + } catch (IOException e) { + // Closed. + } + } + } + + // next waits for the next payload. The SDK sends it before it answers + // the END, so it is there, or on its way, once the END is answered. + String next() throws InterruptedException { + return payloads.poll(30, TimeUnit.SECONDS); + } + + // poll returns a payload already sent, or null. + String poll() throws InterruptedException { + return payloads.poll(200, TimeUnit.MILLISECONDS); + } + + @Override + public void close() throws IOException { + server.close(); } } } diff --git a/scripts/smoke-javaagent.sh b/scripts/smoke-javaagent.sh index 74a3e9e..fd00a46 100755 --- a/scripts/smoke-javaagent.sh +++ b/scripts/smoke-javaagent.sh @@ -129,9 +129,11 @@ public final class SmokeHarness { OutputStream output = socket.getOutputStream(); output.write((command + "\n").getBytes(StandardCharsets.UTF_8)); output.flush(); + // The whole answer: one line saying the command was taken, then + // the close. String ack = readAll(socket.getInputStream()); - if (!ack.contains("ACK")) { - throw new IllegalStateException("missing ACK for " + command + ": " + ack); + if (!"ACK proto=2\n".equals(ack)) { + throw new IllegalStateException("not taken: " + command + ": " + ack); } } } From 844f7fcae5101cff7704332d41054447def5419b Mon Sep 17 00:00:00 2001 From: slayerjain Date: Sun, 8 Nov 2026 15:52:13 +0530 Subject: [PATCH 2/3] fix(dedup): close the control socket on its own thread; IGNORED on TCP stop() closed the unix control socket from the caller's thread and returned at once, with the accept thread still running. junixsocket's accept holds the socket's raw fd number, and when it finds the socket file gone it closes that number. With start() right after stop(), that number could already be the new server's socket, which was then closed with its file still there: every later command was refused. The old thread also deleted the socket file by path once it ended, so a start() on the same path, as the default /tmp/coverage_control.sock always is, lost its file, and every later command found no socket. Once junixsocket had closed the socket under it, the accept loop went on: it spun a core, logged without end, and isStarted() still said true. Now only the accept thread closes its socket. stop() wakes that accept with a connection of its own and waits up to 5 s for the thread to end, so a command in flight is answered first; a start() after it waits for that thread too. A socket closed under the server (its file deleted, or bound again by another process) is said once and ends the server. On the TCP transport, a START whose counter reset failed was answered ACK, and its END sent the lines of the tests before it as this test's. It is now answered "IGNORED " and then ACK, as an END not taken is, and its END is ignored too. A START whose reset threw (a JaCoCo port out of range) ended the TCP thread, so no later command was answered; it is now ignored and the connection goes on. Signed-off-by: slayerjain --- .../io/keploy/dedup/KeployDedupAgent.java | 170 +++++++++++++++--- .../io/keploy/dedup/ControlSocketTest.java | 134 +++++++++----- .../keploy/dedup/CoverageTcpClientTest.java | 80 ++++++++- .../test/java/io/keploy/dedup/FakeJacoco.java | 67 +++++++ 4 files changed, 378 insertions(+), 73 deletions(-) create mode 100644 keploy-sdk/src/test/java/io/keploy/dedup/FakeJacoco.java diff --git a/keploy-sdk/src/main/java/io/keploy/dedup/KeployDedupAgent.java b/keploy-sdk/src/main/java/io/keploy/dedup/KeployDedupAgent.java index 9ad923f..19b992e 100644 --- a/keploy-sdk/src/main/java/io/keploy/dedup/KeployDedupAgent.java +++ b/keploy-sdk/src/main/java/io/keploy/dedup/KeployDedupAgent.java @@ -94,11 +94,11 @@ static String encodeBytecodeFrame(String buildTag, String manifestJson, byte[] z private static final int DEFAULT_JACOCO_PORT = 36320; private static final int SOCKET_TIMEOUT_MILLIS = 3000; private static final int SOCKET_BACKLOG = 50; - // The answer to an END not taken (no START for that test came first, as - // after a restart): nothing was published for it, so Keploy must not use - // the test's coverage. On the unix socket it starts the whole answer - // (UNIX_REPLY_IGNORED); on TCP it is a frame naming the test, and ACK - // still follows (CoverageTcpClient). + // The answer to a command not taken (a START whose reset failed, an END + // with no START for that test, as after a restart): nothing is published + // for that test, so Keploy must not use its coverage. On the unix socket + // it starts the whole answer (UNIX_REPLY_IGNORED); on TCP it is a frame + // naming the test, and ACK still follows (CoverageTcpClient). static final String IGNORED_REPLY = "IGNORED"; // The answers on the unix control socket: one line for every command the // SDK reads, written before it closes the connection. ACK says it took the @@ -115,6 +115,13 @@ static String encodeBytecodeFrame(String buildTag, String manifestJson, byte[] z // The active transport worker: a CommandServer (unix, local/docker) or a // CoverageTcpClient (TCP, k8s). Both are Closeable so stop() is transport-agnostic. private static volatile Closeable coverageWorker; + // The worker's thread. stop() waits for it to end, and start() does too + // when stop() gave up waiting (a command still running): until it ends it + // owns its socket, and a second worker on the same path would lose its + // socket file to the first one's close. + private static volatile Thread coverageThread; + // How long stop(), and a start() after it, wait for the worker's thread. + private static final long WORKER_STOP_WAIT_MILLIS = 5000; private KeployDedupAgent() { } @@ -151,6 +158,11 @@ public static boolean start() { if (!STARTED.compareAndSet(false, true)) { return true; } + if (!awaitWorkerThread()) { + STARTED.set(false); + log(Level.SEVERE, "Java dedup is not started: the worker stopped before is still running a command", null); + return false; + } CoverageCollector collector = new CoverageCollector( new JacocoClient(resolveHost(), resolvePort()), @@ -214,6 +226,7 @@ public static boolean start() { Thread thread = new Thread(worker, threadName); thread.setDaemon(true); + coverageThread = thread; thread.start(); registerShutdownHook(); return true; @@ -241,9 +254,44 @@ public static void stop() { } } coverageWorker = null; + // Wait for the worker to end, so that a start() after this one does + // not overlap it. Still running after the wait, it is in a command or + // in an accept close()'s wake-up did not reach (the socket file was + // bound again by someone else). Either way it is past the point where + // closing its socket from here races its accept, so close it: an + // accept ends, and a command ends on its own once it is answered. + if (!awaitWorkerThread() && worker instanceof CommandServer) { + ((CommandServer) worker).closeSocket(); + if (!awaitWorkerThread()) { + log(Level.WARNING, "Java dedup worker is still running a command; it stops once the command ends", + null); + } + } + if (!isAlive(coverageThread)) { + coverageThread = null; + } STARTED.set(false); } + // awaitWorkerThread waits up to WORKER_STOP_WAIT_MILLIS for the last + // worker's thread to end and says whether it has. + private static boolean awaitWorkerThread() { + Thread thread = coverageThread; + if (thread == null || thread == Thread.currentThread()) { + return true; + } + try { + thread.join(WORKER_STOP_WAIT_MILLIS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return !thread.isAlive(); + } + + private static boolean isAlive(Thread thread) { + return thread != null && thread.isAlive(); + } + private static boolean isDisabled() { return isTruthy(System.getenv("KEPLOY_JAVA_DEDUP_DISABLED")) || isTruthy(System.getProperty("keploy.java.dedup.disabled")); @@ -372,6 +420,8 @@ private static final class CommandServer implements Runnable, Closeable { private final AtomicBoolean running = new AtomicBoolean(true); private final Object testCaseLock = new Object(); private volatile AFUNIXServerSocket serverSocket; + // Set once bind() has the socket listening, as start() then runs run(). + private volatile boolean listening; private String activeTestId = ""; CommandServer(CoverageCollector collector, CoveragePublisher publisher, File controlSocket) { @@ -388,27 +438,56 @@ void bind() throws IOException { deleteSocketFile(controlSocket); serverSocket = AFUNIXServerSocket.newInstance(); serverSocket.bind(AFUNIXSocketAddress.of(controlSocket), SOCKET_BACKLOG); + listening = true; relaxSocketPermissions(controlSocket); } + // run accepts commands until close(). Only this thread closes the + // server socket, and it does so once it has left accept: junixsocket's + // accept holds the socket's raw fd number, and when it finds the + // socket file gone it closes that number. Closed from another thread + // while this one was inside accept, the number could already belong + // to the next server's socket (start() after stop()), which was then + // closed with its file still there: every command was refused. @Override public void run() { try (AFUNIXServerSocket localServer = serverSocket) { while (running.get()) { + Socket accepted; try { - handle(localServer.accept()); + accepted = localServer.accept(); } catch (IOException e) { - if (running.get()) { - log(Level.SEVERE, "Failed to accept Java dedup coverage command", e); + if (!running.get()) { + break; + } + if (localServer.isClosed()) { + // junixsocket closes the socket when its file was + // deleted or bound again by someone else: no + // command can reach it. Said once; looping on it + // spun a core and logged without end. + STARTED.set(false); + log(Level.SEVERE, "Java dedup control socket " + controlSocket.getAbsolutePath() + + " was closed: its file was deleted or replaced, so Keploy's coverage" + + " commands no longer reach this app", e); + break; } + log(Level.SEVERE, "Failed to accept Java dedup coverage command", e); + continue; } + if (!running.get()) { + // close()'s wake-up, or a command that came with it. + closeQuietly(accepted); + break; + } + handle(accepted); } } catch (Throwable t) { STARTED.set(false); log(Level.SEVERE, "Java dedup control socket server is unavailable", t); - } finally { - deleteSocketFile(controlSocket); } + // The close above deleted the socket file, as it is still this + // server's: a start() after stop() binds only once this thread + // has ended (awaitWorkerThread). } // handle reads one command and answers it before the close: ACK if it @@ -497,20 +576,48 @@ private void writeReply(OutputStream outputStream, String reply) { } } + // close stops run() and leaves the socket for run() to close: it + // wakes run()'s accept with a connection of its own. Only when that + // connection cannot be made (the file is gone, so no command can come + // either) does it close the socket from here, as nothing else can + // wake that accept. A socket not listening (bind failed) is closed here. @Override public void close() { running.set(false); + AFUNIXServerSocket localServer = serverSocket; + if (localServer == null || localServer.isClosed()) { + return; + } + if (!listening) { + closeQuietly(localServer); + return; + } + try (AFUNIXSocket wake = AFUNIXSocket.newInstance()) { + wake.connect(AFUNIXSocketAddress.of(controlSocket), SOCKET_TIMEOUT_MILLIS); + } catch (IOException e) { + log(Level.FINE, "Could not wake the Java dedup control socket; closing it", e); + closeQuietly(localServer); + } + } + + // closeSocket closes the socket from here, for stop() once run() has + // outlived close()'s wake-up. + void closeSocket() { AFUNIXServerSocket localServer = serverSocket; if (localServer != null) { - try { - localServer.close(); - } catch (IOException e) { - log(Level.FINE, "Failed to close Java dedup control socket", e); - } + closeQuietly(localServer); } } } + private static void closeQuietly(Closeable closeable) { + try { + closeable.close(); + } catch (IOException e) { + log(Level.FINE, "Failed to close a Java dedup socket", e); + } + } + /** * TCP transport (k8s): the SDK dials the collector and keeps one bidirectional * connection open for the whole replay. Mirrors {@link CommandServer}'s dispatch @@ -519,7 +626,8 @@ public void close() { * collector -> SDK : "START <id>" | "END <id>" * SDK -> collector : "ACK" (after START reset) * "COV <compact-json>" + "ACK" (after END dump) - * "IGNORED <id>" + "ACK" (after END with no START) + * "IGNORED <id>" + "ACK" (after a START whose reset + * failed, or an END with no START) * * The collector starts listening only when replay begins, so the connect loop * retries until it is reachable. @@ -605,11 +713,31 @@ private void connectAndServe() throws IOException { private void dispatch(CoverageCommand command, OutputStream out) throws IOException { synchronized (testCaseLock) { if (command.action == CommandAction.START) { - activeTestId = command.testId; - // Warm up app classes once before the first measured window so - // one-time lines aren't charged to the first test. - collector.warmup(); - collector.reset(); + // As on the unix socket, the test is set only once the + // counters are reset: counters not reset hold the lines of + // the tests before this one, so this START is not taken and + // its END is ignored too. + activeTestId = ""; + boolean reset; + try { + // Warm up app classes once before the first measured window so + // one-time lines aren't charged to the first test. + collector.warmup(); + reset = collector.reset(); + } catch (RuntimeException e) { + // Not taken, and the connection goes on: thrown past + // here, it ended this thread, and no later command + // was answered. + log(Level.SEVERE, "Failed to reset Java coverage for " + command.testId, e); + reset = false; + } + if (reset) { + activeTestId = command.testId; + } else { + // Said before the ACK, as for an END not taken: the + // collector has no other sign this test went unmeasured. + writeLine(out, IGNORED_REPLY + " " + command.testId); + } writeLine(out, "ACK"); return; } diff --git a/keploy-sdk/src/test/java/io/keploy/dedup/ControlSocketTest.java b/keploy-sdk/src/test/java/io/keploy/dedup/ControlSocketTest.java index 4873dd7..44acdc3 100644 --- a/keploy-sdk/src/test/java/io/keploy/dedup/ControlSocketTest.java +++ b/keploy-sdk/src/test/java/io/keploy/dedup/ControlSocketTest.java @@ -12,8 +12,6 @@ import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; -import java.net.InetAddress; -import java.net.ServerSocket; import java.net.Socket; import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; @@ -22,9 +20,6 @@ import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicInteger; -import org.jacoco.core.runtime.RemoteControlReader; -import org.jacoco.core.runtime.RemoteControlWriter; import org.junit.After; import org.junit.Before; import org.junit.Test; @@ -217,6 +212,91 @@ public void aCommandThatThrowsIsIgnoredAndTheServerGoesOn() throws Exception { assertTrue(KeployDedupAgent.isStarted()); } + @Test(timeout = 60000) + public void stopWaitsForTheCommandInFlightAndForItsThread() throws Exception { + startWithJacoco(); + jacoco.holdMillis = 1000; + BlockingQueue answer = new LinkedBlockingQueue<>(); + Thread client = new Thread(() -> { + try { + answer.add(send("START test-set-0/test-1")); + } catch (IOException e) { + answer.add(e.toString()); + } + }); + client.start(); + // The START is in the SDK, waiting on JaCoCo, when stop() is called. + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(30); + while (jacocoConnections() == 0 && System.nanoTime() < deadline) { + Thread.sleep(5); + } + KeployDedupAgent.stop(); + // Before, stop() closed the socket and returned at once, with the + // thread still inside the START. + assertFalse("the control thread ended before stop() returned", controlThreadAlive()); + assertEquals("the START was answered", TAKEN, answer.poll(30, TimeUnit.SECONDS)); + client.join(); + } + + @Test(timeout = 60000) + public void aStartAfterStopOnTheSamePathKeepsItsSocket() throws Exception { + for (int i = 0; i < 20; i++) { + assertTrue(KeployDedupAgent.start()); + assertEquals("round " + i, IGNORED, send("END test-set-0/test-1")); + KeployDedupAgent.stop(); + assertTrue(KeployDedupAgent.start()); + // Bound by the closing server's thread, the file was gone or its + // socket closed: these were refused. + assertEquals("round " + i, IGNORED, send("END test-set-0/test-1")); + assertEquals("round " + i, IGNORED, send("END test-set-0/test-1")); + KeployDedupAgent.stop(); + } + } + + @Test(timeout = 60000) + public void aSocketFileDeletedUnderTheServerEndsItOnceAndSaysSo() throws Exception { + startWithJacoco(); + jacoco.holdMillis = 500; + BlockingQueue answer = new LinkedBlockingQueue<>(); + Thread client = new Thread(() -> { + try { + answer.add(send("START test-set-0/test-1")); + } catch (IOException e) { + answer.add(e.toString()); + } + }); + client.start(); + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(30); + while (jacocoConnections() == 0 && System.nanoTime() < deadline) { + Thread.sleep(5); + } + // Deleted while the START runs: the next accept finds the file gone, + // and junixsocket closes the socket. + assertTrue(new File(KeployDedupAgent.controlSocketPath).delete()); + assertEquals(TAKEN, answer.poll(30, TimeUnit.SECONDS)); + while ((KeployDedupAgent.isStarted() || controlThreadAlive()) && System.nanoTime() < deadline) { + Thread.sleep(5); + } + // Before, the accept loop went on on the closed socket: it spun a core + // and logged without end, and the agent still said it was started. + assertFalse("the agent says it is no longer started", KeployDedupAgent.isStarted()); + assertFalse("the control thread ended", controlThreadAlive()); + client.join(); + } + + private int jacocoConnections() { + return jacoco.accepted.get(); + } + + private static boolean controlThreadAlive() { + for (Thread t : Thread.getAllStackTraces().keySet()) { + if (t.getName().equals("keploy-java-dedup-control") && t.isAlive()) { + return true; + } + } + return false; + } + private void startWithJacoco() throws IOException { jacoco = new FakeJacoco(); System.setProperty(JACOCO_PORT_PROPERTY, String.valueOf(jacoco.server.getLocalPort())); @@ -271,50 +351,6 @@ private static String readAll(InputStream in) throws IOException { return new String(all.toByteArray(), StandardCharsets.UTF_8); } - /** - * A JaCoCo tcpserver: it answers each dump or reset with an empty dump. - * dropNext closes that many connections first with no answer, as a JaCoCo - * that went away does. - */ - private static final class FakeJacoco implements Closeable { - - final ServerSocket server; - final AtomicInteger dropNext = new AtomicInteger(); - final AtomicInteger served = new AtomicInteger(); - - FakeJacoco() throws IOException { - server = new ServerSocket(0, 50, InetAddress.getLoopbackAddress()); - Thread thread = new Thread(this::serve, "fake-jacoco"); - thread.setDaemon(true); - thread.start(); - } - - private void serve() { - while (!server.isClosed()) { - try (Socket socket = server.accept()) { - socket.setSoTimeout(30000); - if (dropNext.getAndUpdate(n -> n > 0 ? n - 1 : 0) > 0) { - continue; - } - RemoteControlWriter writer = new RemoteControlWriter(socket.getOutputStream()); - RemoteControlReader reader = new RemoteControlReader(socket.getInputStream()); - reader.setRemoteCommandVisitor((dump, reset) -> { - served.incrementAndGet(); - writer.sendCmdOk(); - }); - reader.read(); - } catch (IOException e) { - // Closed, or the client went away. - } - } - } - - @Override - public void close() throws IOException { - server.close(); - } - } - /** Keploy's data socket: it keeps each payload sent to it. */ private static final class DataReceiver implements Closeable { diff --git a/keploy-sdk/src/test/java/io/keploy/dedup/CoverageTcpClientTest.java b/keploy-sdk/src/test/java/io/keploy/dedup/CoverageTcpClientTest.java index ddb83dc..38b4a99 100644 --- a/keploy-sdk/src/test/java/io/keploy/dedup/CoverageTcpClientTest.java +++ b/keploy-sdk/src/test/java/io/keploy/dedup/CoverageTcpClientTest.java @@ -16,28 +16,39 @@ import org.junit.Test; /** - * The TCP transport, with the test as the collector the SDK dials (k8s). An END - * the SDK does not take publishes no COV, so it must say which test went - * unmeasured, and must still answer ACK, which is what ends the collector's wait. + * The TCP transport, with the test as the collector the SDK dials (k8s). A + * command the SDK does not take (an END with no START, a START whose reset + * failed) publishes no COV, so it must say which test went unmeasured, and must + * still answer ACK, which is what ends the collector's wait. */ public class CoverageTcpClientTest { private static final String ENDPOINT_PROPERTY = "keploy.coverage.endpoint"; + private static final String JACOCO_PORT_PROPERTY = "keploy.jacoco.port"; + private static final String WARMUP_DISABLED_PROPERTY = "keploy.java.dedup.warmup.disabled"; private ServerSocket collector; + private FakeJacoco jacoco; @Before public void setUp() throws IOException { collector = new ServerSocket(0, 1, InetAddress.getLoopbackAddress()); collector.setSoTimeout(30000); System.setProperty(ENDPOINT_PROPERTY, "127.0.0.1:" + collector.getLocalPort()); + // The warmup loads every class on the test classpath; nothing here needs it. + System.setProperty(WARMUP_DISABLED_PROPERTY, "true"); } @After public void tearDown() throws IOException { KeployDedupAgent.stop(); System.clearProperty(ENDPOINT_PROPERTY); + System.clearProperty(JACOCO_PORT_PROPERTY); + System.clearProperty(WARMUP_DISABLED_PROPERTY); collector.close(); + if (jacoco != null) { + jacoco.close(); + } } @Test @@ -59,4 +70,67 @@ public void endWithNoStartNamesTheTestThenAnswersAck() throws Exception { assertEquals("ACK", in.readLine()); } } + + @Test + public void aStartWhoseResetFailedNamesTheTestThenAnswersAckAndSoDoesItsEnd() throws Exception { + jacoco = new FakeJacoco(); + System.setProperty(JACOCO_PORT_PROPERTY, String.valueOf(jacoco.server.getLocalPort())); + assertTrue(KeployDedupAgent.start()); + try (Socket sdk = collector.accept()) { + sdk.setSoTimeout(30000); + OutputStream out = sdk.getOutputStream(); + BufferedReader in = new BufferedReader( + new InputStreamReader(sdk.getInputStream(), StandardCharsets.UTF_8)); + + send(out, "START test-set-0/test-1"); + assertEquals("ACK", in.readLine()); + // JaCoCo goes away for the next reset: the counters still hold + // test-1's lines, so test-2 is not measured. + jacoco.dropNext.set(1); + send(out, "START test-set-0/test-2"); + assertEquals("IGNORED test-set-0/test-2", in.readLine()); + assertEquals("ACK", in.readLine()); + // test-1 was left when test-2 started, and test-2 never began. + send(out, "END test-set-0/test-1"); + assertEquals("IGNORED test-set-0/test-1", in.readLine()); + assertEquals("ACK", in.readLine()); + send(out, "END test-set-0/test-2"); + assertEquals("IGNORED test-set-0/test-2", in.readLine()); + assertEquals("ACK", in.readLine()); + + // JaCoCo is back: the next test is measured as before. + send(out, "START test-set-0/test-3"); + assertEquals("ACK", in.readLine()); + send(out, "END test-set-0/test-3"); + String cov = in.readLine(); + assertTrue("a COV for test-3, got " + cov, + cov != null && cov.startsWith("COV ") && cov.contains("test-set-0/test-3")); + assertEquals("ACK", in.readLine()); + } + } + + @Test + public void aStartThatThrowsNamesTheTestThenAnswersAckAndTheConnectionGoesOn() throws Exception { + // A port no socket can have: the reset throws IllegalArgumentException. + System.setProperty(JACOCO_PORT_PROPERTY, "70000"); + assertTrue(KeployDedupAgent.start()); + try (Socket sdk = collector.accept()) { + sdk.setSoTimeout(30000); + OutputStream out = sdk.getOutputStream(); + BufferedReader in = new BufferedReader( + new InputStreamReader(sdk.getInputStream(), StandardCharsets.UTF_8)); + + send(out, "START test-set-0/test-1"); + assertEquals("IGNORED test-set-0/test-1", in.readLine()); + assertEquals("ACK", in.readLine()); + send(out, "END test-set-0/test-1"); + assertEquals("the connection still answers", "IGNORED test-set-0/test-1", in.readLine()); + assertEquals("ACK", in.readLine()); + } + } + + private static void send(OutputStream out, String command) throws IOException { + out.write((command + "\n").getBytes(StandardCharsets.UTF_8)); + out.flush(); + } } diff --git a/keploy-sdk/src/test/java/io/keploy/dedup/FakeJacoco.java b/keploy-sdk/src/test/java/io/keploy/dedup/FakeJacoco.java new file mode 100644 index 0000000..dfdd28e --- /dev/null +++ b/keploy-sdk/src/test/java/io/keploy/dedup/FakeJacoco.java @@ -0,0 +1,67 @@ +package io.keploy.dedup; + +import java.io.Closeable; +import java.io.IOException; +import java.net.InetAddress; +import java.net.ServerSocket; +import java.net.Socket; +import java.util.concurrent.atomic.AtomicInteger; +import org.jacoco.core.runtime.RemoteControlReader; +import org.jacoco.core.runtime.RemoteControlWriter; + +/** + * A JaCoCo tcpserver: it answers each dump or reset with an empty dump. + * dropNext closes that many connections first with no answer, as a JaCoCo + * that went away does. + */ +final class FakeJacoco implements Closeable { + + final ServerSocket server; + final AtomicInteger dropNext = new AtomicInteger(); + final AtomicInteger served = new AtomicInteger(); + final AtomicInteger accepted = new AtomicInteger(); + // holdMillis holds each answer that long, as a JaCoCo busy with a big dump does. + volatile long holdMillis; + + FakeJacoco() throws IOException { + server = new ServerSocket(0, 50, InetAddress.getLoopbackAddress()); + Thread thread = new Thread(this::serve, "fake-jacoco"); + thread.setDaemon(true); + thread.start(); + } + + private void serve() { + while (!server.isClosed()) { + try (Socket socket = server.accept()) { + accepted.incrementAndGet(); + socket.setSoTimeout(30000); + if (dropNext.getAndUpdate(n -> n > 0 ? n - 1 : 0) > 0) { + continue; + } + RemoteControlWriter writer = new RemoteControlWriter(socket.getOutputStream()); + RemoteControlReader reader = new RemoteControlReader(socket.getInputStream()); + reader.setRemoteCommandVisitor((dump, reset) -> { + hold(); + served.incrementAndGet(); + writer.sendCmdOk(); + }); + reader.read(); + } catch (IOException e) { + // Closed, or the client went away. + } + } + } + + private void hold() { + try { + Thread.sleep(holdMillis); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + + @Override + public void close() throws IOException { + server.close(); + } +} From bcef6abc071fff684264c840c6fea12765b64fe1 Mon Sep 17 00:00:00 2001 From: slayerjain Date: Sun, 8 Nov 2026 16:34:22 +0530 Subject: [PATCH 3/3] fix(dedup): leave another server's socket file; IGNORED on a failed dump junixsocket's delete on close deletes the socket file by path. When another server had bound the control socket's path since (the file deleted and bound again), stop() deleted that server's file. The SDK now turns it off and deletes the file once its socket is closed, only while the file is still the one it bound (the same device and inode). On the TCP transport, an END whose JaCoCo dump failed sent an empty COV, the frame of a test that ran no line of the app, so the collector could not tell the two apart. It is now answered "IGNORED " and then ACK, as on the unix socket, and it ends the test. stop() waits for the worker's thread through one interface for both transports. When its wait is over with the thread still running, it closes the socket, says so at WARNING and returns, where it waited a second time before: a start() waits for that thread in turn. bind() closes the socket it opened when it fails, so close() is only ever called on a listening socket: it wakes the accept, and closes the socket from the caller's thread only when the wake-up cannot connect. The tests shorten the waits. New rows: stop() during a command leaves the socket open to its own thread; stop() after another server took the path ends the thread and leaves that server's file; stop() after the socket file was deleted ends at once; start() while the stopped worker still runs a command refuses, and binds once that worker has ended; a TCP END whose dump failed. The deleted-file row now also checks that it is said once. Signed-off-by: slayerjain --- .../io/keploy/dedup/KeployDedupAgent.java | 170 ++++++++++-------- .../io/keploy/dedup/ControlSocketTest.java | 160 ++++++++++++++++- .../keploy/dedup/CoverageTcpClientTest.java | 35 ++++ 3 files changed, 291 insertions(+), 74 deletions(-) diff --git a/keploy-sdk/src/main/java/io/keploy/dedup/KeployDedupAgent.java b/keploy-sdk/src/main/java/io/keploy/dedup/KeployDedupAgent.java index 19b992e..035e3fa 100644 --- a/keploy-sdk/src/main/java/io/keploy/dedup/KeployDedupAgent.java +++ b/keploy-sdk/src/main/java/io/keploy/dedup/KeployDedupAgent.java @@ -28,7 +28,9 @@ import java.net.Socket; import java.nio.charset.StandardCharsets; import java.nio.file.Files; +import java.nio.file.LinkOption; import java.nio.file.Path; +import java.nio.file.attribute.BasicFileAttributes; import java.util.ArrayList; import java.util.Collections; import java.util.Comparator; @@ -38,6 +40,7 @@ import java.util.List; import java.util.Locale; import java.util.Map; +import java.util.Objects; import java.util.Set; import java.util.concurrent.atomic.AtomicBoolean; import java.util.jar.JarEntry; @@ -113,15 +116,17 @@ static String encodeBytecodeFrame(String buildTag, String manifestJson, byte[] z private static final AtomicBoolean STARTED = new AtomicBoolean(false); private static final AtomicBoolean SHUTDOWN_HOOK_REGISTERED = new AtomicBoolean(false); // The active transport worker: a CommandServer (unix, local/docker) or a - // CoverageTcpClient (TCP, k8s). Both are Closeable so stop() is transport-agnostic. - private static volatile Closeable coverageWorker; + // CoverageTcpClient (TCP, k8s), so stop() is transport-agnostic. + private static volatile CoverageWorker coverageWorker; // The worker's thread. stop() waits for it to end, and start() does too // when stop() gave up waiting (a command still running): until it ends it // owns its socket, and a second worker on the same path would lose its - // socket file to the first one's close. - private static volatile Thread coverageThread; + // socket file to the first one's close. Before the first start() it is a + // thread never started, which join() does not wait for. + private static volatile Thread coverageThread = new Thread("keploy-java-dedup-none"); // How long stop(), and a start() after it, wait for the worker's thread. - private static final long WORKER_STOP_WAIT_MILLIS = 5000; + // Not final: the tests shorten it. + static volatile long workerStopWaitMillis = 5000; private KeployDedupAgent() { } @@ -168,7 +173,7 @@ public static boolean start() { new JacocoClient(resolveHost(), resolvePort()), new CoverageIndex()); - Runnable worker; + CoverageWorker worker; String threadName; String endpoint = resolveEndpoint(); if (endpoint != null) { @@ -214,7 +219,7 @@ public static boolean start() { server.bind(); } catch (Throwable t) { // Never let premain fail the app's start: run it without dedup. - server.close(); + // bind() has closed the socket it opened. STARTED.set(false); log(Level.SEVERE, "Java dedup control socket server is unavailable", t); return false; @@ -245,53 +250,39 @@ public static boolean isStarted() { * Stops the background control socket listener. */ public static void stop() { - Closeable worker = coverageWorker; - if (worker != null) { - try { - worker.close(); - } catch (IOException e) { - log(Level.FINE, "Failed to close Java dedup coverage worker", e); - } - } + CoverageWorker worker = coverageWorker; coverageWorker = null; - // Wait for the worker to end, so that a start() after this one does - // not overlap it. Still running after the wait, it is in a command or - // in an accept close()'s wake-up did not reach (the socket file was - // bound again by someone else). Either way it is past the point where - // closing its socket from here races its accept, so close it: an - // accept ends, and a command ends on its own once it is answered. - if (!awaitWorkerThread() && worker instanceof CommandServer) { - ((CommandServer) worker).closeSocket(); + if (worker != null) { + worker.close(); + // Wait for the worker to end, so that a start() after this one + // does not overlap it. Still running after the wait, it is in a + // command or in an accept close()'s wake-up did not reach (the + // socket file was bound again by someone else). Either way it is + // past the point where closing its socket from here races its + // accept, so close it: an accept ends, and a command ends on its + // own once it is answered. A start() waits for it too. if (!awaitWorkerThread()) { - log(Level.WARNING, "Java dedup worker is still running a command; it stops once the command ends", - null); + worker.closeNow(); + log(Level.WARNING, "Java dedup worker did not stop within " + workerStopWaitMillis + + " ms (a command still running, or its socket file bound again by another process);" + + " its socket is closed, and it stops once its command ends", null); } } - if (!isAlive(coverageThread)) { - coverageThread = null; - } STARTED.set(false); } - // awaitWorkerThread waits up to WORKER_STOP_WAIT_MILLIS for the last + // awaitWorkerThread waits up to workerStopWaitMillis for the last // worker's thread to end and says whether it has. private static boolean awaitWorkerThread() { Thread thread = coverageThread; - if (thread == null || thread == Thread.currentThread()) { - return true; - } try { - thread.join(WORKER_STOP_WAIT_MILLIS); + thread.join(workerStopWaitMillis); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } return !thread.isAlive(); } - private static boolean isAlive(Thread thread) { - return thread != null && thread.isAlive(); - } - private static boolean isDisabled() { return isTruthy(System.getenv("KEPLOY_JAVA_DEDUP_DISABLED")) || isTruthy(System.getProperty("keploy.java.dedup.disabled")); @@ -376,6 +367,17 @@ private static final class ManifestEntry { private int probeCount; } + // fileKey is the file's identity (its device and inode on Linux and + // macOS), or null when it is gone or the platform has none. + private static Object fileKey(File file) { + try { + return Files.readAttributes(file.toPath(), BasicFileAttributes.class, LinkOption.NOFOLLOW_LINKS) + .fileKey(); + } catch (IOException e) { + return null; + } + } + private static void deleteSocketFile(File file) { if (file.exists() && !file.delete()) { log(Level.FINE, "Failed to delete socket file " + file.getAbsolutePath(), null); @@ -412,7 +414,18 @@ private static void diagnostic(String message) { LOGGER.log(Level.INFO, formatted); } - private static final class CommandServer implements Runnable, Closeable { + // A transport worker: run() serves Keploy's commands until close(). + private interface CoverageWorker extends Runnable, Closeable { + // close asks run() to end; it does not wait for it. + @Override + void close(); + + // closeNow closes the worker's socket from the caller's thread, for + // stop() once run() has outlived close(). + void closeNow(); + } + + private static final class CommandServer implements CoverageWorker { private final CoverageCollector collector; private final CoveragePublisher publisher; @@ -420,8 +433,9 @@ private static final class CommandServer implements Runnable, Closeable { private final AtomicBoolean running = new AtomicBoolean(true); private final Object testCaseLock = new Object(); private volatile AFUNIXServerSocket serverSocket; - // Set once bind() has the socket listening, as start() then runs run(). - private volatile boolean listening; + // The socket file's identity once bound (fileKey), so that run() + // deletes the file only while it is still this socket's. + private volatile Object socketFileKey; private String activeTestId = ""; CommandServer(CoverageCollector collector, CoveragePublisher publisher, File controlSocket) { @@ -433,13 +447,22 @@ private static final class CommandServer implements Runnable, Closeable { // bind replaces a stale socket file and listens on the control socket. // start() calls it before it returns, so a command sent once start() // has returned waits in the backlog for run() instead of being refused. - // On a failure the caller closes this server, which closes the socket. + // On a failure it closes the socket it opened, as no run() will. void bind() throws IOException { deleteSocketFile(controlSocket); - serverSocket = AFUNIXServerSocket.newInstance(); - serverSocket.bind(AFUNIXSocketAddress.of(controlSocket), SOCKET_BACKLOG); - listening = true; - relaxSocketPermissions(controlSocket); + AFUNIXServerSocket socket = AFUNIXServerSocket.newInstance(); + // Not junixsocket's delete on close: it deletes by path, so it + // deleted the file of a server that had bound the path since. + socket.setDeleteOnClose(false); + try { + socket.bind(AFUNIXSocketAddress.of(controlSocket), SOCKET_BACKLOG); + socketFileKey = fileKey(controlSocket); + relaxSocketPermissions(controlSocket); + } catch (Throwable t) { + closeQuietly(socket); + throw t; + } + serverSocket = socket; } // run accepts commands until close(). Only this thread closes the @@ -474,20 +497,21 @@ public void run() { log(Level.SEVERE, "Failed to accept Java dedup coverage command", e); continue; } - if (!running.get()) { - // close()'s wake-up, or a command that came with it. - closeQuietly(accepted); - break; - } + // close()'s wake-up is a connection with nothing sent: + // handle() answers none, and the loop ends. handle(accepted); } } catch (Throwable t) { STARTED.set(false); log(Level.SEVERE, "Java dedup control socket server is unavailable", t); } - // The close above deleted the socket file, as it is still this - // server's: a start() after stop() binds only once this thread - // has ended (awaitWorkerThread). + // The socket file goes once the socket is closed, only if it is + // still this socket's: deleted or bound again by someone else, it + // is theirs. A start() after stop() binds only once this thread + // has ended (awaitWorkerThread), so it is never the next server's. + if (Objects.equals(socketFileKey, fileKey(controlSocket))) { + deleteSocketFile(controlSocket); + } } // handle reads one command and answers it before the close: ACK if it @@ -580,33 +604,21 @@ private void writeReply(OutputStream outputStream, String reply) { // wakes run()'s accept with a connection of its own. Only when that // connection cannot be made (the file is gone, so no command can come // either) does it close the socket from here, as nothing else can - // wake that accept. A socket not listening (bind failed) is closed here. + // wake that accept. @Override public void close() { running.set(false); - AFUNIXServerSocket localServer = serverSocket; - if (localServer == null || localServer.isClosed()) { - return; - } - if (!listening) { - closeQuietly(localServer); - return; - } try (AFUNIXSocket wake = AFUNIXSocket.newInstance()) { wake.connect(AFUNIXSocketAddress.of(controlSocket), SOCKET_TIMEOUT_MILLIS); } catch (IOException e) { log(Level.FINE, "Could not wake the Java dedup control socket; closing it", e); - closeQuietly(localServer); + closeNow(); } } - // closeSocket closes the socket from here, for stop() once run() has - // outlived close()'s wake-up. - void closeSocket() { - AFUNIXServerSocket localServer = serverSocket; - if (localServer != null) { - closeQuietly(localServer); - } + @Override + public void closeNow() { + closeQuietly(serverSocket); } } @@ -627,12 +639,13 @@ private static void closeQuietly(Closeable closeable) { * SDK -> collector : "ACK" (after START reset) * "COV <compact-json>" + "ACK" (after END dump) * "IGNORED <id>" + "ACK" (after a START whose reset - * failed, or an END with no START) + * failed, an END with no START, + * or an END whose dump failed) * * The collector starts listening only when replay begins, so the connect loop * retries until it is reachable. */ - private static final class CoverageTcpClient implements Runnable, Closeable { + private static final class CoverageTcpClient implements CoverageWorker { private static final long RECONNECT_DELAY_MILLIS = 1000; @@ -765,8 +778,13 @@ private void dispatch(CoverageCommand command, OutputStream out) throws IOExcept try { executedProbesByClass = collector.capture(); } catch (Exception e) { + // Not taken, as on the unix socket: an empty COV is + // the frame of a test that ran no line of the app. log(Level.SEVERE, "Failed to collect Java coverage for " + command.testId, e); - executedProbesByClass = Collections.emptyMap(); + activeTestId = ""; + writeLine(out, IGNORED_REPLY + " " + command.testId); + writeLine(out, "ACK"); + return; } if (executedProbesByClass.isEmpty()) { log(Level.FINE, "No Java coverage collected for " + command.testId, null); @@ -812,6 +830,12 @@ public void close() { } } } + // closeNow is close(): it already closes the socket from the caller's + // thread, which no accept races here. + @Override + public void closeNow() { + close(); + } } private enum CommandAction { diff --git a/keploy-sdk/src/test/java/io/keploy/dedup/ControlSocketTest.java b/keploy-sdk/src/test/java/io/keploy/dedup/ControlSocketTest.java index 44acdc3..9ebdedb 100644 --- a/keploy-sdk/src/test/java/io/keploy/dedup/ControlSocketTest.java +++ b/keploy-sdk/src/test/java/io/keploy/dedup/ControlSocketTest.java @@ -17,9 +17,15 @@ import java.nio.charset.StandardCharsets; import java.nio.file.Files; import java.nio.file.attribute.PosixFilePermission; +import java.util.List; import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; +import java.util.logging.Handler; +import java.util.logging.Level; +import java.util.logging.LogRecord; +import java.util.logging.Logger; import org.junit.After; import org.junit.Before; import org.junit.Test; @@ -52,9 +58,30 @@ public class ControlSocketTest { private File dir; private FakeJacoco jacoco; private DataReceiver data; + // What the agent logged at SEVERE during the test: the first 20 of them, + // so that a server looping on a closed socket fails a row, not the JVM. + private final List severe = new CopyOnWriteArrayList<>(); + private final Handler severeLog = new Handler() { + @Override + public void publish(LogRecord record) { + if (record.getLevel().intValue() >= Level.SEVERE.intValue() && severe.size() < 20) { + severe.add(record.getMessage()); + } + } + + @Override + public void flush() { + } + + @Override + public void close() { + } + }; + private AFUNIXServerSocket foreign; @Before public void setUp() throws IOException { + Logger.getLogger(KeployDedupAgent.class.getName()).addHandler(severeLog); savedControlPath = KeployDedupAgent.controlSocketPath; savedDataPath = KeployDedupAgent.dataSocketPath; dir = Files.createTempDirectory("keploy-dedup-control").toFile(); @@ -67,6 +94,12 @@ public void setUp() throws IOException { @After public void tearDown() throws Exception { KeployDedupAgent.stop(); + KeployDedupAgent.workerStopWaitMillis = 5000; + Logger.getLogger(KeployDedupAgent.class.getName()).removeHandler(severeLog); + Logger.getLogger(KeployDedupAgent.class.getName()).setUseParentHandlers(true); + if (foreign != null) { + foreign.close(); + } KeployDedupAgent.controlSocketPath = savedControlPath; KeployDedupAgent.dataSocketPath = savedDataPath; System.clearProperty(JACOCO_PORT_PROPERTY); @@ -273,17 +306,142 @@ public void aSocketFileDeletedUnderTheServerEndsItOnceAndSaysSo() throws Excepti // Deleted while the START runs: the next accept finds the file gone, // and junixsocket closes the socket. assertTrue(new File(KeployDedupAgent.controlSocketPath).delete()); + // A server that loops on the closed socket logs without end: kept + // off the console, so that it fails this row rather than the run. + Logger.getLogger(KeployDedupAgent.class.getName()).setUseParentHandlers(false); assertEquals(TAKEN, answer.poll(30, TimeUnit.SECONDS)); - while ((KeployDedupAgent.isStarted() || controlThreadAlive()) && System.nanoTime() < deadline) { + long ends = System.nanoTime() + TimeUnit.SECONDS.toNanos(5); + while ((KeployDedupAgent.isStarted() || controlThreadAlive()) && System.nanoTime() < ends) { Thread.sleep(5); } // Before, the accept loop went on on the closed socket: it spun a core // and logged without end, and the agent still said it was started. assertFalse("the agent says it is no longer started", KeployDedupAgent.isStarted()); assertFalse("the control thread ended", controlThreadAlive()); + assertEquals("said once, and nothing else at SEVERE: " + severe, 1, severe.size()); + assertTrue(severe.get(0), severe.get(0).contains("was closed: its file was deleted or replaced")); + client.join(); + } + + @Test(timeout = 60000) + public void aStopDuringACommandLeavesTheSocketToItsOwnThread() throws Exception { + startWithJacoco(); + jacoco.holdMillis = 1500; + BlockingQueue answer = new LinkedBlockingQueue<>(); + Thread client = new Thread(() -> { + try { + answer.add(send("START test-set-0/test-1")); + } catch (IOException e) { + answer.add(e.toString()); + } + }); + client.start(); + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(30); + while (jacocoConnections() == 0 && System.nanoTime() < deadline) { + Thread.sleep(5); + } + Thread stopper = new Thread(KeployDedupAgent::stop); + stopper.start(); + Thread.sleep(300); + // stop() is waiting for the START. The socket is still open, as only + // its own thread closes it: closed from here, junixsocket's accept on + // that thread could close whatever socket took its fd number next. + assertTrue("stop() is still waiting for the command", stopper.isAlive()); + File socket = new File(KeployDedupAgent.controlSocketPath); + assertTrue("the socket file is still there", socket.exists()); + assertConnects("the socket still listens", socket); + stopper.join(); + assertEquals("the START was answered", TAKEN, answer.poll(30, TimeUnit.SECONDS)); + assertFalse("the control thread ended", controlThreadAlive()); + assertFalse("its own thread deleted the socket file", socket.exists()); + assertEquals("nothing at SEVERE: " + severe, 0, severe.size()); + client.join(); + } + + @Test(timeout = 60000) + public void aStopAfterAnotherServerTookThePathEndsTheThreadAndLeavesTheirSocket() throws Exception { + KeployDedupAgent.workerStopWaitMillis = 300; + assertTrue(KeployDedupAgent.start()); + assertEquals(IGNORED, send("END test-set-0/test-1")); + Thread.sleep(100); + severe.clear(); + // Another server takes the path while the control thread waits in its + // accept: close()'s wake-up reaches that server, not this one. + File socket = new File(KeployDedupAgent.controlSocketPath); + assertTrue(socket.delete()); + foreign = AFUNIXServerSocket.newInstance(); + foreign.bind(AFUNIXSocketAddress.of(socket)); + KeployDedupAgent.stop(); + assertFalse(KeployDedupAgent.isStarted()); + // stop() closed the socket once its wait was over: the thread ends. + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(30); + while (controlThreadAlive() && System.nanoTime() < deadline) { + Thread.sleep(5); + } + assertFalse("the control thread ended", controlThreadAlive()); + assertTrue("the other server's socket file is left alone", socket.exists()); + assertConnects("the other server still listens", socket); + assertEquals("a stop says nothing at SEVERE: " + severe, 0, severe.size()); + } + + @Test(timeout = 60000) + public void aStopAfterTheSocketFileWasDeletedEndsAtOnce() throws Exception { + KeployDedupAgent.workerStopWaitMillis = 20000; + assertTrue(KeployDedupAgent.start()); + assertEquals(IGNORED, send("END test-set-0/test-1")); + Thread.sleep(100); + assertTrue(new File(KeployDedupAgent.controlSocketPath).delete()); + long began = System.nanoTime(); + KeployDedupAgent.stop(); + long tookMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - began); + // No wake-up can reach the accept, so close() closes the socket at + // once; left to stop()'s second try, it took the whole wait. + assertTrue("stop() took " + tookMillis + " ms", tookMillis < 10000); + assertFalse("the control thread ended", controlThreadAlive()); + } + + @Test(timeout = 60000) + public void aStartWhileTheStoppedWorkerStillRunsACommandIsRefused() throws Exception { + KeployDedupAgent.workerStopWaitMillis = 100; + startWithJacoco(); + // Under the SDK's 3 s read timeout, so the START is answered ACK. + jacoco.holdMillis = 2000; + BlockingQueue answer = new LinkedBlockingQueue<>(); + Thread client = new Thread(() -> { + try { + answer.add(send("START test-set-0/test-1")); + } catch (IOException e) { + answer.add(e.toString()); + } + }); + client.start(); + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(30); + while (jacocoConnections() == 0 && System.nanoTime() < deadline) { + Thread.sleep(5); + } + KeployDedupAgent.stop(); + assertTrue("stop() gave up while the START runs", controlThreadAlive()); + // A second server on the path now would lose its socket file when + // the first one's thread ends and closes its own. + assertFalse("start() refuses while the old worker runs", KeployDedupAgent.start()); + assertFalse(KeployDedupAgent.isStarted()); + assertEquals("the START was answered", TAKEN, answer.poll(30, TimeUnit.SECONDS)); + while (controlThreadAlive() && System.nanoTime() < deadline) { + Thread.sleep(5); + } + assertTrue("start() binds once the old worker has ended", KeployDedupAgent.start()); + assertEquals(IGNORED, send("END test-set-0/test-1")); client.join(); } + private static void assertConnects(String what, File socket) { + try (AFUNIXSocket probe = AFUNIXSocket.newInstance()) { + probe.connect(AFUNIXSocketAddress.of(socket), 3000); + } catch (IOException e) { + throw new AssertionError(what + ": " + e, e); + } + } + private int jacocoConnections() { return jacoco.accepted.get(); } diff --git a/keploy-sdk/src/test/java/io/keploy/dedup/CoverageTcpClientTest.java b/keploy-sdk/src/test/java/io/keploy/dedup/CoverageTcpClientTest.java index 38b4a99..1d48f5f 100644 --- a/keploy-sdk/src/test/java/io/keploy/dedup/CoverageTcpClientTest.java +++ b/keploy-sdk/src/test/java/io/keploy/dedup/CoverageTcpClientTest.java @@ -109,6 +109,41 @@ public void aStartWhoseResetFailedNamesTheTestThenAnswersAckAndSoDoesItsEnd() th } } + @Test + public void anEndWhoseDumpFailedNamesTheTestThenAnswersAckAndEndsTheTest() throws Exception { + jacoco = new FakeJacoco(); + System.setProperty(JACOCO_PORT_PROPERTY, String.valueOf(jacoco.server.getLocalPort())); + assertTrue(KeployDedupAgent.start()); + try (Socket sdk = collector.accept()) { + sdk.setSoTimeout(30000); + OutputStream out = sdk.getOutputStream(); + BufferedReader in = new BufferedReader( + new InputStreamReader(sdk.getInputStream(), StandardCharsets.UTF_8)); + + send(out, "START test-set-0/test-1"); + assertEquals("ACK", in.readLine()); + // JaCoCo goes away for the dump: test-1's lines never reach the + // SDK. Before, it sent an empty COV, as for a test that ran no + // line of the app. + jacoco.dropNext.set(1); + send(out, "END test-set-0/test-1"); + assertEquals("IGNORED test-set-0/test-1", in.readLine()); + assertEquals("ACK", in.readLine()); + // That END ended test-1. + send(out, "END test-set-0/test-1"); + assertEquals("IGNORED test-set-0/test-1", in.readLine()); + assertEquals("ACK", in.readLine()); + + send(out, "START test-set-0/test-2"); + assertEquals("ACK", in.readLine()); + send(out, "END test-set-0/test-2"); + String cov = in.readLine(); + assertTrue("a COV for test-2, got " + cov, + cov != null && cov.startsWith("COV ") && cov.contains("test-set-0/test-2")); + assertEquals("ACK", in.readLine()); + } + } + @Test public void aStartThatThrowsNamesTheTestThenAnswersAckAndTheConnectionGoesOn() throws Exception { // A port no socket can have: the reset throws IllegalArgumentException.