Function round_trip_with_timeout

src/extraction_worker.rs:407–460

Sends one extract request and reads the response, killing the worker process and returning [`RoundTripOutcome::Timeout`] if the read takes longer than `timeout`. The worker's child handle is `kill()`ed in place to unblock the read; the caller is expected to `spawn_worker` a fresh subprocess after either a timeout or a crash.

(
    worker: &mut WorkerHandle,
    req: &ExtractRequest,
    timeout: Duration,
)

/// to unblock the read; the caller is expected to `spawn_worker` a fresh
/// subprocess after either a timeout or a crash.
fn round_trip_with_timeout(
    worker: &mut WorkerHandle,
    req: &ExtractRequest,
    timeout: Duration,
) -> RoundTripOutcome {
    let Some(stdin) = worker.stdin.as_mut() else {
        return RoundTripOutcome::Err(io::Error::other("worker stdin already closed"));
    };
    if let Err(e) = write_message(stdin, req).and_then(|()| stdin.flush()) {
        return RoundTripOutcome::Err(e);
    }

    // Split-borrow `stdout` (owned by the read thread) and `child` (owned by
    // the watchdog thread). Rust allows this because they're disjoint fields
    // of `*worker`.
    let WorkerHandle {
        ref mut stdout,
        ref mut child,
        ..
    } = *worker;

    let timed_out = AtomicBool::new(false);
    let (cancel_tx, cancel_rx) = std::sync::mpsc::channel::<()>();

    let read_result: io::Result<ExtractResponse> = std::thread::scope(|s| {
        // `move` the Receiver into the watchdog so it owns it (Receiver
        // is `Send` but not `Sync`). `&timed_out` and `&mut *child` are
        // borrowed from the outer scope under `'scope`.
        let timed_out = &timed_out;
        s.spawn(move || {
            // Watchdog: if the read doesn't finish in `timeout`, kill the
            // child so the read returns EOF and unblocks. The kill failing
            // (child already exited) is fine — we just won't have a clean
            // way to distinguish "crashed at exactly the wrong moment" from
            // "timed out", and that's OK; both cases get respawned.
            if cancel_rx.recv_timeout(timeout).is_err() {
                timed_out.store(true, Ordering::SeqCst);
                let _ = child.kill();
            }
        });
        let r = read_message(stdout);
        let _ = cancel_tx.send(());
        r
    });

    if timed_out.load(Ordering::SeqCst) {
        RoundTripOutcome::Timeout
    } else {
        match read_result {
            Ok(resp) => RoundTripOutcome::Ok(resp),
            Err(e) => RoundTripOutcome::Err(e),
        }
    }
}

fn spawn_worker(self_path: &Path, token: &[u8; TOKEN_LEN]) -> io::Result<WorkerHandle> {
    let token_hex = hex::encode(token);