fixup! Test: Bind to port 0

Author: ReinierMaas

libs/opsqueue_python/tests/conftest.py

@@ -92,7 +92,9 @@ def opsqueue_service(
             assert process.poll() is None, "Opsqueue process failed to start"
             try:
                 actual_port = int.from_bytes(
-                    os.read(read_fd, 2), byteorder="big", signed=False
+                    read_exact_fd(read_fd, 2),
+                    byteorder="big",
+                    signed=False,
                 )
                 wrapper = OpsqueueProcess(port=actual_port, process=process)
                 yield wrapper
@@ -105,6 +107,18 @@ def opsqueue_service(
         os.close(read_fd)
 
 
+def read_exact_fd(fd: int, num_bytes: int) -> bytes:
+    data = bytearray()
+    while len(data) < num_bytes:
+        chunk = os.read(fd, num_bytes - len(data))
+        if not chunk:
+            raise EOFError(
+                f"Unexpected EOF: expected {num_bytes} bytes, got {len(data)}: {bytes(data)!r}"
+            )
+        data.extend(chunk)
+    return bytes(data)
+
+
 @contextmanager
 def background_process(
     function: Callable[..., None],

opsqueue/src/config.rs

@@ -23,11 +23,14 @@ pub struct Config {
     #[arg(short, long, default_value_t = 3999)]
     pub port: u16,
 
-    /// Optional file descriptor where the final bound TCP port is written.
+    /// File descriptor where the final bound TCP port is written.
     ///
-    /// This is useful when `--port 0` is used and a parent process wants
-    /// to receive the assigned port without filesystem IO.
-    #[arg(long, value_parser = parse_report_bound_port_fd)]
+    /// This is useful when `--port 0` is used and a parent process wants to receive the assigned
+    /// port without filesystem IO.
+    ///
+    /// The port is written as a 16-bit (u16) big-endian integer to the given file descriptor which
+    /// is then closed. On error the file descriptor is closed without writing anything.
+    #[arg(long, value_parser = parse_report_bound_port_fd, default_value = "-1")]
     pub report_bound_port_pipe: ReportBoundPortPipe,
 
     /// Name of the SQLite database file used by this opsqueue.
@@ -144,6 +147,10 @@ fn parse_report_bound_port_fd(value: &str) -> Result<ReportBoundPortPipe, String
         .parse::<i32>()
         .map_err(|err| format!("invalid file descriptor {value:?}: {err}"))?;
 
+    if fd == -1 {
+        return Ok(ReportBoundPortPipe::default());
+    }
+
     if fd < 0 {
         return Err(format!(
             "invalid file descriptor {value:?}: must be non-negative"

opsqueue/src/server.rs

@@ -68,7 +68,11 @@ pub async fn serve_producer_and_consumer(
     })
     .retry(retry_policy())
     .notify(|e, d| tracing::error!("Error when binding server address: {e:?}, retrying in {d:?}"))
-    .await
+    .await.inspect_err(|_|{
+        // Drop the pipe after the server start retries have been exhausted. This ensures that the
+        // parent process can safely block on reading from the pipe.
+        config.report_bound_port_pipe.take();
+    })
 }
 
 #[cfg(feature = "server-logic")]