fix: close remaining daemon coordination races

Signed-off-by: Martin Vogel <martin.vogel.tech@gmail.com>
This commit is contained in:
Martin Vogel committed 2026-07-18 15:26:38 +02:00
1 parent f35b10b79b
commit ad2874bcff
14 files changed
+515 -195

No files matched your search

+21
View File
@@ -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"
+14 -3
View File
@@ -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;
}
+16 -15
View File
@@ -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;
}
+144 -55
View File
@@ -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 <stdio.h>
@@ -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
+17 -6
View File
@@ -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
+2 -2
View File
@@ -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
+12
View File
@@ -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;
}
+8
View File
@@ -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 "<service-stub>.<method>" 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).
+16 -84
View File
@@ -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 <sqlite3/sqlite3.h>
#include <stdatomic.h>
@@ -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);
+37
View File
@@ -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__
+25 -5
View File
@@ -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) {
+11 -2
View File
@@ -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);
+157 -23
View File
@@ -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);
+35
View File
@@ -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