test: skip mock IPC tests on CRAN and fix mock server races

r-pkg/tests/testthat/helper-mock-server.R

@@ -11,37 +11,74 @@
 # 3. Responds to metrics_request messages with approximate values
 # 4. Collects all received JSON messages
 # 5. Returns collected messages when the device sends "close"
+#
+# These subprocess/socket integration tests run in CI, but are skipped on CRAN.
+
+# Shared by both transports: wait() does not terminate a process on timeout.
+collect_mock_server = function(bg, timeout, name = "Mock server") {
+  bg$wait(timeout)
+  if (bg$is_alive()) {
+    bg$kill()
+    stop(name, " timed out after ", timeout,
+         " ms waiting for completion; background process terminated")
+  }
 
-start_mock_server_local = function(send_welcome = FALSE, transport = "unix") {
+  status = bg$get_exit_status()
+  read_error = function() {
+    # kill() can close stderr; preserve the exit-status error in that case.
+    tryCatch(bg$read_error(), error = function(e) "stderr unavailable")
+  }
+  if (is.null(status) || is.na(status)) {
+    stop(name, " exited with unknown exit status: ", read_error())
+  }
+  if (status != 0) {
+    stop(name, " exited with error (status ", status, "): ", read_error())
+  }
+  bg$get_result()
+}
+
+start_mock_server_local = function(
+  send_welcome = FALSE,
+  transport = "unix",
+  envir = parent.frame()
+) {
+  skip_on_cran()
   skip_if_not_installed("callr")
   skip_if_not_installed("processx")
   skip_if_not_installed("jsonlite")
 
   is_windows = (.Platform$OS.type == "windows")
+  # Keep paths until the caller finishes, rather than until this helper returns.
+  ready_file = withr::local_tempfile(
+    pattern = "jgd-test-ready-", fileext = ".txt", .local_envir = envir
+  )
 
   if (is_windows) {
     # Windows: named pipe via processx
     pipe_name = sprintf("jgd-test-%d-%s", Sys.getpid(),
-                        basename(tempfile()))
+                        basename(ready_file))
     # processx expects \\?\pipe\NAME (extended-length path prefix);
     # the C client uses \\.\pipe\NAME (device namespace) — both resolve
     # to the same kernel pipe object
     win_path = paste0("\\\\?\\pipe\\", pipe_name)
-    ready_file = tempfile(pattern = "jgd-test-ready-", fileext = ".txt")
   } else {
     socket_path = tempfile(pattern = "jgd-test-", fileext = ".sock")
+    # local_tempfile() uses recursive unlink, which does not remove Unix sockets.
+    withr::defer(unlink(socket_path), envir = envir)
   }
 
   bg = callr::r_bg(
     function(conn_path, ready_file, send_welcome, transport) {
       `%||%` = function(x, y) if (is.null(x)) y else x
       server = processx::conn_create_unix_socket(conn_path)
 
-      # Signal readiness: on Windows write a ready file (pipe has no filesystem
-      # presence); on Unix the socket file itself signals readiness
-      if (!is.null(ready_file)) {
-        writeLines("ready", ready_file)
-      }
+      # On Windows the first poll starts ConnectNamedPipe. Arm it before a
+      # fast client can connect and close, leaving an unaccepted closed pipe.
+      processx::poll(list(server), 0)
+
+      # Signal readiness only after socket creation (including listen) returns.
+      # The Unix socket path can appear before the server is listening.
+      writeLines("ready", ready_file)
 
       # Wait for client connection (30s timeout)
       # poll() returns "connect" (not "ready") for new connections on a
@@ -124,66 +161,61 @@ start_mock_server_local = function(send_welcome = FALSE, transport = "unix") {
     },
     args = list(
       conn_path = if (is_windows) win_path else socket_path,
-      ready_file = if (is_windows) ready_file else NULL,
+      ready_file = ready_file,
       send_welcome = send_welcome,
       transport = transport
     ),
     supervise = TRUE
   )
 
+  # Both Unix sockets and Windows named pipes use an explicit readiness signal.
+  for (i in seq_len(30)) {
+    if (file.exists(ready_file)) break
+    Sys.sleep(0.1)
+  }
+  if (!file.exists(ready_file)) {
+    bg$kill()
+    unlink(ready_file)
+    if (!is_windows) unlink(socket_path)
+    skip("Mock server not ready in time")
+  }
   if (is_windows) {
-    # Wait for the ready file to appear (pipe has no filesystem presence)
-    for (i in seq_len(30)) {
-      if (file.exists(ready_file)) break
-      Sys.sleep(0.1)
-    }
-    if (!file.exists(ready_file)) {
-      bg$kill()
-      skip("Mock server pipe not ready in time")
-    }
     client_uri = paste0("npipe:////./pipe/", pipe_name)
   } else {
-    # Wait for the socket file to appear
-    for (i in seq_len(30)) {
-      if (file.exists(socket_path)) break
-      Sys.sleep(0.1)
-    }
-    if (!file.exists(socket_path)) {
-      bg$kill()
-      skip("Mock server socket not created in time")
-    }
     client_uri = socket_path
   }
 
   list(
     bg = bg,
     socket_path = client_uri,
     collect = function(timeout = 10000) {
-      bg$wait(timeout)
-      if (bg$get_exit_status() != 0) {
-        stop("Mock server exited with error: ", bg$read_error())
-      }
-      bg$get_result()
+      collect_mock_server(bg, timeout)
     },
     cleanup = function() {
       if (bg$is_alive()) {
         bg$kill()
       }
       if (!is_windows) {
         unlink(socket_path)
-      } else {
-        unlink(ready_file)
       }
+      unlink(ready_file)
     }
   )
 }
 
 # TCP mock server using base R sockets (works on all platforms including Windows)
-start_mock_server_tcp = function(send_welcome = FALSE, transport = "tcp") {
+start_mock_server_tcp = function(
+  send_welcome = FALSE,
+  transport = "tcp",
+  envir = parent.frame()
+) {
+  skip_on_cran()
   skip_if_not_installed("callr")
   skip_if_not_installed("jsonlite")
 
-  port_file = tempfile(pattern = "jgd-tcp-port-", fileext = ".txt")
+  port_file = withr::local_tempfile(
+    pattern = "jgd-tcp-port-", fileext = ".txt", .local_envir = envir
+  )
 
   bg = callr::r_bg(
     function(port_file, send_welcome, transport) {
@@ -295,11 +327,7 @@ start_mock_server_tcp = function(send_welcome = FALSE, transport = "tcp") {
     port = port,
     socket_url = sprintf("tcp://127.0.0.1:%d", port),
     collect = function(timeout = 10000) {
-      bg$wait(timeout)
-      if (bg$get_exit_status() != 0) {
-        stop("Mock TCP server exited with error: ", bg$read_error())
-      }
-      bg$get_result()
+      collect_mock_server(bg, timeout, "Mock TCP server")
     },
     cleanup = function() {
       if (bg$is_alive()) {

r-pkg/tests/testthat/test-mock-server.R

@@ -0,0 +1,138 @@
+test_that("mock server entry points skip before starting subprocesses on CRAN", {
+  withr::local_envvar(NOT_CRAN = "false")
+
+  expect_condition(start_mock_server_local(), "On CRAN", class = "skip")
+  expect_condition(start_mock_server_tcp(), "On CRAN", class = "skip")
+  expect_condition(
+    with_mock_jgd(stop("must not run")),
+    "On CRAN",
+    class = "skip"
+  )
+  expect_condition(
+    with_mock_jgd(stop("must not run"), transport = "tcp"),
+    "On CRAN",
+    class = "skip"
+  )
+})
+
+test_that("collection reports unknown and nonzero exit statuses", {
+  # Unknown statuses are rare in real processx processes, so model them here.
+  for (status in list(NULL, NA_integer_, 1L)) {
+    bg = list(
+      wait = function(timeout) NULL,
+      is_alive = function() FALSE,
+      get_exit_status = function() status,
+      read_error = function() "server diagnostic",
+      get_result = function() stop("must not read a failed result")
+    )
+    message = if (is.null(status) || is.na(status)) {
+      "unknown exit status: server diagnostic"
+    } else {
+      "error \\(status 1\\): server diagnostic"
+    }
+    expect_error(collect_mock_server(bg, 1), message)
+  }
+})
+
+test_that("local collection timeout terminates the waiting mock server", {
+  server = start_mock_server_local()
+  withr::defer(server$cleanup())
+
+  expect_error(server$collect(timeout = 1), "Mock server timed out after 1 ms")
+  expect_false(server$bg$is_alive())
+})
+
+test_that("TCP collection timeout terminates the waiting mock server", {
+  server = start_mock_server_tcp()
+  withr::defer(server$cleanup())
+
+  expect_error(
+    server$collect(timeout = 1),
+    "Mock TCP server timed out after 1 ms"
+  )
+  expect_false(server$bg$is_alive())
+})
+
+test_that("collection reports a terminated subprocess for both transports", {
+  lapply(list(start_mock_server_local, start_mock_server_tcp), function(start) {
+    server = start()
+    withr::defer(server$cleanup())
+    server$bg$kill()
+
+    expect_error(server$collect(), "exited with error \\(status")
+  })
+})
+
+test_that("local readiness waits for socket creation and connection polling", {
+  skip_on_cran()
+  skip_if_not_installed("callr")
+  skip_if_not_installed("processx")
+  skip_if_not_installed("jsonlite")
+
+  returned_file = withr::local_tempfile()
+  polled_file = withr::local_tempfile()
+  files = new.env(parent = emptyenv())
+  r_bg = callr::r_bg
+  local_mocked_bindings(
+    r_bg = function(func, args, ...) {
+      files$ready = args$ready_file
+      r_bg(
+        function(func, args, returned_file, polled_file) {
+          create_socket = processx::conn_create_unix_socket
+          # Delay the constructor after the socket/pipe exists. The parent must
+          # still wait until the constructor returns and publishes readiness.
+          assignInNamespace(
+            "conn_create_unix_socket",
+            function(...) {
+              server = create_socket(...)
+              Sys.sleep(0.5)
+              writeLines("returned", returned_file)
+              server
+            },
+            ns = "processx"
+          )
+          poll_connections = processx::poll
+          assignInNamespace(
+            "poll",
+            function(processes, ms) {
+              # Let the client connect and close before the blocking poll.
+              # Windows must already have a pending ConnectNamedPipe request.
+              if (ms == 30000) Sys.sleep(0.5)
+              result = poll_connections(processes, ms)
+              if (ms == 0) writeLines("polled", polled_file)
+              result
+            },
+            ns = "processx"
+          )
+          do.call(func, args)
+        },
+        args = list(
+          func = func, args = args,
+          returned_file = returned_file, polled_file = polled_file
+        ),
+        ...
+      )
+    },
+    .package = "callr"
+  )
+
+  run_server = function() {
+    server = start_mock_server_local()
+    withr::defer(if (server$bg$is_alive()) server$bg$kill())
+    expect_true(file.exists(returned_file))
+    expect_true(file.exists(polled_file))
+    expect_true(file.exists(files$ready))
+
+    jgd(socket = server$socket_path)
+    dev.off()
+    expect_identical(tail(server$collect(), 1)[[1]]$type, "close")
+    server$socket_path
+  }
+  socket_path = run_server()
+  # withr removes the paths when the caller finishes, even without
+  # an explicit server$cleanup() call.
+  expect_false(file.exists(files$ready))
+  if (.Platform$OS.type != "windows") {
+    expect_false(file.exists(socket_path))
+  }
+})