From ad2874bcff605834b9b72092e0f675be53fcb478 Mon Sep 17 00:00:00 2001 From: Martin Vogel Date: Sat, 18 Jul 2026 15:26:38 +0200 Subject: [PATCH] fix: close remaining daemon coordination races Signed-off-by: Martin Vogel --- scripts/test-windows.ps1 | 21 +++ src/daemon/bootstrap.c | 17 ++- src/daemon/host.c | 31 ++-- src/foundation/subprocess.c | 199 +++++++++++++++++++------- src/foundation/subprocess.h | 23 ++- src/mcp/mcp.c | 4 +- src/pipeline/pass_cross_repo.c | 12 ++ src/pipeline/pipeline_internal.h | 8 ++ tests/test_cross_repo.c | 100 +++---------- tests/test_daemon_bootstrap.c | 37 +++++ tests/test_daemon_frontend.c | 30 +++- tests/test_daemon_ipc.c | 13 +- tests/test_subprocess.c | 180 ++++++++++++++++++++--- tests/test_windows_bundle_contract.sh | 35 +++++ 14 files changed, 515 insertions(+), 195 deletions(-) diff --git a/scripts/test-windows.ps1 b/scripts/test-windows.ps1 index 8fb975f91..ee28f7e95 100644 --- a/scripts/test-windows.ps1 +++ b/scripts/test-windows.ps1 @@ -142,6 +142,27 @@ try { if (-not $userProfile) { throw "could not resolve the current user's profile directory" } $guardRoot = Join-Path $userProfile ("cbm-windows-guards-root-" + [guid]::NewGuid().ToString("N")) New-Item -ItemType Directory -Path $guardRoot | Out-Null + + # GitHub-hosted runner profile children can inherit mutation-capable ACEs + # even though the profile ancestry itself passes the launcher's bounded + # trust policy. Replace that inheritance before creating any executable or + # Python temporary descendant. Use SIDs rather than localized account names. + $currentSid = [System.Security.Principal.WindowsIdentity]::GetCurrent().User + if (-not $currentSid) { throw "could not resolve the current user's SID" } + $guardAcl = [System.Security.AccessControl.DirectorySecurity]::new() + $guardAcl.SetOwner($currentSid) + $guardAcl.SetAccessRuleProtection($true, $false) + $guardRule = [System.Security.AccessControl.FileSystemAccessRule]::new( + $currentSid, + [System.Security.AccessControl.FileSystemRights]::FullControl, + ([System.Security.AccessControl.InheritanceFlags]::ContainerInherit -bor + [System.Security.AccessControl.InheritanceFlags]::ObjectInherit), + [System.Security.AccessControl.PropagationFlags]::None, + [System.Security.AccessControl.AccessControlType]::Allow + ) + $guardAcl.AddAccessRule($guardRule) | Out-Null + Set-Acl -LiteralPath $guardRoot -AclObject $guardAcl + $guardBundle = Join-Path $guardRoot ("cbm-windows-guards-" + [guid]::NewGuid().ToString("N")) New-Item -ItemType Directory -Path $guardBundle | Out-Null $guardBin = Join-Path $guardBundle "codebase-memory-mcp.exe" diff --git a/src/daemon/bootstrap.c b/src/daemon/bootstrap.c index f41eba8fb..8115572d2 100644 --- a/src/daemon/bootstrap.c +++ b/src/daemon/bootstrap.c @@ -445,16 +445,27 @@ cbm_daemon_bootstrap_status_t cbm_daemon_bootstrap_execute_with_ops( } result_out->daemon_spawned = true; - /* Keep startup ownership until the child becomes observable. A - * RESERVED response here belongs to the generation we just launched, - * so it remains a wait state rather than a reason to release/spawn. */ + /* Keep startup ownership only until the child becomes observable. + * Windows participant teardown reacquires the startup transition; + * retaining it while probing an observable generation can deadlock the + * bootstrap against a daemon that is trying to cleanly stand down. */ do { bootstrap_pause(deadline); probe = bootstrap_probe(config, ops, &client, &connect_result); + if (probe == CBM_DAEMON_BOOTSTRAP_PROBE_RESERVED || + probe == CBM_DAEMON_BOOTSTRAP_PROBE_TERMINAL) { + generation_observed = true; + bootstrap_startup_lock_release_complete(ops, &startup_lock); + lock_acquired = false; + break; + } if (!bootstrap_probe_is_waitable(probe)) { break; } } while (cbm_now_ms() < deadline); + if (!lock_acquired) { + continue; + } break; } diff --git a/src/daemon/host.c b/src/daemon/host.c index e7cde9cb0..c19f5519e 100644 --- a/src/daemon/host.c +++ b/src/daemon/host.c @@ -798,21 +798,22 @@ int cbm_daemon_host_run(const cbm_daemon_host_config_t *config) { host_log_close(); return -1; } - cbm_daemon_ipc_lifetime_reservation_t *lifetime_reservation = NULL; - if (host_lifetime_reservation_acquire(config->endpoint, &lifetime_reservation) != 1) { - cbm_log_error("daemon.start_failed", "component", "lifetime"); - host_participant_guard_close(&participant_guard); - host_cohort_close(&cohort_lease, &cohort_manager); - host_log_close(); - return -1; - } cbm_version_cohort_daemon_claim_t *daemon_claim = NULL; if (cbm_version_cohort_daemon_claim_acquire(cohort_manager, &daemon_claim) != CBM_VERSION_COHORT_OK) { cbm_log_error("daemon.start_failed", "component", "claim"); - cbm_daemon_ipc_lifetime_reservation_release(lifetime_reservation); - host_daemon_claim_close(&daemon_claim); host_participant_guard_close(&participant_guard); + host_daemon_claim_close(&daemon_claim); + host_cohort_close(&cohort_lease, &cohort_manager); + host_log_close(); + return -1; + } + cbm_daemon_ipc_lifetime_reservation_t *lifetime_reservation = NULL; + if (host_lifetime_reservation_acquire(config->endpoint, &lifetime_reservation) != 1) { + cbm_log_error("daemon.start_failed", "component", "lifetime"); + cbm_daemon_ipc_lifetime_reservation_release(lifetime_reservation); + host_participant_guard_close(&participant_guard); + host_daemon_claim_close(&daemon_claim); host_cohort_close(&cohort_lease, &cohort_manager); host_log_close(); return -1; @@ -827,8 +828,8 @@ int cbm_daemon_host_run(const cbm_daemon_host_config_t *config) { host_state_free(&host); host_log_close(); cbm_daemon_ipc_lifetime_reservation_release(lifetime_reservation); - host_daemon_claim_close(&daemon_claim); host_participant_guard_close(&participant_guard); + host_daemon_claim_close(&daemon_claim); host_cohort_close(&cohort_lease, &cohort_manager); return -1; } @@ -853,8 +854,8 @@ int cbm_daemon_host_run(const cbm_daemon_host_config_t *config) { cbm_log_error("daemon.start_failed", "component", "runtime"); host_state_free(&host); host_log_close(); - host_daemon_claim_close(&daemon_claim); host_participant_guard_close(&participant_guard); + host_daemon_claim_close(&daemon_claim); host_cohort_close(&cohort_lease, &cohort_manager); return -1; } @@ -870,8 +871,8 @@ int cbm_daemon_host_run(const cbm_daemon_host_config_t *config) { } host_state_free(&host); host_log_close(); - host_daemon_claim_close(&daemon_claim); host_participant_guard_close(&participant_guard); + host_daemon_claim_close(&daemon_claim); host_cohort_close(&cohort_lease, &cohort_manager); return -1; } @@ -914,11 +915,11 @@ int cbm_daemon_host_run(const cbm_daemon_host_config_t *config) { host_log_close(); /* Keep the exact-build lifetime lease through listener/lifetime teardown, * application destruction, diagnostics shutdown, the durable stop record, - * daemon-claim release, and participant release. An activation transaction + * participant release, and daemon-claim release. An activation transaction * may treat exclusive acquisition as proof that coordinated cleanup for * this generation has completed. */ - host_daemon_claim_close(&daemon_claim); host_participant_guard_close(&participant_guard); + host_daemon_claim_close(&daemon_claim); host_cohort_close(&cohort_lease, &cohort_manager); return 0; } diff --git a/src/foundation/subprocess.c b/src/foundation/subprocess.c index 36eb38c9b..6c2c9ad43 100644 --- a/src/foundation/subprocess.c +++ b/src/foundation/subprocess.c @@ -5,7 +5,9 @@ */ #include "subprocess.h" -#include "compat.h" /* cbm_nanosleep */ +#include "compat.h" /* cbm_nanosleep */ +#include "compat_fs.h" +#include "log.h" #include "platform.h" /* cbm_now_ms */ #include @@ -94,16 +96,28 @@ const char *cbm_proc_outcome_str(cbm_proc_outcome_t o) { } } -/* Tail newly-appended complete lines from the child log, starting at *tail_pos. - * A partial (non-newline-terminated) final line is left buffered: *tail_pos is - * not advanced past it, so it is re-read once completed. Returns true if any - * complete line was consumed (i.e. there was progress). */ -static bool cbm_tail_log(const char *log_file, long *tail_pos, cbm_proc_log_cb cb, void *ud) { - if (!log_file) { - return false; +typedef enum { + CBM_TAIL_MORE = 0, + CBM_TAIL_CAUGHT_UP, + CBM_TAIL_ERROR, +} cbm_tail_status_t; + +typedef struct { + cbm_tail_status_t status; + bool progressed; +} cbm_tail_result_t; + +/* Tail one bounded batch from the child log. While the owned tree can still + * write, a partial final line remains buffered. Once the tree is quiescent, + * final=true delivers that last fragment exactly once. */ +static cbm_tail_result_t cbm_tail_log(const char *log_file, long *tail_pos, cbm_proc_log_cb cb, + void *ud, bool final) { + cbm_tail_result_t result = {.status = CBM_TAIL_ERROR, .progressed = false}; + if (!log_file || !tail_pos) { + return result; } #ifdef _WIN32 - FILE *lf = fopen(log_file, "r"); + FILE *lf = cbm_fopen(log_file, "r"); #else int open_flags = O_RDONLY | O_NONBLOCK; #ifdef O_CLOEXEC @@ -114,12 +128,12 @@ static bool cbm_tail_log(const char *log_file, long *tail_pos, cbm_proc_log_cb c #endif int log_fd = open(log_file, open_flags); if (log_fd < 0) { - return false; + return result; } struct stat log_status; if (fstat(log_fd, &log_status) != 0 || !S_ISREG(log_status.st_mode)) { (void)close(log_fd); - return false; + return result; } FILE *lf = fdopen(log_fd, "r"); #endif @@ -127,9 +141,8 @@ static bool cbm_tail_log(const char *log_file, long *tail_pos, cbm_proc_log_cb c #ifndef _WIN32 (void)close(log_fd); #endif - return false; + return result; } - bool progressed = false; if (fseek(lf, *tail_pos, SEEK_SET) == 0) { char line[1024]; size_t delivered_lines = 0; @@ -142,6 +155,7 @@ static bool cbm_tail_log(const char *log_file, long *tail_pos, cbm_proc_log_cb c delivered_bytes < CBM_TAIL_MAX_BYTES_PER_POLL) { long before = ftell(lf); if (!fgets(line, sizeof(line), lf)) { + result.status = ferror(lf) ? CBM_TAIL_ERROR : CBM_TAIL_CAUGHT_UP; break; } size_t l = strlen(line); @@ -150,7 +164,7 @@ static bool cbm_tail_log(const char *log_file, long *tail_pos, cbm_proc_log_cb c if (complete) { line[l - 1] = '\0'; *tail_pos = ftell(lf); - progressed = true; + result.progressed = true; delivered_lines++; if (line[0] && cb) { cb(line, ud); @@ -159,20 +173,37 @@ static bool cbm_tail_log(const char *log_file, long *tail_pos, cbm_proc_log_cb c /* Oversized line filled the buffer without a newline — consume it * anyway (counts as progress) so we never stall on one long line. */ *tail_pos = ftell(lf); - progressed = true; + result.progressed = true; delivered_lines++; if (cb) { cb(line, ud); } - } else { + } else if (!final) { /* Genuine partial final line — keep it buffered for next poll. */ *tail_pos = before; + result.status = CBM_TAIL_MORE; + break; + } else { + /* No writer remains: deliver the final unterminated fragment. */ + *tail_pos = ftell(lf); + result.progressed = true; + if (line[0] && cb) { + cb(line, ud); + } + result.status = CBM_TAIL_CAUGHT_UP; break; } } + if (result.status == CBM_TAIL_ERROR && !ferror(lf)) { + /* Reaching either work cap is conservatively MORE. An exact batch + * boundary needs one empty follow-up poll to prove EOF. */ + result.status = CBM_TAIL_MORE; + } } - fclose(lf); - return progressed; + if (fclose(lf) != 0) { + result.status = CBM_TAIL_ERROR; + } + return result; } /* ── Windows command-line quoting (pure; unit-tested on every platform) ─────── */ @@ -268,6 +299,13 @@ bool cbm_build_win_cmdline(char *buf, size_t cap, const char *const *argv) { enum { CBM_SUBPROCESS_ARGV_LIMIT = 4096 }; +typedef enum { + CBM_SUBPROCESS_ACTIVE = 0, + CBM_SUBPROCESS_CANCEL_REQUESTED, + CBM_SUBPROCESS_DRAINING, + CBM_SUBPROCESS_TERMINAL, +} cbm_subprocess_lifecycle_t; + struct cbm_subprocess { char *bin; char **argv; @@ -282,12 +320,11 @@ struct cbm_subprocess { long tail_pos; uint64_t last_activity_ms; uint64_t termination_started_ms; - atomic_bool cancellation_requested; + atomic_int lifecycle; bool timed_out; bool termination_started; bool force_sent; bool root_reaped; - atomic_bool terminal; uint64_t force_started_ms; bool containment_failed; cbm_proc_result_t result; @@ -379,17 +416,23 @@ static cbm_subprocess_t *cbm_subprocess_copy_opts(const cbm_proc_opts_t *opts) { process->cancel_grace_ms = CBM_SUBPROCESS_MAX_CANCEL_GRACE_MS; } process->delete_log_on_exit = opts->delete_log_on_exit; - atomic_init(&process->cancellation_requested, false); - atomic_init(&process->terminal, false); + atomic_init(&process->lifecycle, CBM_SUBPROCESS_ACTIVE); cbm_subprocess_result_init(&process->result); return process; } -static void cbm_subprocess_poll_log(cbm_subprocess_t *process) { - if (cbm_tail_log(process->log_file, &process->tail_pos, process->on_log_line, - process->log_ud)) { +static cbm_tail_result_t cbm_subprocess_poll_log(cbm_subprocess_t *process, bool final) { + cbm_tail_result_t result = cbm_tail_log(process->log_file, &process->tail_pos, + process->on_log_line, process->log_ud, final); + if (result.progressed) { process->last_activity_ms = cbm_now_ms(); } + return result; +} + +static bool cbm_subprocess_cancellation_requested(const cbm_subprocess_t *process) { + return atomic_load_explicit(&process->lifecycle, memory_order_acquire) == + CBM_SUBPROCESS_CANCEL_REQUESTED; } static void cbm_subprocess_delete_log(cbm_subprocess_t *process) { @@ -407,26 +450,53 @@ static void cbm_subprocess_delete_log(cbm_subprocess_t *process) { #endif } -static cbm_proc_poll_t cbm_subprocess_finish(cbm_subprocess_t *process, cbm_proc_result_t *out) { - /* Capture lines written between the first tail in this poll and process exit. */ - cbm_subprocess_poll_log(process); - process->result.cancellation_requested = - atomic_load_explicit(&process->cancellation_requested, memory_order_acquire); - process->result.tree_quiesced = true; - process->result.supervision_failed = false; - atomic_store_explicit(&process->terminal, true, memory_order_release); - cbm_subprocess_delete_log(process); +static bool cbm_subprocess_begin_terminal_transition(cbm_subprocess_t *process) { + int lifecycle = atomic_load_explicit(&process->lifecycle, memory_order_acquire); + for (;;) { + if (lifecycle == CBM_SUBPROCESS_DRAINING) { + return true; + } + if (lifecycle == CBM_SUBPROCESS_TERMINAL) { + return false; + } + int desired = CBM_SUBPROCESS_DRAINING; + if (atomic_compare_exchange_weak_explicit(&process->lifecycle, &lifecycle, desired, + memory_order_acq_rel, memory_order_acquire)) { + process->result.cancellation_requested = lifecycle == CBM_SUBPROCESS_CANCEL_REQUESTED; + return true; + } + } +} + +static cbm_proc_poll_t cbm_subprocess_publish_terminal(cbm_subprocess_t *process, + cbm_proc_result_t *out, bool delete_log) { + if (delete_log) { + cbm_subprocess_delete_log(process); + } + atomic_store_explicit(&process->lifecycle, CBM_SUBPROCESS_TERMINAL, memory_order_release); if (out) { *out = process->result; } return CBM_PROC_POLL_TERMINAL; } +static cbm_proc_poll_t cbm_subprocess_finish(cbm_subprocess_t *process, cbm_proc_result_t *out) { + if (!cbm_subprocess_begin_terminal_transition(process)) { + return CBM_PROC_POLL_ERROR; + } + process->result.tree_quiesced = true; + process->result.supervision_failed = false; + if (process->log_file && process->on_log_line) { + return CBM_PROC_POLL_RUNNING; + } + return cbm_subprocess_publish_terminal(process, out, true); +} + static cbm_proc_poll_t cbm_subprocess_finish_failed(cbm_subprocess_t *process, cbm_proc_result_t *out) { - cbm_subprocess_poll_log(process); - process->result.cancellation_requested = - atomic_load_explicit(&process->cancellation_requested, memory_order_acquire); + if (!cbm_subprocess_begin_terminal_transition(process)) { + return CBM_PROC_POLL_ERROR; + } process->result.tree_quiesced = false; process->result.supervision_failed = true; process->result.forced = true; @@ -436,11 +506,7 @@ static cbm_proc_poll_t cbm_subprocess_finish_failed(cbm_subprocess_t *process, process->result.outcome = CBM_PROC_KILLED; } process->containment_failed = true; - atomic_store_explicit(&process->terminal, true, memory_order_release); - if (out) { - *out = process->result; - } - return CBM_PROC_POLL_TERMINAL; + return cbm_subprocess_publish_terminal(process, out, false); } #ifdef _WIN32 @@ -634,8 +700,7 @@ static void cbm_win_capture_root(cbm_subprocess_t *process, DWORD code) { if (process->timed_out) { process->result.outcome = CBM_PROC_HANG; } else if (process->root_forced || - (atomic_load_explicit(&process->cancellation_requested, memory_order_acquire) && - code == CBM_WIN_CONTROL_C_EXIT)) { + (cbm_subprocess_cancellation_requested(process) && code == CBM_WIN_CONTROL_C_EXIT)) { process->result.outcome = CBM_PROC_KILLED; } else { process->result.outcome = cbm_proc_classify(true, (int)code, 0, false); @@ -643,7 +708,6 @@ static void cbm_win_capture_root(cbm_subprocess_t *process, DWORD code) { } static cbm_proc_poll_t cbm_subprocess_poll_win(cbm_subprocess_t *process, cbm_proc_result_t *out) { - cbm_subprocess_poll_log(process); uint64_t now = cbm_now_ms(); if (!process->root_reaped) { @@ -652,7 +716,6 @@ static cbm_proc_poll_t cbm_subprocess_poll_win(cbm_subprocess_t *process, cbm_pr DWORD code = 1; (void)GetExitCodeProcess(process->process, &code); cbm_win_capture_root(process, code); - cbm_subprocess_poll_log(process); } else if (waited == WAIT_FAILED) { DWORD code = STILL_ACTIVE; if (GetExitCodeProcess(process->process, &code) && code != STILL_ACTIVE) { @@ -667,7 +730,7 @@ static cbm_proc_poll_t cbm_subprocess_poll_win(cbm_subprocess_t *process, cbm_pr bool job_known = false; bool job_active = cbm_win_job_active(process, &job_known); if (!process->termination_started) { - if (atomic_load_explicit(&process->cancellation_requested, memory_order_acquire)) { + if (cbm_subprocess_cancellation_requested(process)) { cbm_win_begin_termination(process, now); } else if (!process->root_reaped && process->quiet_timeout_ms > 0 && now - process->last_activity_ms >= (uint64_t)process->quiet_timeout_ms) { @@ -888,7 +951,6 @@ static void cbm_posix_capture_root(cbm_subprocess_t *process, int status) { static cbm_proc_poll_t cbm_subprocess_poll_posix(cbm_subprocess_t *process, cbm_proc_result_t *out) { - cbm_subprocess_poll_log(process); uint64_t now = cbm_now_ms(); if (!process->root_reaped) { @@ -896,7 +958,6 @@ static cbm_proc_poll_t cbm_subprocess_poll_posix(cbm_subprocess_t *process, pid_t waited = waitpid(process->pid, &status, WNOHANG); if (waited == process->pid) { cbm_posix_capture_root(process, status); - cbm_subprocess_poll_log(process); } else if (waited < 0 && errno != EINTR) { /* ECHILD means another reaper consumed the status. Other permanent * wait failures are treated the same: retain containment, stop the @@ -911,7 +972,7 @@ static cbm_proc_poll_t cbm_subprocess_poll_posix(cbm_subprocess_t *process, bool group_active = cbm_posix_group_active(process); if (!process->termination_started) { - if (atomic_load_explicit(&process->cancellation_requested, memory_order_acquire)) { + if (cbm_subprocess_cancellation_requested(process)) { cbm_posix_begin_termination(process, now); } else if (!process->root_reaped && process->quiet_timeout_ms > 0 && now - process->last_activity_ms >= (uint64_t)process->quiet_timeout_ms) { @@ -967,12 +1028,27 @@ cbm_proc_poll_t cbm_subprocess_poll(cbm_subprocess_t *process, cbm_proc_result_t if (!process) { return CBM_PROC_POLL_ERROR; } - if (atomic_load_explicit(&process->terminal, memory_order_acquire)) { + int lifecycle = atomic_load_explicit(&process->lifecycle, memory_order_acquire); + if (lifecycle == CBM_SUBPROCESS_TERMINAL) { if (out) { *out = process->result; } return CBM_PROC_POLL_TERMINAL; } + if (lifecycle == CBM_SUBPROCESS_DRAINING) { + cbm_tail_result_t tail = cbm_subprocess_poll_log(process, true); + if (tail.status == CBM_TAIL_MORE) { + return CBM_PROC_POLL_RUNNING; + } + if (tail.status == CBM_TAIL_ERROR) { + cbm_log_error("subprocess.log_drain_failed", "reason", "io_error"); + return cbm_subprocess_publish_terminal(process, out, false); + } + return cbm_subprocess_publish_terminal(process, out, true); + } + /* The one owner-thread tail batch for this public poll. Platform-specific + * reap paths never tail again, preserving the exact per-poll work cap. */ + (void)cbm_subprocess_poll_log(process, false); #ifdef _WIN32 return cbm_subprocess_poll_win(process, out); #else @@ -981,15 +1057,28 @@ cbm_proc_poll_t cbm_subprocess_poll(cbm_subprocess_t *process, cbm_proc_result_t } bool cbm_subprocess_request_cancel(cbm_subprocess_t *process) { - if (!process || atomic_load_explicit(&process->terminal, memory_order_acquire)) { + if (!process) { return false; } - atomic_store_explicit(&process->cancellation_requested, true, memory_order_release); - return true; + int lifecycle = atomic_load_explicit(&process->lifecycle, memory_order_acquire); + for (;;) { + if (lifecycle == CBM_SUBPROCESS_CANCEL_REQUESTED) { + return true; + } + if (lifecycle != CBM_SUBPROCESS_ACTIVE) { + return false; + } + int desired = CBM_SUBPROCESS_CANCEL_REQUESTED; + if (atomic_compare_exchange_weak_explicit(&process->lifecycle, &lifecycle, desired, + memory_order_acq_rel, memory_order_acquire)) { + return true; + } + } } void cbm_subprocess_destroy(cbm_subprocess_t *process) { - if (!process || !atomic_load_explicit(&process->terminal, memory_order_acquire)) { + if (!process || atomic_load_explicit(&process->lifecycle, memory_order_acquire) != + CBM_SUBPROCESS_TERMINAL) { return; } #ifdef _WIN32 diff --git a/src/foundation/subprocess.h b/src/foundation/subprocess.h index b3900b660..df02f1e86 100644 --- a/src/foundation/subprocess.h +++ b/src/foundation/subprocess.h @@ -14,8 +14,8 @@ * tree-sitter scanners that infinite-loop (a hang, not a crash). * * The reap loop is EINTR-safe. Line tailing keeps a partial final line buffered - * (an incomplete, un-newline-terminated line is not yet "progress" and is not - * mis-read as a completed marker). + * while the child tree can still write, then delivers that final fragment once + * the tree is quiescent. */ #ifndef CBM_SUBPROCESS_H #define CBM_SUBPROCESS_H @@ -44,8 +44,12 @@ typedef struct { bool supervision_failed; /* the bounded containment deadline expired; tree_quiesced is false */ } cbm_proc_result_t; -/* Called for each newly-completed (newline-terminated) log line while the child - * runs. A completed line also resets the quiet-timeout (it is progress). */ +/* Called synchronously for each newly-completed log chunk while the child runs. + * Newline-terminated lines are delivered without their newline; oversized lines + * may be split into 1023-byte chunks, and the final unterminated remainder is + * delivered after the tree is quiescent. Each delivered chunk resets the quiet + * timeout. Poll bounds callback work by chunk/byte count, not elapsed time, so + * callbacks must return promptly. */ typedef void (*cbm_proc_log_cb)(const char *line, void *ud); typedef struct { @@ -98,6 +102,12 @@ int cbm_subprocess_spawn(const cbm_proc_opts_t *opts, cbm_subprocess_t **out); * true and tree_quiesced is false; callers must log this as a critical teardown * failure. *out is optional and is not modified for RUNNING or ERROR. * + * Each poll delivers at most 64 log chunks / 64 KiB. On the normal callback + * delivery path, RUNNING may continue after the process tree is quiescent while + * remaining log batches drain, and TERMINAL follows successful catch-up. A + * final-drain I/O error instead terminates without changing process-tree + * classification and does not attempt to delete the log. + * * Poll performs the graceful->force state transition: after explicit cancel (or * quiet-timeout), it requests graceful termination once, then force-terminates the * tree when cancel_grace_ms elapses. Callers must keep polling to make progress. */ @@ -106,8 +116,9 @@ cbm_proc_poll_t cbm_subprocess_poll(cbm_subprocess_t *process, cbm_proc_result_t /* Record an explicit cancellation request without waiting. Safe to repeat and * safe to call from a cancellation thread while one owner thread polls. The * owner must stop cancellation producers before destroying the handle. true - * means the process is live and cancellation is now/already pending; false means - * process is NULL or already terminal. Poll performs signal delivery/escalation. */ + * means the process tree is live and cancellation is now/already pending; false + * means process is NULL, draining terminal logs, or already terminal. Poll + * performs signal delivery/escalation. */ bool cbm_subprocess_request_cancel(cbm_subprocess_t *process); /* Release a terminal handle. This never waits or implicitly cancels; passing a diff --git a/src/mcp/mcp.c b/src/mcp/mcp.c index 8709bb74a..39a8b3708 100644 --- a/src/mcp/mcp.c +++ b/src/mcp/mcp.c @@ -6306,10 +6306,10 @@ static char *handle_trace_call_path(cbm_mcp_server_t *srv, const char *args) { * attached so data_flow args resolve for boundary nodes whose incoming * edge originated on an earlier page. */ cbm_traverse_result_t view_out = tr_out; - view_out.visited += out_start; + view_out.visited = tr_out.visited ? tr_out.visited + out_start : NULL; view_out.visited_count = out_len; cbm_traverse_result_t view_in = tr_in; - view_in.visited += in_start; + view_in.visited = tr_in.visited ? tr_in.visited + in_start : NULL; view_in.visited_count = in_len; /* Totals must count what the caller can actually enumerate: when diff --git a/src/pipeline/pass_cross_repo.c b/src/pipeline/pass_cross_repo.c index 285be212d..c606719d3 100644 --- a/src/pipeline/pass_cross_repo.c +++ b/src/pipeline/pass_cross_repo.c @@ -58,6 +58,15 @@ typedef struct { bool mutated; } cr_run_context_t; +static CBM_TLS cbm_cross_repo_after_insert_test_hook_t cr_after_insert_test_hook = NULL; +static CBM_TLS void *cr_after_insert_test_context = NULL; + +void cbm_cross_repo_set_after_insert_hook_for_tests(cbm_cross_repo_after_insert_test_hook_t hook, + void *context) { + cr_after_insert_test_hook = hook; + cr_after_insert_test_context = context; +} + typedef struct { int count; cr_run_status_t status; @@ -240,6 +249,9 @@ static bool insert_cross_edge(cbm_store_t *store, const char *project, int64_t f bool inserted = cbm_store_insert_edge(store, &edge) > 0; if (inserted) { ctx->mutated = true; + if (cr_after_insert_test_hook) { + cr_after_insert_test_hook(project, edge_type, cr_after_insert_test_context); + } } return inserted; } diff --git a/src/pipeline/pipeline_internal.h b/src/pipeline/pipeline_internal.h index b6ac6575b..e6c211f44 100644 --- a/src/pipeline/pipeline_internal.h +++ b/src/pipeline/pipeline_internal.h @@ -639,6 +639,14 @@ void cbm_pipeline_set_rename_hook_for_tests(cbm_pipeline_t *p, int (*hook)(const char *, const char *, void *), void *ctx); +/* Synchronous thread-local seam for deterministic cross-repo cancellation + * tests. The callback runs immediately after a CROSS_* edge is committed and + * is never retained; it must not re-enter cross-repo matching. */ +typedef void (*cbm_cross_repo_after_insert_test_hook_t)(const char *project, const char *edge_type, + void *context); +void cbm_cross_repo_set_after_insert_hook_for_tests(cbm_cross_repo_after_insert_test_hook_t hook, + void *context); + /* Parse a gRPC stub call "." into the canonical proto * service name + method. Returns true ONLY when a recognized gRPC stub/client * suffix is present (the stub-type signal that gates Route emission, #294). diff --git a/tests/test_cross_repo.c b/tests/test_cross_repo.c index 1150c2a73..084dae9ff 100644 --- a/tests/test_cross_repo.c +++ b/tests/test_cross_repo.c @@ -6,8 +6,8 @@ #include "test_helpers.h" #include "foundation/compat.h" -#include "foundation/compat_thread.h" #include "pipeline/pass_cross_repo.h" +#include "pipeline/pipeline_internal.h" #include #include @@ -382,77 +382,18 @@ TEST(cross_repo_failed_bidirectional_insert_is_not_counted) { PASS(); } -/* Keep the source scan active after the target-b match is published. This - * gives the watcher a deterministic observable hand-off point without adding - * a test hook to the production pass. */ -static bool cross_repo_seed_cancel_scan_tail(const cross_repo_fixture_t *fixture, - const char *source_project) { - enum { CANCEL_SCAN_TOTAL_ROWS = 4096, SEEDED_MATCH_ROWS = 3 }; - char source_path[512]; - if (!cross_repo_project_path(fixture, source_project, source_path, sizeof(source_path))) { - return false; - } - cbm_store_t *source = cbm_store_open_path_existing(source_path); - if (!source) { - return false; - } - bool ok = - sqlite3_exec(cbm_store_get_db(source), "BEGIN IMMEDIATE", NULL, NULL, NULL) == SQLITE_OK; - cbm_node_t caller = {.project = source_project, - .label = "Function", - .name = "cancel_tail_caller", - .qualified_name = "cancel.source.tail-caller", - .file_path = "cancel-tail.c"}; - int64_t caller_id = ok ? cbm_store_upsert_node(source, &caller) : 0; - ok = ok && caller_id > 0; - for (int i = SEEDED_MATCH_ROWS; ok && i < CANCEL_SCAN_TOTAL_ROWS; i++) { - char name[64]; - char qn[96]; - snprintf(name, sizeof(name), "cancel_tail_route_%d", i); - snprintf(qn, sizeof(qn), "%s.cancel-tail.%d", source_project, i); - cbm_node_t route = {.project = source_project, - .label = "Route", - .name = name, - .qualified_name = qn, - .file_path = "cancel-tail.c"}; - int64_t route_id = cbm_store_upsert_node(source, &route); - cbm_edge_t edge = {.project = source_project, - .source_id = caller_id, - .target_id = route_id, - .type = "HTTP_CALLS", - .properties_json = "{}"}; - ok = route_id > 0 && caller_id > 0 && cbm_store_insert_edge(source, &edge) > 0; - } - if (ok) { - ok = sqlite3_exec(cbm_store_get_db(source), "COMMIT", NULL, NULL, NULL) == SQLITE_OK; - } else { - (void)sqlite3_exec(cbm_store_get_db(source), "ROLLBACK", NULL, NULL, NULL); - } - cbm_store_close(source); - return ok; -} - typedef struct { - const cross_repo_fixture_t *fixture; atomic_int *cancelled; - atomic_bool observed_target_write; -} cross_repo_cancel_watcher_t; + int fired; +} cross_repo_cancel_hook_t; -static void *cross_repo_cancel_after_target_write(void *arg) { - enum { CANCEL_POLL_LIMIT = 10000 }; - cross_repo_cancel_watcher_t *watcher = arg; - for (int i = 0; i < CANCEL_POLL_LIMIT; i++) { - int count = cross_repo_count_edges(watcher->fixture, "cancel-target-b", "CROSS_HTTP_CALLS"); - if (count > 0) { - atomic_store_explicit(&watcher->observed_target_write, true, memory_order_release); - atomic_store_explicit(watcher->cancelled, 1, memory_order_release); - return NULL; - } - cbm_usleep(1000); +static void cross_repo_cancel_after_target_write(const char *project, const char *edge_type, + void *opaque) { + cross_repo_cancel_hook_t *hook = opaque; + if (strcmp(project, "cancel-target-b") == 0 && strcmp(edge_type, "CROSS_HTTP_CALLS") == 0) { + hook->fired++; + atomic_store_explicit(hook->cancelled, 1, memory_order_release); } - /* Bound a broken implementation too: the test should fail, not hang. */ - atomic_store_explicit(watcher->cancelled, 1, memory_order_release); - return NULL; } TEST(cross_repo_cancel_mid_run_keeps_completed_target_and_stops_before_later_target) { @@ -461,8 +402,7 @@ TEST(cross_repo_cancel_mid_run_keeps_completed_target_and_stops_before_later_tar cross_repo_fixture_begin(&fixture) && cross_repo_seed_http_pair(&fixture, "cancel-source", "cancel-target-a", "/cancel-a", "a") && cross_repo_seed_http_pair(&fixture, "cancel-source", "cancel-target-b", "/cancel-b", "b") && - cross_repo_seed_http_pair(&fixture, "cancel-source", "cancel-target-c", "/cancel-c", "c") && - cross_repo_seed_cancel_scan_tail(&fixture, "cancel-source"); + cross_repo_seed_http_pair(&fixture, "cancel-source", "cancel-target-c", "/cancel-c", "c"); if (!setup) { cross_repo_fixture_end(&fixture); FAIL("failed to seed cancellation fixture"); @@ -470,23 +410,16 @@ TEST(cross_repo_cancel_mid_run_keeps_completed_target_and_stops_before_later_tar atomic_int cancelled; atomic_init(&cancelled, 0); - cross_repo_cancel_watcher_t watcher = { - .fixture = &fixture, + cross_repo_cancel_hook_t hook = { .cancelled = &cancelled, }; - atomic_init(&watcher.observed_target_write, false); - cbm_thread_t watcher_thread; - bool watcher_started = - cbm_thread_create(&watcher_thread, 0, cross_repo_cancel_after_target_write, &watcher) == 0; const char *targets[] = {"cancel-target-c", "cancel-target-a", "cancel-target-b"}; - cbm_cross_repo_result_t result = {0}; - if (watcher_started) { - result = cbm_cross_repo_match_cancellable("cancel-source", targets, 3, &cancelled); - (void)cbm_thread_join(&watcher_thread); - } + cbm_cross_repo_set_after_insert_hook_for_tests(cross_repo_cancel_after_target_write, &hook); + cbm_cross_repo_result_t result = + cbm_cross_repo_match_cancellable("cancel-source", targets, 3, &cancelled); + cbm_cross_repo_set_after_insert_hook_for_tests(NULL, NULL); - bool observed = atomic_load_explicit(&watcher.observed_target_write, memory_order_acquire); int completed_target_edges = cross_repo_count_edges(&fixture, "cancel-target-a", "CROSS_HTTP_CALLS"); int interrupted_target_edges = @@ -495,8 +428,7 @@ TEST(cross_repo_cancel_mid_run_keeps_completed_target_and_stops_before_later_tar cross_repo_count_edges(&fixture, "cancel-target-c", "CROSS_HTTP_CALLS"); cross_repo_fixture_end(&fixture); - ASSERT_TRUE(watcher_started); - ASSERT_TRUE(observed); + ASSERT_EQ(hook.fired, 1); ASSERT_TRUE(result.cancelled); ASSERT_TRUE(result.partial_results); ASSERT_FALSE(result.failed); diff --git a/tests/test_daemon_bootstrap.c b/tests/test_daemon_bootstrap.c index dbd90ce4f..f95b1900b 100644 --- a/tests/test_daemon_bootstrap.c +++ b/tests/test_daemon_bootstrap.c @@ -44,6 +44,7 @@ typedef struct { atomic_int terminal_probes_remaining; atomic_bool available; atomic_bool connect_after_reserved; + atomic_bool connect_requires_unlocked; cbm_daemon_bootstrap_probe_status_t forced_probe; cbm_version_cohort_status_t forced_cohort; char diagnostic[CBM_DAEMON_CONFLICT_MESSAGE_SIZE]; @@ -140,6 +141,10 @@ static cbm_daemon_bootstrap_probe_status_t bootstrap_fake_probe( if (fake->forced_probe == CBM_DAEMON_BOOTSTRAP_PROBE_TERMINAL) { return fake->forced_probe; } + if (atomic_load(&fake->available) && atomic_load(&fake->connect_requires_unlocked) && + atomic_load(&fake->lock_held) != 0) { + return CBM_DAEMON_BOOTSTRAP_PROBE_RESERVED; + } if (atomic_load(&fake->available)) { result_out->status = CBM_DAEMON_RUNTIME_CONNECT_ACCEPTED; result_out->hello_status = CBM_DAEMON_HELLO_COMPATIBLE; @@ -584,6 +589,37 @@ TEST(daemon_bootstrap_reserved_then_absent_spawns_replacement) { PASS(); } +/* RED for the native Windows lock order: after spawn, the generation claim or + * lifetime reservation becomes visible before the client can connect. Daemon + * participant teardown needs startup ownership, so bootstrap must release its + * handoff once that generation is observable. */ +TEST(daemon_bootstrap_releases_handoff_when_spawned_generation_is_reserved) { + bootstrap_endpoint_fixture_t fixture; + ASSERT_TRUE(bootstrap_endpoint_fixture_start(&fixture, "spawn-admission")); + bootstrap_fake_ops_t fake = {0}; + atomic_store(&fake.connect_requires_unlocked, true); + cbm_daemon_bootstrap_ops_t ops = bootstrap_fake_callbacks(&fake); + cbm_daemon_build_identity_t identity = bootstrap_identity("2.4.0", BOOTSTRAP_BUILD_A); + cbm_daemon_bootstrap_config_t config = { + .role = CBM_DAEMON_PROCESS_MCP_CLIENT, + .endpoint = fixture.endpoint, + .identity = &identity, + .executable_path = "/tmp/cbm", + .connect_timeout_ms = 1, + .startup_timeout_ms = BOOTSTRAP_TEST_TIMEOUT_MS, + }; + cbm_daemon_bootstrap_result_t result; + ASSERT_EQ(cbm_daemon_bootstrap_execute_with_ops(&config, &ops, &result), + CBM_DAEMON_BOOTSTRAP_CONNECTED); + ASSERT_EQ(result.status, CBM_DAEMON_BOOTSTRAP_CONNECTED); + ASSERT_TRUE(result.daemon_spawned); + ASSERT_NOT_NULL(result.client); + ASSERT_EQ(atomic_load(&fake.spawn_count), 1); + ASSERT_EQ(atomic_load(&fake.lock_held), 0); + bootstrap_endpoint_fixture_finish(&fixture); + PASS(); +} + TEST(daemon_bootstrap_rejected_connect_is_reserved_and_never_unavailable) { cbm_daemon_runtime_connect_result_t capacity = {0}; capacity.status = CBM_DAEMON_RUNTIME_CONNECT_REJECTED; @@ -697,6 +733,7 @@ SUITE(daemon_bootstrap) { RUN_TEST(daemon_bootstrap_terminal_then_absent_spawns_replacement); RUN_TEST(daemon_bootstrap_reserved_generation_becomes_connectable_without_spawn); RUN_TEST(daemon_bootstrap_reserved_then_absent_spawns_replacement); + RUN_TEST(daemon_bootstrap_releases_handoff_when_spawned_generation_is_reserved); RUN_TEST(daemon_bootstrap_rejected_connect_is_reserved_and_never_unavailable); RUN_TEST(daemon_bootstrap_concurrent_first_clients_spawn_one_daemon); #ifdef __APPLE__ diff --git a/tests/test_daemon_frontend.c b/tests/test_daemon_frontend.c index c24639be2..2e3f35fac 100644 --- a/tests/test_daemon_frontend.c +++ b/tests/test_daemon_frontend.c @@ -66,8 +66,7 @@ static const char FRONTEND_TEST_CACHE[] = enum { FRONTEND_EOF_TEST_REQUEST_TIMEOUT_MS = 2000, - FRONTEND_EOF_TEST_OVERFLOW_TIMEOUT_S = 4, - FRONTEND_EOF_TEST_DRAIN_TIMEOUT_S = 10, + FRONTEND_EOF_TEST_CATASTROPHIC_TIMEOUT_S = 30, FRONTEND_BACKPRESSURE_MESSAGE_BYTES = 2 * 1024 * 1024, FRONTEND_BACKPRESSURE_FRONTEND_TIMEOUT_S = 12, FRONTEND_BACKPRESSURE_DAEMON_TIMEOUT_S = 20, @@ -427,8 +426,10 @@ static bool frontend_eof_run_isolated(const char *tag, bool overflow) { pid_t child = fork(); if (child == 0) { (void)signal(SIGALRM, SIG_DFL); - (void)alarm(overflow ? FRONTEND_EOF_TEST_OVERFLOW_TIMEOUT_S - : FRONTEND_EOF_TEST_DRAIN_TIMEOUT_S); + /* This alarm bounds a broken isolated harness, not product behavior. + * Valid startup, drain/watchdog, close, and fixture teardown can exceed + * ten seconds under sanitizers while remaining within their own bounds. */ + (void)alarm(FRONTEND_EOF_TEST_CATASTROPHIC_TIMEOUT_S); _exit(frontend_eof_child_run(parent, overflow)); } int status = 0; @@ -437,7 +438,26 @@ static bool frontend_eof_run_isolated(const char *tag, bool overflow) { waited = child > 0 ? waitpid(child, &status, 0) : -1; } while (waited < 0 && errno == EINTR); bool cleaned = th_rmtree(parent) == 0; - return child > 0 && waited == child && WIFEXITED(status) && WEXITSTATUS(status) == 0 && cleaned; + bool child_ok = child > 0 && waited == child && WIFEXITED(status) && WEXITSTATUS(status) == 0; + if (!child_ok) { + if (child <= 0) { + (void)fprintf(stderr, "frontend EOF fixture %s: fork failed\n", tag); + } else if (waited != child) { + (void)fprintf(stderr, "frontend EOF fixture %s: waitpid failed errno=%d\n", tag, errno); + } else if (WIFSIGNALED(status)) { + (void)fprintf(stderr, "frontend EOF fixture %s: child signal=%d\n", tag, + WTERMSIG(status)); + } else if (WIFEXITED(status)) { + (void)fprintf(stderr, "frontend EOF fixture %s: child exit=%d\n", tag, + WEXITSTATUS(status)); + } else { + (void)fprintf(stderr, "frontend EOF fixture %s: unexpected child status\n", tag); + } + } + if (!cleaned) { + (void)fprintf(stderr, "frontend EOF fixture %s: cleanup failed\n", tag); + } + return child_ok && cleaned; } static void frontend_test_release_lease(cbm_version_cohort_lease_t **lease) { diff --git a/tests/test_daemon_ipc.c b/tests/test_daemon_ipc.c index d1a0edadc..045d7202b 100644 --- a/tests/test_daemon_ipc.c +++ b/tests/test_daemon_ipc.c @@ -821,10 +821,15 @@ TEST(daemon_ipc_windows_legacy_bridge_covers_handoff_and_lifetime) { ? cbm_daemon_ipc_lifetime_reservation_try_acquire(endpoint, &lifetime) : -1; int visible_during_lifetime = lifetime ? cbm_daemon_ipc_legacy_generation_probe(endpoint) : -1; - cbm_daemon_ipc_startup_lock_release(&startup); - startup = NULL; cbm_daemon_ipc_lifetime_reservation_release(lifetime); lifetime = NULL; + /* Native Windows teardown is deliberately ordered through startup-v2. + * A participant must remain retryable while a bootstrap handoff owner + * retains that transition; releasing the bootstrap then unblocks it. */ + bool participant_blocked_by_startup = + participant && !cbm_daemon_ipc_participant_guard_release(&participant); + bool participant_retained = participant != NULL; + bool startup_released = cbm_daemon_ipc_startup_lock_release(&startup); bool participant_released = cbm_daemon_ipc_participant_guard_release(&participant); int absent_after_lifetime = endpoint ? cbm_daemon_ipc_legacy_generation_probe(endpoint) : -1; @@ -894,6 +899,10 @@ TEST(daemon_ipc_windows_legacy_bridge_covers_handoff_and_lifetime) { ASSERT_EQ(participant_status, 1); ASSERT_EQ(lifetime_status, 1); ASSERT_EQ(visible_during_lifetime, 1); + ASSERT_TRUE(participant_blocked_by_startup); + ASSERT_TRUE(participant_retained); + ASSERT_TRUE(startup_released); + ASSERT_NULL(startup); ASSERT_EQ(absent_after_lifetime, 0); ASSERT_TRUE(participant_released); ASSERT_NULL(participant); diff --git a/tests/test_subprocess.c b/tests/test_subprocess.c index 4cecbb335..c8cf009cc 100644 --- a/tests/test_subprocess.c +++ b/tests/test_subprocess.c @@ -307,12 +307,42 @@ static int spawn_ignoring_tree(const char *pid_path, int quiet_timeout_ms, int c return cbm_subprocess_spawn(&opts, out); } -static void slow_log_callback(const char *line, void *opaque) { - (void)line; - int *count = opaque; - (*count)++; - const struct timespec delay = {0, 2000000L}; /* 2 ms per delivered line */ - (void)cbm_nanosleep(&delay, NULL); +typedef struct { + cbm_subprocess_t *process; + int count; + bool ordered; + bool late_cancel_attempted; + bool late_cancel_accepted; +} subprocess_log_capture_t; + +static void ordered_log_callback(const char *line, void *opaque) { + subprocess_log_capture_t *capture = opaque; + int index = -1; + char trailing = '\0'; + bool parsed = sscanf(line, "line-%d%c", &index, &trailing) == 1; + capture->ordered = capture->ordered && parsed && index == capture->count; + capture->count++; + if (index == 399) { + capture->late_cancel_attempted = true; + capture->late_cancel_accepted = cbm_subprocess_request_cancel(capture->process); + } +} + +static bool wait_for_log_marker(const char *path, const char *marker, uint64_t deadline_ms) { + char contents[8192]; + do { + FILE *file = fopen(path, "rb"); + if (file) { + size_t used = fread(contents, 1, sizeof(contents) - 1, file); + contents[used] = '\0'; + (void)fclose(file); + if (strstr(contents, marker)) { + return true; + } + } + subprocess_test_pause(); + } while (cbm_now_ms() < deadline_ms); + return false; } #endif /* !_WIN32 */ @@ -521,7 +551,7 @@ TEST(subprocess_cancel_grace_is_hard_capped) { #endif } -TEST(subprocess_poll_has_a_strict_log_delivery_budget) { +TEST(subprocess_poll_log_delivery_is_bounded_and_terminal_is_lossless) { #ifdef _WIN32 SKIP_PLATFORM("POSIX shell log budget probe; native Windows coverage pending"); #else @@ -530,38 +560,141 @@ TEST(subprocess_poll_has_a_strict_log_delivery_budget) { ASSERT_TRUE(log_fd >= 0); (void)close(log_fd); - const char *script = "i=0; while [ $i -lt 400 ]; do echo line-$i; i=$((i+1)); done; sleep 4"; + const char *script = + "i=0; while [ $i -lt 399 ]; do echo line-$i; i=$((i+1)); done; printf line-399"; const char *argv[] = {"/bin/sh", "-c", script, NULL}; - int delivered = 0; + subprocess_log_capture_t capture = {.ordered = true}; cbm_proc_opts_t opts = {0}; opts.bin = "/bin/sh"; opts.argv = argv; opts.log_file = log_path; - opts.on_log_line = slow_log_callback; - opts.log_ud = &delivered; + opts.on_log_line = ordered_log_callback; + opts.log_ud = &capture; opts.cancel_grace_ms = 100; + opts.delete_log_on_exit = true; cbm_subprocess_t *process = NULL; ASSERT_EQ(cbm_subprocess_spawn(&opts, &process), 0); ASSERT_NOT_NULL(process); + capture.process = process; - const struct timespec fill_delay = {0, 150000000L}; - (void)cbm_nanosleep(&fill_delay, NULL); + bool backlog_ready = wait_for_log_marker(log_path, "line-399", cbm_now_ms() + 2000U); cbm_proc_result_t result = {0}; - uint64_t before = cbm_now_ms(); - cbm_proc_poll_t first = cbm_subprocess_poll(process, &result); - uint64_t elapsed = cbm_now_ms() - before; - bool cancel_accepted = cbm_subprocess_request_cancel(process); - bool terminal = poll_until_terminal(process, 3000, &result); + bool terminal = false; + int max_poll_delivery = 0; + uint64_t deadline = cbm_now_ms() + 5000U; + while (backlog_ready && cbm_now_ms() < deadline) { + int before = capture.count; + cbm_proc_poll_t state = cbm_subprocess_poll(process, &result); + int delivered = capture.count - before; + if (delivered > max_poll_delivery) { + max_poll_delivery = delivered; + } + if (state == CBM_PROC_POLL_TERMINAL) { + terminal = true; + break; + } + if (state == CBM_PROC_POLL_ERROR) { + break; + } + subprocess_test_pause(); + } + bool log_deleted = access(log_path, F_OK) != 0 && errno == ENOENT; if (terminal) { cbm_subprocess_destroy(process); + } else { + (void)cbm_subprocess_request_cancel(process); + cbm_proc_result_t cleanup_result; + if (poll_until_terminal(process, 1000, &cleanup_result)) { + cbm_subprocess_destroy(process); + } } (void)unlink(log_path); - ASSERT_EQ(first, CBM_PROC_POLL_RUNNING); - ASSERT_LT(elapsed, 500); - ASSERT_LT(delivered, 400); - ASSERT_TRUE(cancel_accepted); + ASSERT_TRUE(backlog_ready); ASSERT_TRUE(terminal); + ASSERT_EQ(max_poll_delivery, 64); + ASSERT_EQ(capture.count, 400); + ASSERT_TRUE(capture.ordered); + ASSERT_TRUE(capture.late_cancel_attempted); + ASSERT_FALSE(capture.late_cancel_accepted); + ASSERT_FALSE(result.cancellation_requested); + ASSERT_TRUE(log_deleted); + PASS(); +#endif +} + +TEST(subprocess_final_log_drain_error_is_terminal_and_preserves_classification) { +#ifdef _WIN32 + SKIP_PLATFORM("POSIX log-rename final-drain probe; UTF-8 Windows open uses cbm_fopen"); +#else + char log_path[] = "/tmp/cbm-subprocess-log-drain-XXXXXX"; + int log_fd = cbm_mkstemp(log_path); + ASSERT_TRUE(log_fd >= 0); + (void)close(log_fd); + char saved_path[sizeof(log_path) + 16]; + int saved_written = snprintf(saved_path, sizeof(saved_path), "%s.saved", log_path); + ASSERT_TRUE(saved_written > 0 && (size_t)saved_written < sizeof(saved_path)); + + const char *script = "i=0; while [ $i -lt 130 ]; do echo line-$i; i=$((i+1)); done"; + const char *argv[] = {"/bin/sh", "-c", script, NULL}; + subprocess_log_capture_t capture = {.ordered = true}; + cbm_proc_opts_t opts = {0}; + opts.bin = "/bin/sh"; + opts.argv = argv; + opts.log_file = log_path; + opts.on_log_line = ordered_log_callback; + opts.log_ud = &capture; + opts.cancel_grace_ms = 100; + opts.delete_log_on_exit = true; + cbm_subprocess_t *process = NULL; + ASSERT_EQ(cbm_subprocess_spawn(&opts, &process), 0); + ASSERT_NOT_NULL(process); + capture.process = process; + + bool backlog_ready = wait_for_log_marker(log_path, "line-129", cbm_now_ms() + 2000U); + cbm_proc_result_t result = {0}; + cbm_proc_poll_t first = + backlog_ready ? cbm_subprocess_poll(process, &result) : CBM_PROC_POLL_ERROR; + int first_delivery = capture.count; + bool moved = first == CBM_PROC_POLL_RUNNING && rename(log_path, saved_path) == 0; + bool terminal = moved && poll_until_terminal(process, 2000, &result); + int callbacks_at_terminal = capture.count; + cbm_proc_result_t cached = {0}; + cbm_proc_poll_t cached_state = + terminal ? cbm_subprocess_poll(process, &cached) : CBM_PROC_POLL_ERROR; + bool callbacks_stable = capture.count == callbacks_at_terminal; + bool saved_preserved = access(saved_path, F_OK) == 0; + bool original_absent = access(log_path, F_OK) != 0 && errno == ENOENT; + if (terminal) { + cbm_subprocess_destroy(process); + } else { + (void)cbm_subprocess_request_cancel(process); + cbm_proc_result_t cleanup_result; + if (poll_until_terminal(process, 1000, &cleanup_result)) { + cbm_subprocess_destroy(process); + } + } + (void)unlink(log_path); + (void)unlink(saved_path); + + ASSERT_TRUE(backlog_ready); + ASSERT_EQ(first, CBM_PROC_POLL_RUNNING); + ASSERT_EQ(first_delivery, 64); + ASSERT_TRUE(moved); + ASSERT_TRUE(terminal); + ASSERT_TRUE(capture.ordered); + ASSERT_EQ(callbacks_at_terminal, 64); + ASSERT_EQ(cached_state, CBM_PROC_POLL_TERMINAL); + ASSERT_TRUE(callbacks_stable); + ASSERT_EQ(result.outcome, CBM_PROC_CLEAN); + ASSERT_EQ(result.exit_code, 0); + ASSERT_TRUE(result.tree_quiesced); + ASSERT_FALSE(result.supervision_failed); + ASSERT_EQ(cached.outcome, result.outcome); + ASSERT_EQ(cached.exit_code, result.exit_code); + ASSERT_EQ(cached.tree_quiesced, result.tree_quiesced); + ASSERT_TRUE(saved_preserved); + ASSERT_TRUE(original_absent); PASS(); #endif } @@ -823,7 +956,8 @@ SUITE(subprocess) { RUN_TEST(subprocess_cancel_is_idempotent_and_kills_ignoring_tree); RUN_TEST(subprocess_quiet_timeout_kills_ignoring_tree); RUN_TEST(subprocess_cancel_grace_is_hard_capped); - RUN_TEST(subprocess_poll_has_a_strict_log_delivery_budget); + RUN_TEST(subprocess_poll_log_delivery_is_bounded_and_terminal_is_lossless); + RUN_TEST(subprocess_final_log_drain_error_is_terminal_and_preserves_classification); RUN_TEST(subprocess_posix_child_closes_unrelated_descriptors); RUN_TEST(subprocess_root_exit_drains_surviving_descendant); RUN_TEST(win_cmdline_index_worker_json); diff --git a/tests/test_windows_bundle_contract.sh b/tests/test_windows_bundle_contract.sh index 53b524ea7..32dfa46cb 100644 --- a/tests/test_windows_bundle_contract.sh +++ b/tests/test_windows_bundle_contract.sh @@ -374,6 +374,41 @@ require( "Windows launcher guard setup must be covered by profile-fixture cleanup", ) +# The hosted runner profile can itself be trusted while newly-created children +# still inherit mutation-capable principals. Require the guard root to replace +# that inherited DACL with a protected, current-account-owned ACL before any +# launcher bundle or Python temporary descendants are created below it. +guard_root_creation = ( + 'New-Item -ItemType Directory -Path $guardRoot | Out-Null' +) +guard_bundle_creation = '$guardBundle = Join-Path $guardRoot ' +acl_start = windows_test_driver.find(guard_root_creation) +acl_end = windows_test_driver.find(guard_bundle_creation, acl_start + 1) +guard_acl_setup = ( + windows_test_driver[acl_start:acl_end] + if acl_start >= 0 and acl_end > acl_start + else "" +) +require( + all( + needle in guard_acl_setup + for needle in ( + '[System.Security.Principal.WindowsIdentity]::GetCurrent().User', + '[System.Security.AccessControl.DirectorySecurity]::new()', + '$guardAcl.SetOwner($currentSid)', + '$guardAcl.SetAccessRuleProtection($true, $false)', + '[System.Security.AccessControl.FileSystemRights]::FullControl', + '[System.Security.AccessControl.InheritanceFlags]::ContainerInherit', + '[System.Security.AccessControl.InheritanceFlags]::ObjectInherit', + '[System.Security.AccessControl.PropagationFlags]::None', + '[System.Security.AccessControl.AccessControlType]::Allow', + 'Set-Acl -LiteralPath $guardRoot -AclObject $guardAcl', + ) + ), + "Windows launcher guards must protect the guard-root DACL and grant only " + "the current account inheritable full control before creating descendants", +) + # Launcher supervision has two distinct failure directions: killing the # launcher must kill its payload job, and killing only the launcher's immediate # parent must terminate the launcher plus every descendant. Require the native