diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md index a02459b4a..2f0d6717a 100644 --- a/docs/CONFIGURATION.md +++ b/docs/CONFIGURATION.md @@ -94,7 +94,7 @@ Current keys: | `index_max_files` | `off` | Optional maximum number of accepted source files in one discovery run. | | `index_max_source_mb` | `off` | Optional maximum accepted source size in MiB in one discovery run. | | `index_max_rss_mb` | `off` | Optional maximum charged memory in MiB for the complete contained index-worker process tree (`64..1048576`). On macOS this is phys_footprint, not resident size. | -| `index_max_duration_seconds` | `off` | Optional maximum wall-clock duration in seconds for the whole index request, including crash/hang recovery (`1..86400`). | +| `index_max_duration_seconds` | `off` | Optional maximum wall-clock duration in seconds for the whole index request, including crash/hang recovery (`1..86400`). Like `index_max_rss_mb`, it stops a worker only before it starts to publish the new index; see [INDEX_RESOURCE_LIMITS.md](INDEX_RESOURCE_LIMITS.md). | For a watched project, `index_status` names the polling method in `watch.strategy`: `git` (HEAD and dirty-state polling), `tree` (a non-git root polled by the diff --git a/docs/INDEX_RESOURCE_LIMITS.md b/docs/INDEX_RESOURCE_LIMITS.md index 11fcf4a3d..d76b162d7 100644 --- a/docs/INDEX_RESOURCE_LIMITS.md +++ b/docs/INDEX_RESOURCE_LIMITS.md @@ -58,6 +58,18 @@ terminal only after the tree is quiescent or a bounded containment failure is explicitly surfaced. Resource termination is not retried and does not quarantine a source file. +A worker limit never stops a worker after it has started to publish. The +supervisor enforces the limits, but only the worker replaces the serving +database, so the two coordinate through a per-attempt publication claim that +exactly one of them can win. The worker claims it immediately before it +replaces the serving database; the supervisor claims it before it stops the +worker for a limit. If the limit is reached first, the worker can no longer +publish and the previous index stays untouched. If the worker claimed first, +the limits no longer apply to that attempt: the new index is published and the +rest of the attempt, including the persistence artifact export, runs to +completion. The supervisor then logs `index.supervisor.resource_limit_after_publication` +instead of stopping it. With both worker limits `off` there is no claim. + If RSS is enabled and three consecutive probes cannot obtain any trustworthy tree measurement while the root worker is still running, CBM fails closed with `code=resource_probe_failed`. A probe is untrustworthy as soon as one member of @@ -102,7 +114,10 @@ Worker limit failures use the same shape with `stage=worker`, `resource=rss_bytes` and `unit=bytes`, or `resource=duration_ms` and `unit=milliseconds`. RSS measurement failures use `code=resource_probe_failed` and omit `observed`, `limit`, and `unit` because no trustworthy observation was -available. +available. For a worker failure `serving_index_preserved` describes the whole +request: it is true only when a servable index existed when the request began, +no attempt of the request (including an earlier one that crashed and was +recovered) claimed publication, and that index still serves. ## Trust and compatibility diff --git a/src/daemon/application.c b/src/daemon/application.c index 0f696d59c..207735e04 100644 --- a/src/daemon/application.c +++ b/src/daemon/application.c @@ -160,6 +160,10 @@ struct cbm_daemon_application_job { bool async_owned; bool worker_started; uint64_t request_started_ms; + /* What a worker resource failure of this job is reported against; set + * before the first attempt, folded over every attempt (mcp_internal.h). */ + bool request_facts_begun; + cbm_mcp_index_request_facts_t request_facts; cbm_daemon_application_job_t *next; }; @@ -1346,6 +1350,17 @@ static void application_cancelled_result(cbm_index_worker_result_t *result) { result->tree_quiesced = true; } +/* Only an enabled worker limit can end a job in a worker resource failure, the + * one response the request facts feed; without one nothing is read. */ +static void application_job_request_facts_begin(cbm_daemon_application_job_t *job) { + cbm_index_resource_policy_t policy; + char error[CBM_SZ_256] = {0}; + if (cbm_mcp_index_policy_from_internal_args(job->args_json, &policy, error, sizeof(error)) && + cbm_index_policy_worker_enabled(&policy)) { + cbm_mcp_index_request_facts_begin(job->args_json, &job->request_facts); + } +} + static application_attempt_status_t application_job_run_attempt(cbm_daemon_application_job_t *job, const char *marker_path, const char *quarantine_path, @@ -1382,6 +1397,10 @@ static application_attempt_status_t application_job_run_attempt(cbm_daemon_appli if (job->request_started_ms == 0) { job->request_started_ms = cbm_index_worker_now_ms(); } + if (!job->request_facts_begun) { + job->request_facts_begun = true; + application_job_request_facts_begin(job); + } application_tmp_lock(); int start_result = application->worker_ops.start( application->worker_ops.context, job->args_json, memory_budget_bytes, marker_path, @@ -1573,6 +1592,7 @@ static application_attempt_decision_t application_consume_attempt( application_attempt_disposition(job, attempt); application_record_attempt(attempt, &execution->last_result, &execution->have_last_result, execution->last_log); + cbm_mcp_index_request_facts_record(&job->request_facts, &attempt->result); if (disposition == CBM_MCP_SUPERVISED_RESULT_SUCCESS) { execution->response = attempt->result.response; attempt->result.response = NULL; @@ -1581,8 +1601,8 @@ static application_attempt_decision_t application_consume_attempt( return APPLICATION_ATTEMPT_DECISION_SUCCESS; } if (disposition == CBM_MCP_SUPERVISED_RESULT_RESOURCE_FAILURE) { - execution->response = - cbm_mcp_index_worker_resource_response(job->args_json, &attempt->result); + execution->response = cbm_mcp_index_worker_resource_response( + job->args_json, &attempt->result, &job->request_facts); application_attempt_free(attempt); return APPLICATION_ATTEMPT_DECISION_STOP; } diff --git a/src/daemon/bootstrap.c b/src/daemon/bootstrap.c index cb384cd20..617b9c06f 100644 --- a/src/daemon/bootstrap.c +++ b/src/daemon/bootstrap.c @@ -141,6 +141,12 @@ static bool bootstrap_worker_argv_exact(int argc, char *const argv[]) { } next += 2; } + if (next < argc && bootstrap_arg_is(argv[next], "--index-worker-publish-fence")) { + if (next + 1 >= argc || !argv[next + 1] || !argv[next + 1][0]) { + return false; + } + next += 2; + } return next == argc; } diff --git a/src/main.c b/src/main.c index 54bad875e..82be46e03 100644 --- a/src/main.c +++ b/src/main.c @@ -897,6 +897,8 @@ static int run_cli(int argc, char **argv, cbm_project_lock_manager_t *project_lo const char *worker_marker = cli_strip_flag_value(&argc, argv, CBM_INDEX_WORKER_MARKER_ARG); const char *worker_quarantine = cli_strip_flag_value(&argc, argv, CBM_INDEX_WORKER_QUARANTINE_ARG); + const char *worker_publish_fence = + cli_strip_flag_value(&argc, argv, CBM_INDEX_WORKER_PUBLISH_FENCE_ARG); if (index_worker) { /* The graph lives on this process's heaps: SQLite gets heaps of its * own here, and only here (cbm_sqlite_dedicated_heap). */ @@ -906,6 +908,7 @@ static int run_cli(int argc, char **argv, cbm_project_lock_manager_t *project_lo cbm_index_set_worker_role_options(index_worker, response_out, worker_single_thread, worker_marker, worker_quarantine, cbm_index_worker_memory_budget_bytes()); + cbm_index_set_worker_publish_fence(index_worker ? worker_publish_fence : NULL); if (argc < MAIN_MIN_ARGC) { (void)fprintf(stderr, CLI_USAGE); @@ -3188,6 +3191,7 @@ int main(int argc, char **argv) { cbm_index_set_worker_role_options(true, invocation.response_out, invocation.single_thread, invocation.marker_file, invocation.quarantine_file, invocation.memory_budget_bytes); + cbm_index_set_worker_publish_fence(invocation.publish_fence_file); #ifndef _WIN32 /* Split into three ordered steps rather than one condition, because the * ORDER is load-bearing and the middle step only exists in test builds: diff --git a/src/mcp/index_supervisor.c b/src/mcp/index_supervisor.c index e84dff755..c8c4224ad 100644 --- a/src/mcp/index_supervisor.c +++ b/src/mcp/index_supervisor.c @@ -92,6 +92,63 @@ size_t cbm_index_worker_memory_budget_bytes(void) { return g_worker_memory_budget_bytes; } +/* ── Publication fence (see index_supervisor.h) ───────────────────── */ + +typedef enum { + INDEX_PUBLICATION_CLAIMED = 0, /* this caller created the fence file */ + INDEX_PUBLICATION_TAKEN, /* the other side created it first */ + INDEX_PUBLICATION_CLAIM_ERROR, /* neither could be established */ +} index_publication_claim_t; + +/* Exclusive creation is the single atomic decision both processes race for: + * "x" is O_CREAT|O_EXCL on POSIX and CREATE_NEW on Windows, so exactly one + * creator succeeds and the other gets EEXIST. A deletion-based claim would + * be ambiguous on Windows, where a delete-pending file reports access denied. */ +static index_publication_claim_t index_publication_claim(const char *fence_file) { + if (!fence_file || !fence_file[0]) { + return INDEX_PUBLICATION_CLAIM_ERROR; + } + errno = 0; + FILE *claim = cbm_fopen(fence_file, "wbx"); + if (claim) { + (void)fclose(claim); + return INDEX_PUBLICATION_CLAIMED; + } + return errno == EEXIST ? INDEX_PUBLICATION_TAKEN : INDEX_PUBLICATION_CLAIM_ERROR; +} + +static char g_worker_publish_fence[CBM_SZ_4K] = {0}; +static bool g_worker_publication_claimed = false; + +void cbm_index_set_worker_publish_fence(const char *fence_file) { + if (fence_file && fence_file[0]) { + (void)snprintf(g_worker_publish_fence, sizeof(g_worker_publish_fence), "%s", fence_file); + } else { + g_worker_publish_fence[0] = '\0'; + } + g_worker_publication_claimed = false; +} + +const char *cbm_index_worker_publish_fence(void) { + return g_worker_publish_fence[0] ? g_worker_publish_fence : NULL; +} + +bool cbm_index_worker_claim_publication(void) { + if (!g_worker_publish_fence[0] || g_worker_publication_claimed) { + return true; + } + index_publication_claim_t claim = index_publication_claim(g_worker_publish_fence); + if (claim == INDEX_PUBLICATION_CLAIMED) { + g_worker_publication_claimed = true; + cbm_log_info("index.worker.publication_claimed", "fence", g_worker_publish_fence); + return true; + } + cbm_log_warn("index.worker.publication_refused", "reason", + claim == INDEX_PUBLICATION_TAKEN ? "resource_limit" : "claim_failed", "fence", + g_worker_publish_fence); + return false; +} + /* Worker log startup header — see index_supervisor.h. */ static atomic_flag g_worker_log_begun = ATOMIC_FLAG_INIT; @@ -250,6 +307,13 @@ cbm_index_worker_argv_status_t cbm_index_worker_parse_process_argv( parsed.quarantine_file = argv[next + 1]; next += 2; } + if (next < argc && argv[next] && strcmp(argv[next], CBM_INDEX_WORKER_PUBLISH_FENCE_ARG) == 0) { + if (next + 1 >= argc || !argv[next + 1] || !argv[next + 1][0]) { + return CBM_INDEX_WORKER_ARGV_INVALID; + } + parsed.publish_fence_file = argv[next + 1]; + next += 2; + } if (next != argc) { return CBM_INDEX_WORKER_ARGV_INVALID; } @@ -477,7 +541,7 @@ static void worker_log_last_phase(const char *log_path, char *out, size_t out_si enum { INDEX_WORKER_PATH_CAP = CBM_SZ_4K, - INDEX_WORKER_ARGV_CAP = 17, + INDEX_WORKER_ARGV_CAP = 19, INDEX_WORKER_SYNC_POLL_NS = 10000000, INDEX_WORKER_RELAY_LINES_PER_POLL = 64, INDEX_WORKER_RELAY_BYTES_PER_POLL = 64 * 1024, @@ -492,6 +556,14 @@ typedef enum { INDEX_WORKER_TERMINATION_RESOURCE, } index_worker_termination_reason_t; +/* Who owns the publication fence of one attempt. Only the polling owner thread + * reads or writes it. */ +typedef enum { + INDEX_WORKER_PUBLICATION_OPEN = 0, /* unclaimed, or no fence at all */ + INDEX_WORKER_PUBLICATION_SUPERVISOR, + INDEX_WORKER_PUBLICATION_WORKER, +} index_worker_publication_owner_t; + #ifdef CBM_ENABLE_TEST_SEAMS static cbm_index_supervisor_clock_fn g_resource_clock; static cbm_index_supervisor_rss_fn g_resource_rss; @@ -545,11 +617,24 @@ struct cbm_index_worker_handle { uint64_t last_rss_probe_ms; unsigned int rss_probe_failures; bool rss_probe_started; + /* Publication fence: a never-created path next to a reserved unique file + * (both removed at terminal). Empty when no worker limit is enabled. */ + char publish_fence_reserve[INDEX_WORKER_PATH_CAP]; + char publish_fence_path[INDEX_WORKER_PATH_CAP]; + index_worker_publication_owner_t publication_owner; + bool publication_claim_error_logged; atomic_int termination_reason; atomic_bool terminal; cbm_index_worker_result_t result; }; +#ifdef CBM_ENABLE_TEST_SEAMS +const char *cbm_index_worker_publish_fence_path_for_testing( + const cbm_index_worker_handle_t *handle) { + return handle && handle->publish_fence_path[0] ? handle->publish_fence_path : NULL; +} +#endif + /* The generic subprocess supervisor tails logs in bounded batches to preserve * its nonblocking poll contract. A short-lived worker can therefore become * terminal with more completed lines still on disk. Keep the request callback @@ -801,9 +886,58 @@ size_t cbm_index_worker_job_memory_limit(size_t memory_budget_bytes) { return memory_budget_bytes > SIZE_MAX - headroom ? SIZE_MAX : memory_budget_bytes + headroom; } +/* Claim the publication fence before stopping the worker for a resource limit + * (see index_supervisor.h). True when the termination may proceed. */ +static bool worker_claim_publication_for_limit(cbm_index_worker_handle_t *handle, + cbm_index_resource_t resource, uint64_t observed, + uint64_t limit, bool probe_failed) { + if (!handle->publish_fence_path[0] || + handle->publication_owner == INDEX_WORKER_PUBLICATION_SUPERVISOR) { + return true; + } + if (handle->publication_owner == INDEX_WORKER_PUBLICATION_WORKER) { + return false; + } + index_publication_claim_t claim = index_publication_claim(handle->publish_fence_path); + if (claim == INDEX_PUBLICATION_CLAIMED) { + handle->publication_owner = INDEX_WORKER_PUBLICATION_SUPERVISOR; + return true; + } + if (claim == INDEX_PUBLICATION_TAKEN) { + /* The worker committed to publishing before the limit was observed. + * Stopping it now could leave a new generation serving behind a report + * that the previous index was preserved, so this attempt is no longer + * limited; the crossing is recorded here instead. */ + handle->publication_owner = INDEX_WORKER_PUBLICATION_WORKER; + char observed_text[CBM_SZ_32]; + char limit_text[CBM_SZ_32]; + (void)snprintf(observed_text, sizeof(observed_text), "%llu", (unsigned long long)observed); + (void)snprintf(limit_text, sizeof(limit_text), "%llu", (unsigned long long)limit); + cbm_log_warn("index.supervisor.resource_limit_after_publication", "resource", + cbm_index_resource_name(resource), "observed", + probe_failed ? "unmeasured" : observed_text, "limit", limit_text, "action", + "not_enforced", "log", handle->log_path); + return false; + } + /* Neither side holds the fence and it could not be created. Terminating + * now could race a later successful claim by the worker, so the decision + * is retried on the next poll instead. */ + if (!handle->publication_claim_error_logged) { + handle->publication_claim_error_logged = true; + char errno_text[CBM_SZ_32]; + (void)snprintf(errno_text, sizeof(errno_text), "%d", errno); + cbm_log_error("index.supervisor.publication_claim_failed", "errno", errno_text, "action", + "retry_next_poll", "log", handle->log_path); + } + return false; +} + static void worker_request_resource_termination(cbm_index_worker_handle_t *handle, cbm_index_resource_t resource, uint64_t observed, uint64_t limit, bool probe_failed) { + if (!worker_claim_publication_for_limit(handle, resource, observed, limit, probe_failed)) { + return; + } int expected = INDEX_WORKER_TERMINATION_NONE; if (!atomic_compare_exchange_strong_explicit(&handle->termination_reason, &expected, INDEX_WORKER_TERMINATION_RESOURCE, @@ -824,7 +958,9 @@ static void worker_request_resource_termination(cbm_index_worker_handle_t *handl } static bool worker_check_resource_limits(cbm_index_worker_handle_t *handle) { - if (atomic_load_explicit(&handle->termination_reason, memory_order_acquire) != + /* A worker that committed to publishing is no longer resource-limited. */ + if (handle->publication_owner == INDEX_WORKER_PUBLICATION_WORKER || + atomic_load_explicit(&handle->termination_reason, memory_order_acquire) != INDEX_WORKER_TERMINATION_NONE || cbm_subprocess_termination_pending(handle->process) || !cbm_subprocess_supervision_active(handle->process)) { @@ -877,6 +1013,56 @@ duration_check: return rss_probe_failed; } +/* Reserve a unique private name and derive the fence path from it. The fence + * itself is never created here: creating it IS the claim. */ +static bool worker_publish_fence_create(cbm_index_worker_handle_t *handle) { + if (!worker_unique_file(handle->publish_fence_reserve, sizeof(handle->publish_fence_reserve), + "publish")) { + return false; + } + int written = snprintf(handle->publish_fence_path, sizeof(handle->publish_fence_path), + "%s.claim", handle->publish_fence_reserve); + if (written <= 0 || written >= (int)sizeof(handle->publish_fence_path)) { + (void)cbm_unlink(handle->publish_fence_reserve); + handle->publish_fence_reserve[0] = '\0'; + handle->publish_fence_path[0] = '\0'; + return false; + } + return true; +} + +static void worker_publish_fence_remove(cbm_index_worker_handle_t *handle) { + if (handle->publish_fence_path[0]) { + (void)cbm_unlink(handle->publish_fence_path); + } + if (handle->publish_fence_reserve[0]) { + (void)cbm_unlink(handle->publish_fence_reserve); + } +} + +/* Settle who held the fence once the attempt is terminal. Claiming an unowned + * fence now also bars any later claim, so a stray member of an unquiesced tree + * can never publish after this report. */ +static bool worker_publication_committed_at_terminal(cbm_index_worker_handle_t *handle) { + if (!handle->publish_fence_path[0] || + handle->publication_owner == INDEX_WORKER_PUBLICATION_SUPERVISOR) { + return false; + } + if (handle->publication_owner == INDEX_WORKER_PUBLICATION_WORKER) { + return true; + } + index_publication_claim_t claim = index_publication_claim(handle->publish_fence_path); + if (claim == INDEX_PUBLICATION_CLAIMED) { + handle->publication_owner = INDEX_WORKER_PUBLICATION_SUPERVISOR; + return false; + } + if (claim == INDEX_PUBLICATION_TAKEN) { + handle->publication_owner = INDEX_WORKER_PUBLICATION_WORKER; + } + /* Taken: the worker committed. Unreadable: preservation cannot be proven. */ + return true; +} + static int worker_start_internal(const char *args_json, size_t memory_budget_bytes, const cbm_index_resource_policy_t *resource_policy, bool single_thread, const char *marker_file, @@ -935,6 +1121,19 @@ static int worker_start_internal(const char *args_json, size_t memory_budget_byt free(handle); return -1; } + /* Opt-in: only an enabled worker limit can stop a worker, so only then is + * there a publication to coordinate. */ + if (cbm_index_policy_worker_enabled(&handle->resource_policy) && + !worker_publish_fence_create(handle)) { + char error_text[CBM_SZ_32]; + (void)snprintf(error_text, sizeof(error_text), "%d", errno); + cbm_log_error("index.supervisor.artifact_create_failed", "artifact", "publish_fence", + "errno", error_text); + (void)cbm_unlink(handle->response_path); + (void)cbm_unlink(handle->log_path); + free(handle); + return -1; + } const char *argv[INDEX_WORKER_ARGV_CAP]; size_t argc = 0; @@ -964,6 +1163,10 @@ static int worker_start_internal(const char *args_json, size_t memory_budget_byt argv[argc++] = CBM_INDEX_WORKER_QUARANTINE_ARG; argv[argc++] = quarantine_file; } + if (handle->publish_fence_path[0]) { + argv[argc++] = CBM_INDEX_WORKER_PUBLISH_FENCE_ARG; + argv[argc++] = handle->publish_fence_path; + } argv[argc] = NULL; cbm_proc_opts_t options = {0}; @@ -981,6 +1184,7 @@ static int worker_start_internal(const char *args_json, size_t memory_budget_byt if (cbm_subprocess_spawn(&options, &handle->process) != 0) { (void)cbm_unlink(handle->response_path); (void)cbm_unlink(handle->log_path); + worker_publish_fence_remove(handle); free(handle); cbm_log_error("index.supervisor.spawn_failed", "action", "fail_closed"); return -1; @@ -1089,6 +1293,12 @@ cbm_index_worker_poll_t cbm_index_worker_poll(cbm_index_worker_handle_t *handle, handle->result.forced = process_result->forced; handle->result.tree_quiesced = process_result->tree_quiesced; handle->result.supervision_failed = process_result->supervision_failed; + handle->result.publication_committed = worker_publication_committed_at_terminal(handle); + if (handle->result.tree_quiesced) { + /* An unquiesced tree keeps the fence as it is: removing a claim the + * supervisor holds would let a surviving member publish later. */ + worker_publish_fence_remove(handle); + } if (worker_result_succeeded(&handle->result)) { worker_response_read_status_t response_status; handle->result.response = slurp_worker_response(handle->response_path, &response_status); diff --git a/src/mcp/index_supervisor.h b/src/mcp/index_supervisor.h index 2cb244a7e..359a3e25e 100644 --- a/src/mcp/index_supervisor.h +++ b/src/mcp/index_supervisor.h @@ -42,6 +42,7 @@ void cbm_index_set_worker_role(bool is_worker, const char *response_out); #define CBM_INDEX_WORKER_QUARANTINE_ARG "--index-worker-quarantine" #define CBM_INDEX_WORKER_MEMORY_BUDGET_ARG "--index-worker-memory-budget-bytes" #define CBM_INDEX_WORKER_BUILD_ARG "--index-worker-build" +#define CBM_INDEX_WORKER_PUBLISH_FENCE_ARG "--index-worker-publish-fence" #define CBM_INDEX_WORKER_BUILD_FINGERPRINT_LENGTH 64U #define CBM_INDEX_WORKER_BUILD_FINGERPRINT_SIZE 65U void cbm_index_set_worker_role_options(bool is_worker, const char *response_out, bool single_thread, @@ -51,6 +52,32 @@ bool cbm_index_worker_active(void); const char *cbm_index_worker_response_out(void); size_t cbm_index_worker_memory_budget_bytes(void); +/* Publication fence. A worker resource limit (index_max_rss_mb, + * index_max_duration_seconds) is enforced by the supervisor, while only the + * worker replaces the serving database. When such a limit is enabled, every + * worker attempt gets a fence file path, and publication and resource + * termination both CLAIM it by creating that file exclusively, so exactly one + * of them can win: + * - the worker claims immediately before it replaces the serving database; + * losing means the supervisor already decided to stop it for a limit, so + * the run aborts before publication and the previous index stays intact; + * - the supervisor claims before it terminates the worker for a limit; + * losing means the worker already committed to publishing, so the limit is + * no longer enforced for that attempt and the attempt runs to completion. + * A reported resource failure therefore never follows a publication by the + * same attempt. Without an enabled worker limit there is no fence and nothing + * changes. The supervisor passes the path on the worker argv; the worker entry + * installs it here (NULL clears it). */ +void cbm_index_set_worker_publish_fence(const char *fence_file); +const char *cbm_index_worker_publish_fence(void); + +/* Worker side: claim publication. True when publishing may proceed: no fence + * is installed, or this call created the fence file. False when the + * supervisor already claimed it (a resource limit stopped this attempt) or the + * claim could not be made; the caller must then leave the serving database + * untouched. Idempotent after a successful claim. */ +bool cbm_index_worker_claim_publication(void); + /* Event name of the worker log's startup header. Shared with the tests so the * contract has exactly one spelling. */ #define CBM_INDEX_WORKER_LOG_START_EVENT "index.worker.start" @@ -96,6 +123,7 @@ typedef struct { const char *marker_file; const char *quarantine_file; size_t memory_budget_bytes; + const char *publish_fence_file; } cbm_index_worker_invocation_t; typedef enum { @@ -160,6 +188,12 @@ typedef struct { bool response_rejected; /* clean worker exceeded the bounded response protocol */ cbm_index_resource_violation_t resource_violation; bool resource_probe_failed; + /* The worker claimed the publication fence: it committed to replacing the + * serving database (whether or not it got that far). Also true when the + * fence state could not be read, because a report may only claim the + * previous index was preserved when it can prove it. Always false without + * a fence (no worker limit enabled) and for an attempt stopped by a limit. */ + bool publication_committed; char *response; /* worker result only after a contained, uncancelled CLEAN exit; * borrowed for async polls, caller-owned from the sync wrapper */ } cbm_index_worker_result_t; @@ -283,6 +317,9 @@ void cbm_index_supervisor_set_resource_hooks_for_testing(cbm_index_supervisor_cl cbm_index_supervisor_rss_fn rss_fn, void *context); void cbm_index_supervisor_reset_resource_hooks_for_testing(void); +/* Borrowed fence path of an attempt; NULL when it has no fence. */ +const char *cbm_index_worker_publish_fence_path_for_testing( + const cbm_index_worker_handle_t *handle); #endif #endif /* CBM_INDEX_SUPERVISOR_H */ diff --git a/src/mcp/mcp.c b/src/mcp/mcp.c index f23a00e5c..3dc5e72ee 100644 --- a/src/mcp/mcp.c +++ b/src/mcp/mcp.c @@ -11277,16 +11277,50 @@ static char *build_worker_unsafe_terminal_response(const char *args, cbm_proc_ou return index_worker_json_result(doc, repo_path, NULL, NULL); } -char *cbm_mcp_index_worker_resource_response(const char *args, - const cbm_index_worker_result_t *worker_result) { +/* Whether the project named by internal worker args has a servable index now. */ +static bool index_args_serving_index_servable(const char *args) { char *repo_path = cbm_mcp_get_string_arg(args, "repo_path"); char *name_override = cbm_mcp_get_string_arg(args, "name"); char *project_name = cbm_project_name_from_path(name_override && name_override[0] ? name_override : repo_path); - char db_path[CBM_SZ_1K] = {0}; + bool servable = false; if (project_name) { + char db_path[CBM_SZ_1K] = {0}; project_db_path(project_name, db_path, sizeof(db_path)); + servable = project_db_is_servable(project_name, db_path); } + free(repo_path); + free(name_override); + free(project_name); + return servable; +} + +void cbm_mcp_index_request_facts_begin(const char *args, cbm_mcp_index_request_facts_t *facts) { + if (!facts) { + return; + } + facts->serving_index_existed = index_args_serving_index_servable(args); + facts->publication_committed = false; +} + +void cbm_mcp_index_request_facts_record(cbm_mcp_index_request_facts_t *facts, + const cbm_index_worker_result_t *worker_result) { + if (facts && worker_result && worker_result->publication_committed) { + facts->publication_committed = true; + } +} + +char *cbm_mcp_index_worker_resource_response(const char *args, + const cbm_index_worker_result_t *worker_result, + const cbm_mcp_index_request_facts_t *facts) { + char *repo_path = cbm_mcp_get_string_arg(args, "repo_path"); + /* The publication fence makes a limit-stopped attempt and a publication by + * that attempt mutually exclusive; an EARLIER attempt of the request may + * still have published (it crashed after committing), and an index that + * appeared during the request (artifact bootstrap) was not preserved. */ + bool serving_index_preserved = + facts && facts->serving_index_existed && !facts->publication_committed && + !worker_result->publication_committed && index_args_serving_index_servable(args); const cbm_index_resource_violation_t *violation = &worker_result->resource_violation; const char *config_key = cbm_index_resource_config_key(violation->resource); @@ -11311,13 +11345,12 @@ char *cbm_mcp_index_worker_resource_response(const char *args, yyjson_mut_obj_add_str(doc, root, "unit", cbm_index_resource_unit(violation->resource)); } yyjson_mut_obj_add_bool(doc, root, "retryable", true); - yyjson_mut_obj_add_bool(doc, root, "serving_index_preserved", - project_name && project_db_is_servable(project_name, db_path)); + yyjson_mut_obj_add_bool(doc, root, "serving_index_preserved", serving_index_preserved); yyjson_mut_obj_add_strcpy(doc, root, "message", message); if (repo_path) { yyjson_mut_obj_add_strcpy(doc, root, "repo_path", repo_path); } - return index_worker_json_result(doc, repo_path, name_override, project_name); + return index_worker_json_result(doc, repo_path, NULL, NULL); } /* Drop the cached store so the next query reopens whatever the worker wrote (each @@ -11618,6 +11651,12 @@ static char *index_run_supervised(cbm_mcp_server_t *srv, const char *args, const cbm_index_resource_policy_t *resource_policy) { invalidate_cached_store(srv); uint64_t request_started_ms = cbm_index_worker_now_ms(); + /* Only an enabled worker limit can produce a worker resource failure, the + * one response these facts are reported in. */ + cbm_mcp_index_request_facts_t facts = {0}; + if (cbm_index_policy_worker_enabled(resource_policy)) { + cbm_mcp_index_request_facts_begin(args, &facts); + } /* First attempt: normal parallel run. */ cbm_index_worker_result_t wr; @@ -11627,6 +11666,7 @@ static char *index_run_supervised(cbm_mcp_server_t *srv, const char *args, request_started_ms, &wr); cbm_mcp_supervised_result_disposition_t disposition = cbm_mcp_supervised_result_disposition(rc, &wr); + cbm_mcp_index_request_facts_record(&facts, &wr); if (wr.response_missing) { /* #1300: a clean exit without a response is a named failure. */ @@ -11649,7 +11689,7 @@ static char *index_run_supervised(cbm_mcp_server_t *srv, const char *args, return failure; } if (disposition == CBM_MCP_SUPERVISED_RESULT_RESOURCE_FAILURE) { - char *failure = cbm_mcp_index_worker_resource_response(args, &wr); + char *failure = cbm_mcp_index_worker_resource_response(args, &wr, &facts); cbm_index_worker_result_free(&wr); invalidate_cached_store(srv); return failure; @@ -11709,6 +11749,7 @@ static char *index_run_supervised(cbm_mcp_server_t *srv, const char *args, srv ? &srv->pipeline_cancel_requested : NULL, request_started_ms, &wr2); cbm_mcp_supervised_result_disposition_t recovery_disposition = cbm_mcp_supervised_result_disposition(rc2, &wr2); + cbm_mcp_index_request_facts_record(&facts, &wr2); if (wr2.response_missing) { /* #1300: a recovery run that exits cleanly without a response is * reported as such, not as a failed recovery start. */ @@ -11732,7 +11773,7 @@ static char *index_run_supervised(cbm_mcp_server_t *srv, const char *args, break; } if (recovery_disposition == CBM_MCP_SUPERVISED_RESULT_RESOURCE_FAILURE) { - resp = cbm_mcp_index_worker_resource_response(args, &wr2); + resp = cbm_mcp_index_worker_resource_response(args, &wr2, &facts); cbm_index_worker_result_free(&wr2); break; } @@ -11836,6 +11877,7 @@ static char *index_run_supervised(cbm_mcp_server_t *srv, const char *args, srv ? &srv->pipeline_cancel_requested : NULL, request_started_ms, &wrp); cbm_mcp_supervised_result_disposition_t partial_disposition = cbm_mcp_supervised_result_disposition(rcp, &wrp); + cbm_mcp_index_request_facts_record(&facts, &wrp); if (partial_disposition == CBM_MCP_SUPERVISED_RESULT_SUCCESS) { resp = wrp.response; /* transfer ownership to caller */ wrp.response = NULL; @@ -11848,7 +11890,7 @@ static char *index_run_supervised(cbm_mcp_server_t *srv, const char *args, unsafe_terminal = true; terminal_cancelled = wrp.cancellation_requested; } else if (partial_disposition == CBM_MCP_SUPERVISED_RESULT_RESOURCE_FAILURE) { - resp = cbm_mcp_index_worker_resource_response(args, &wrp); + resp = cbm_mcp_index_worker_resource_response(args, &wrp, &facts); } cbm_index_worker_result_free(&wrp); } @@ -12293,6 +12335,11 @@ static char *handle_index_repository_status(cbm_mcp_server_t *srv, const char *a return result ? result : cbm_mcp_text_result("index job status is unavailable", true); } +static bool index_worker_publish_gate(void *context) { + (void)context; + return cbm_index_worker_claim_publication(); +} + static char *handle_index_repository(cbm_mcp_server_t *srv, const char *args) { bool async_mode = false; bool status_mode = false; @@ -12492,6 +12539,11 @@ static char *handle_index_repository(cbm_mcp_server_t *srv, const char *args) { free(name_override); cbm_pipeline_set_persistence(p, persistence); cbm_pipeline_set_resource_policy(p, &resource_policy); + if (cbm_index_worker_publish_fence()) { + /* Supervised worker with a worker limit: claim publication from the + * supervisor before the serving database is replaced. */ + cbm_pipeline_set_publish_gate(p, index_worker_publish_gate, NULL); + } index_doc_candidates_apply(p, doc_candidates); char *project_name = heap_strdup(cbm_pipeline_project_name(p)); diff --git a/src/mcp/mcp_internal.h b/src/mcp/mcp_internal.h index 3ca7c6ee9..59e769dd9 100644 --- a/src/mcp/mcp_internal.h +++ b/src/mcp/mcp_internal.h @@ -62,8 +62,27 @@ bool cbm_mcp_index_policy_add_to_args(yyjson_mut_doc *doc, yyjson_mut_val *root, const cbm_index_resource_policy_t *policy); bool cbm_mcp_index_policy_from_internal_args(const char *args, cbm_index_resource_policy_t *policy, char *error, size_t error_size); + +/* Request-scoped facts a worker resource failure is reported against. One + * supervised request can run several worker attempts (crash/hang recovery), and + * serving_index_preserved describes the request, not only its last attempt: + * it is true only when a servable index served the project when the request + * began AND no attempt of the request committed to replacing it (the + * publication fence, index_supervisor.h) AND that index is still servable. */ +typedef struct { + bool serving_index_existed; + bool publication_committed; +} cbm_mcp_index_request_facts_t; + +/* Capture the facts at the start of a request, before its first worker. */ +void cbm_mcp_index_request_facts_begin(const char *args, cbm_mcp_index_request_facts_t *facts); +/* Fold one finished attempt into the request facts. */ +void cbm_mcp_index_request_facts_record(cbm_mcp_index_request_facts_t *facts, + const cbm_index_worker_result_t *worker_result); +/* NULL facts claim nothing: serving_index_preserved is then false. */ char *cbm_mcp_index_worker_resource_response(const char *args, - const cbm_index_worker_result_t *worker_result); + const cbm_index_worker_result_t *worker_result, + const cbm_mcp_index_request_facts_t *facts); /* Encode a doc_candidates setting the user set to a boolean on an internal * worker request (the worker reads no config). Unset or unparseable adds no diff --git a/src/pipeline/pipeline.c b/src/pipeline/pipeline.c index e14bfb7f7..f1965ad4b 100644 --- a/src/pipeline/pipeline.c +++ b/src/pipeline/pipeline.c @@ -303,6 +303,9 @@ struct cbm_pipeline { * per pipeline so concurrent test/process activity cannot cross-trigger. */ void (*before_publish_hook)(cbm_pipeline_t *, const char *, void *); void *before_publish_hook_ctx; + /* Production publication gate (cbm_pipeline_set_publish_gate). */ + cbm_pipeline_publish_gate_fn publish_gate; + void *publish_gate_ctx; /* Test-only: sees the complete graph buffer before the dump releases its indexes. */ void (*before_dump_hook)(const cbm_gbuf_t *, void *); void *before_dump_hook_ctx; @@ -897,6 +900,14 @@ void cbm_pipeline_set_persistence(cbm_pipeline_t *p, bool enabled) { } } +void cbm_pipeline_set_publish_gate(cbm_pipeline_t *p, cbm_pipeline_publish_gate_fn gate, + void *context) { + if (p) { + p->publish_gate = gate; + p->publish_gate_ctx = gate ? context : NULL; + } +} + void cbm_pipeline_set_doc_candidates_enabled(cbm_pipeline_t *p, bool enabled) { if (p) { p->doc_candidates_choice = @@ -4763,6 +4774,17 @@ int cbm_pipeline_run(cbm_pipeline_t *p) { return CBM_PIPELINE_ABORT_PRESERVE_DB; } + /* Last point at which the serving database is still untouched: the + * destination preparation below may already move it aside. */ + if (p->publish_gate && !p->publish_gate(p->publish_gate_ctx)) { + cbm_log_warn("pipeline.publish_refused", "path", final_path); + cleanup_staging_db(staging_path); + free(staging_path); + free(final_path); + pipeline_release_test_config(p); + return CBM_PIPELINE_ABORT_PRESERVE_DB; + } + cbm_replacement_prepare_t prepared = {0}; int prepare_rc = prepare_publish_destination(final_path, final_existed, backup_succeeded, &prepared); diff --git a/src/pipeline/pipeline.h b/src/pipeline/pipeline.h index 072674d8d..bbc856b39 100644 --- a/src/pipeline/pipeline.h +++ b/src/pipeline/pipeline.h @@ -56,6 +56,17 @@ void cbm_pipeline_set_persistence(cbm_pipeline_t *p, bool enabled); /* Apply a validated discovery resource policy. The value is copied. */ void cbm_pipeline_set_resource_policy(cbm_pipeline_t *p, const cbm_index_resource_policy_t *policy); +/* Publication gate. cbm_pipeline_run calls it once, after the last + * pre-publication check and immediately before the serving database is + * touched (never for a live no-op, which replaces nothing). Returning false + * aborts the run with CBM_PIPELINE_ABORT_PRESERVE_DB and the previous + * generation stays exactly as it was. A supervised index worker uses it to + * claim publication from its supervisor (see the publication fence in + * mcp/index_supervisor.h). NULL removes the gate. */ +typedef bool (*cbm_pipeline_publish_gate_fn)(void *context); +void cbm_pipeline_set_publish_gate(cbm_pipeline_t *p, cbm_pipeline_publish_gate_fn gate, + void *context); + /* Doc -> code candidates (the doc_candidates setting). Never set, the index * mode decides: on for full, off for moderate and fast. The setter makes an * explicit choice that wins in every mode. Off skips scoring documentation diff --git a/tests/test_index_supervisor.c b/tests/test_index_supervisor.c index bda503238..f3218fdb1 100644 --- a/tests/test_index_supervisor.c +++ b/tests/test_index_supervisor.c @@ -239,18 +239,59 @@ TEST(index_supervisor_worker_argv_requires_exact_build_bound_grammar) { "/tmp/m", CBM_INDEX_WORKER_QUARANTINE_ARG, "/tmp/q", + CBM_INDEX_WORKER_PUBLISH_FENCE_ARG, + "/tmp/f", NULL, }; cbm_index_worker_invocation_t invocation; - ASSERT_EQ(cbm_index_worker_parse_process_argv(16, valid, &invocation), + ASSERT_EQ(cbm_index_worker_parse_process_argv(18, valid, &invocation), CBM_INDEX_WORKER_ARGV_VALID); - ASSERT_EQ(cbm_daemon_process_role(16, valid), CBM_DAEMON_PROCESS_WORKER); + ASSERT_EQ(cbm_daemon_process_role(18, valid), CBM_DAEMON_PROCESS_WORKER); ASSERT_STR_EQ(invocation.args_json, valid[6]); ASSERT_STR_EQ(invocation.response_out, valid[8]); ASSERT_EQ(invocation.memory_budget_bytes, 1024); ASSERT_TRUE(invocation.single_thread); ASSERT_STR_EQ(invocation.marker_file, valid[13]); ASSERT_STR_EQ(invocation.quarantine_file, valid[15]); + ASSERT_STR_EQ(invocation.publish_fence_file, valid[17]); + /* Without a worker limit there is no fence: the shorter grammar stays valid. */ + ASSERT_EQ(cbm_index_worker_parse_process_argv(16, valid, &invocation), + CBM_INDEX_WORKER_ARGV_VALID); + ASSERT_EQ(cbm_daemon_process_role(16, valid), CBM_DAEMON_PROCESS_WORKER); + ASSERT_NULL(invocation.publish_fence_file); + + char *fence_before_quarantine[] = {"test-runner", + "cli", + "--index-worker", + CBM_INDEX_WORKER_BUILD_ARG, + (char *)captured, + "index_repository", + "{}", + "--response-out", + "/tmp/r", + CBM_INDEX_WORKER_PUBLISH_FENCE_ARG, + "/tmp/f", + CBM_INDEX_WORKER_QUARANTINE_ARG, + "/tmp/q", + NULL}; + char *empty_fence[] = {"test-runner", + "cli", + "--index-worker", + CBM_INDEX_WORKER_BUILD_ARG, + (char *)captured, + "index_repository", + "{}", + "--response-out", + "/tmp/r", + CBM_INDEX_WORKER_PUBLISH_FENCE_ARG, + "", + NULL}; + ASSERT_EQ(cbm_index_worker_parse_process_argv(13, fence_before_quarantine, &invocation), + CBM_INDEX_WORKER_ARGV_INVALID); + ASSERT_EQ(cbm_daemon_process_role(13, fence_before_quarantine), CBM_DAEMON_PROCESS_INVALID); + ASSERT_EQ(cbm_index_worker_parse_process_argv(11, empty_fence, &invocation), + CBM_INDEX_WORKER_ARGV_INVALID); + ASSERT_EQ(cbm_daemon_process_role(11, empty_fence), CBM_DAEMON_PROCESS_INVALID); char *missing_build[] = {"test-runner", "cli", "--index-worker", "index_repository", "{}", "--response-out", @@ -1071,12 +1112,15 @@ TEST(index_supervisor_disabled_limits_do_not_probe) { cbm_index_worker_handle_t *handle = NULL; int start_rc = cbm_index_worker_start_with_policy("{\"__cbm_test_worker\":\"clean\"}", 0, &policy, false, NULL, NULL, 0, &handle); + /* Opt-in: no worker limit, no publication fence, no claim. */ + bool fenced = !handle || cbm_index_worker_publish_fence_path_for_testing(handle) != NULL; const cbm_index_worker_result_t *result = NULL; bool terminal = handle && index_supervisor_test_poll_terminal( handle, INDEX_SUPERVISOR_TEST_TERMINAL_MS, &result); cbm_index_resource_t resource = result ? result->resource_violation.resource : CBM_INDEX_RESOURCE_RSS_BYTES; bool probe_failed = result && result->resource_probe_failed; + bool committed = !result || result->publication_committed; if (terminal) { cbm_index_worker_destroy(handle); } else { @@ -1090,6 +1134,8 @@ TEST(index_supervisor_disabled_limits_do_not_probe) { ASSERT_EQ(fake.clock_calls, 0); ASSERT_EQ(resource, CBM_INDEX_RESOURCE_NONE); ASSERT_FALSE(probe_failed); + ASSERT_FALSE(fenced); + ASSERT_FALSE(committed); PASS(); } @@ -1352,6 +1398,179 @@ TEST(index_supervisor_three_failed_rss_probes_fail_closed) { PASS(); } +/* Publication fence (index_supervisor.h). The hang-tree probe reports the fence + * path it received on its argv in its marker, and the test plays the worker's + * publication claim through the production entry point with that path. */ +typedef struct { + char directory[INDEX_SUPERVISOR_TEST_PATH_CAP]; + char marker[INDEX_SUPERVISOR_TEST_PATH_CAP]; + char fence[INDEX_SUPERVISOR_TEST_PATH_CAP]; +} index_supervisor_fence_fixture_t; + +static bool index_supervisor_fence_fixture_init(index_supervisor_fence_fixture_t *fixture) { + memset(fixture, 0, sizeof(*fixture)); + (void)snprintf(fixture->directory, sizeof(fixture->directory), "%s/cbm-index-fence-XXXXXX", + cbm_tmpdir()); + if (!cbm_mkdtemp(fixture->directory)) { + return false; + } + (void)snprintf(fixture->marker, sizeof(fixture->marker), "%s/marker", fixture->directory); + return true; +} + +/* True once the worker started with exactly this attempt's fence. */ +static bool index_supervisor_fence_fixture_started(index_supervisor_fence_fixture_t *fixture, + cbm_index_worker_handle_t *handle) { + const char *fence = handle ? cbm_index_worker_publish_fence_path_for_testing(handle) : NULL; + if (!fence) { + return false; + } + (void)snprintf(fixture->fence, sizeof(fixture->fence), "%s", fence); + char expected[INDEX_SUPERVISOR_TEST_PATH_CAP + 16]; + (void)snprintf(expected, sizeof(expected), "fence=%s\n", fixture->fence); + return index_supervisor_test_wait_file_text(fixture->marker, expected, + INDEX_SUPERVISOR_TEST_READY_MS); +} + +static bool index_supervisor_fence_fixture_worker_claims( + index_supervisor_fence_fixture_t *fixture) { + cbm_index_set_worker_publish_fence(fixture->fence); + bool claimed = cbm_index_worker_claim_publication(); + cbm_index_set_worker_publish_fence(NULL); + return claimed; +} + +static void index_supervisor_fence_fixture_cleanup(index_supervisor_fence_fixture_t *fixture) { + (void)cbm_unlink(fixture->marker); + (void)cbm_rmdir(fixture->directory); +} + +static int g_index_supervisor_after_publication_logs; + +static void index_supervisor_test_after_publication_sink(const char *line) { + if (line && strstr(line, "index.supervisor.resource_limit_after_publication")) { + g_index_supervisor_after_publication_logs++; + } +} + +TEST(index_supervisor_limit_claims_publication_before_terminating) { + index_supervisor_resource_fake_t fake = { + .now_ms = 100, + .rss_status = CBM_PROC_TREE_RSS_EMPTY, + }; + index_supervisor_fence_fixture_t fixture; + bool fixture_ready = index_supervisor_fence_fixture_init(&fixture); + cbm_index_supervisor_set_resource_hooks_for_testing(index_supervisor_fake_clock, + index_supervisor_fake_charged_rss, &fake); + cbm_index_resource_policy_t policy = index_supervisor_test_worker_policy(0, 1000); + cbm_index_worker_handle_t *handle = NULL; + int start_rc = + fixture_ready + ? cbm_index_worker_start_with_policy("{\"__cbm_test_worker\":\"hang-tree\"}", 0, + &policy, false, fixture.marker, NULL, 100, &handle) + : -1; + bool started = start_rc == 0 && index_supervisor_fence_fixture_started(&fixture, handle); + /* The limit is crossed while the worker has not claimed publication: the + * supervisor claims the fence in this poll, before it stops the worker. */ + fake.now_ms = 1101; + const cbm_index_worker_result_t *result = NULL; + cbm_index_worker_poll_t decided = + handle ? cbm_index_worker_poll(handle, &result) : CBM_INDEX_WORKER_POLL_ERROR; + /* Whatever the worker does next, it can no longer publish. */ + bool worker_barred = started && !index_supervisor_fence_fixture_worker_claims(&fixture); + bool terminal = handle && index_supervisor_test_poll_terminal( + handle, INDEX_SUPERVISOR_TEST_TERMINAL_MS, &result); + bool limited = terminal && result && + result->resource_violation.resource == CBM_INDEX_RESOURCE_DURATION_MS && + result->resource_violation.observed == 1001 && !result->publication_committed && + !result->cancellation_requested && result->tree_quiesced; + bool fence_removed = terminal && fixture.fence[0] && cbm_file_size(fixture.fence) < 0; + if (terminal) { + cbm_index_worker_destroy(handle); + } else { + index_supervisor_test_cleanup_handle(handle); + } + cbm_index_supervisor_reset_resource_hooks_for_testing(); + if (!started) { + index_supervisor_test_dump("marker", fixture.marker); + } + index_supervisor_fence_fixture_cleanup(&fixture); + + ASSERT_TRUE(fixture_ready); + ASSERT_EQ(start_rc, 0); + ASSERT_TRUE(started); + ASSERT_EQ(decided, CBM_INDEX_WORKER_POLL_RUNNING); + ASSERT_TRUE(worker_barred); + ASSERT_TRUE(limited); + ASSERT_TRUE(fence_removed); + PASS(); +} + +TEST(index_supervisor_limit_after_worker_publication_claim_is_not_enforced) { + index_supervisor_resource_fake_t fake = { + .now_ms = 100, + .rss_status = CBM_PROC_TREE_RSS_EMPTY, + }; + index_supervisor_fence_fixture_t fixture; + bool fixture_ready = index_supervisor_fence_fixture_init(&fixture); + CBMLogLevel saved_level = cbm_log_get_level(); + cbm_log_set_level(CBM_LOG_INFO); + g_index_supervisor_after_publication_logs = 0; + cbm_log_set_sink_ex(index_supervisor_test_after_publication_sink, CBM_LOG_SINK_TEE); + cbm_index_supervisor_set_resource_hooks_for_testing(index_supervisor_fake_clock, + index_supervisor_fake_charged_rss, &fake); + cbm_index_resource_policy_t policy = index_supervisor_test_worker_policy(0, 1000); + cbm_index_worker_handle_t *handle = NULL; + int start_rc = + fixture_ready + ? cbm_index_worker_start_with_policy("{\"__cbm_test_worker\":\"hang-tree\"}", 0, + &policy, false, fixture.marker, NULL, 100, &handle) + : -1; + bool started = start_rc == 0 && index_supervisor_fence_fixture_started(&fixture, handle); + /* The worker commits to publishing first; only then is the limit crossed. */ + bool worker_claimed = started && index_supervisor_fence_fixture_worker_claims(&fixture); + fake.now_ms = 1101; + const cbm_index_worker_result_t *result = NULL; + cbm_index_worker_poll_t decided = + handle ? cbm_index_worker_poll(handle, &result) : CBM_INDEX_WORKER_POLL_ERROR; + int clock_calls_at_decision = fake.clock_calls; + cbm_index_worker_poll_t after = + handle ? cbm_index_worker_poll(handle, &result) : CBM_INDEX_WORKER_POLL_ERROR; + bool disarmed = fake.clock_calls == clock_calls_at_decision; + /* The attempt runs on; the test ends it with an ordinary cancellation. */ + bool cancel_accepted = handle && cbm_index_worker_request_cancel(handle); + bool terminal = handle && index_supervisor_test_poll_terminal( + handle, INDEX_SUPERVISOR_TEST_TERMINAL_MS, &result); + bool committed = terminal && result && result->cancellation_requested && + result->resource_violation.resource == CBM_INDEX_RESOURCE_NONE && + !result->resource_probe_failed && result->publication_committed && + result->tree_quiesced; + if (terminal) { + cbm_index_worker_destroy(handle); + } else { + index_supervisor_test_cleanup_handle(handle); + } + cbm_index_supervisor_reset_resource_hooks_for_testing(); + cbm_log_set_sink(NULL); + cbm_log_set_level(saved_level); + if (!started) { + index_supervisor_test_dump("marker", fixture.marker); + } + index_supervisor_fence_fixture_cleanup(&fixture); + + ASSERT_TRUE(fixture_ready); + ASSERT_EQ(start_rc, 0); + ASSERT_TRUE(started); + ASSERT_TRUE(worker_claimed); + ASSERT_EQ(decided, CBM_INDEX_WORKER_POLL_RUNNING); + ASSERT_EQ(after, CBM_INDEX_WORKER_POLL_RUNNING); + ASSERT_TRUE(disarmed); + ASSERT_EQ(g_index_supervisor_after_publication_logs, 1); + ASSERT_TRUE(cancel_accepted); + ASSERT_TRUE(committed); + PASS(); +} + SUITE(index_supervisor) { RUN_TEST(index_supervisor_job_memory_limit_has_floor_headroom_and_no_overflow); RUN_TEST(index_supervisor_worker_argv_requires_exact_build_bound_grammar); @@ -1370,4 +1589,6 @@ SUITE(index_supervisor) { RUN_TEST(index_supervisor_quiet_timeout_remains_hang_with_duration_enabled); RUN_TEST(index_supervisor_cancel_precedes_resource_probe); RUN_TEST(index_supervisor_three_failed_rss_probes_fail_closed); + RUN_TEST(index_supervisor_limit_claims_publication_before_terminating); + RUN_TEST(index_supervisor_limit_after_worker_publication_claim_is_not_enforced); } diff --git a/tests/test_main.c b/tests/test_main.c index 23b2621ee..652893ce1 100644 --- a/tests/test_main.c +++ b/tests/test_main.c @@ -455,10 +455,12 @@ static void tf_index_worker_probe(const char *args_json, const char *response_ou FILE *ready = marker ? cbm_fopen(marker, "wb") : NULL; if (ready) { (void)fprintf( - ready, "single=%s\nmarker=%s\nquarantine=%s\nbudget=%zu\ndescendant=%ld\n", + ready, + "single=%s\nmarker=%s\nquarantine=%s\nbudget=%zu\ndescendant=%ld\nfence=%s\n", getenv("CBM_INDEX_SINGLE_THREAD") ? getenv("CBM_INDEX_SINGLE_THREAD") : "", marker, getenv("CBM_INDEX_QUARANTINE_FILE") ? getenv("CBM_INDEX_QUARANTINE_FILE") : "", - cbm_mem_budget(), descendant); + cbm_mem_budget(), descendant, + cbm_index_worker_publish_fence() ? cbm_index_worker_publish_fence() : ""); (void)fclose(ready); } (void)fprintf(stderr, "async worker hang-tree probe\n"); @@ -509,6 +511,7 @@ static int tf_maybe_run_index_worker(int argc, char **argv) { cbm_index_set_worker_role_options(true, invocation.response_out, invocation.single_thread, invocation.marker_file, invocation.quarantine_file, invocation.memory_budget_bytes); + cbm_index_set_worker_publish_fence(invocation.publish_fence_file); cbm_mem_init_with_cap(0.5, invocation.memory_budget_bytes); cbm_log_init_for_process(false, true); tf_index_worker_probe(invocation.args_json, invocation.response_out); diff --git a/tests/test_mcp.c b/tests/test_mcp.c index 3bfcb018f..7b4c69d91 100644 --- a/tests/test_mcp.c +++ b/tests/test_mcp.c @@ -21677,6 +21677,289 @@ TEST(index_worker_resource_limit_is_trusted_structured_and_not_retried) { #endif } +#ifndef _WIN32 +#include + +/* A worker limit must never be reported after the same attempt published. + * The real worker replaces the serving database and only then exports the + * persistence artifact. A FIFO planted at the artifact's temp path holds it + * inside that export, after the rename, for as long as nobody reads the FIFO, + * and the duration clock crosses its limit only once the serving database has + * been replaced. The limit is therefore observed in the post-publication + * window by construction, never by timing: + * - uncoordinated, the supervisor stops the worker there and answers + * resource_limit_exceeded with serving_index_preserved=true although the + * previous generation is already gone; + * - with the publication fence the worker claimed publication before the + * rename, so the supervisor leaves the limit unenforced and logs that. + * Only that log line releases the FIFO; the worker then finishes and + * reports the published index. The artifact path is a directory, so the + * export itself fails afterwards, which the #1665 response names. */ +typedef struct { + char live_db[CBM_SZ_1K]; + ino_t previous_inode; + int over_limit_calls; +} index_publication_race_clock_t; + +static char g_publication_race_fifo[CBM_SZ_1K]; +static bool g_publication_race_released; + +static uint64_t index_publication_race_clock(void *context) { + index_publication_race_clock_t *clock = context; + struct stat status; + if (stat(clock->live_db, &status) != 0 || status.st_ino == clock->previous_inode) { + return 100; + } + clock->over_limit_calls++; + return 1101; +} + +static void index_publication_race_sink(const char *line) { + if (g_publication_race_released || !line || + !strstr(line, "index.supervisor.resource_limit_after_publication")) { + return; + } + g_publication_race_released = true; + /* Blocks only until the held worker opens its end, then reads to EOF. */ + int fifo = open(g_publication_race_fifo, O_RDONLY); + if (fifo >= 0) { + char buffer[CBM_SZ_4K]; + ssize_t got = 0; + do { + got = read(fifo, buffer, sizeof(buffer)); + } while (got > 0); + (void)close(fifo); + } +} + +enum { + IDXPUB_OK = 0, + IDXPUB_SETUP = 81, + IDXPUB_FIRST_INDEX = 82, + IDXPUB_HOLD_SETUP = 83, + IDXPUB_NO_RESPONSE = 84, + IDXPUB_NOT_REPLACED = 85, + IDXPUB_NOT_CROSSED = 86, + IDXPUB_LIMIT_AFTER_PUBLICATION = 87, + IDXPUB_NOT_RELEASED = 88, + IDXPUB_NOT_REPORTED_PUBLISHED = 89, +}; + +static int index_publication_race_first_generation(cbm_mcp_server_t *server, const char *repo, + const char *cache, + index_publication_race_clock_t *clock) { + char args[CBM_SZ_2K]; + (void)snprintf(args, sizeof(args), "{\"repo_path\":\"%s\",\"mode\":\"fast\"}", repo); + char *response = cbm_mcp_handle_tool(server, "index_repository", args); + char *text = extract_text_content(response); + yyjson_doc *doc = text ? yyjson_read(text, strlen(text), 0) : NULL; + yyjson_val *root = doc ? yyjson_doc_get_root(doc) : NULL; + const char *status = yyjson_get_str(yyjson_obj_get(root, "status")); + const char *project = yyjson_get_str(yyjson_obj_get(root, "project")); + struct stat live; + bool indexed = status && strcmp(status, "indexed") == 0 && project && + snprintf(clock->live_db, sizeof(clock->live_db), "%s/%s.db", cache, project) < + (int)sizeof(clock->live_db) && + stat(clock->live_db, &live) == 0; + if (indexed) { + clock->previous_inode = live.st_ino; + } + yyjson_doc_free(doc); + free(text); + free(response); + return indexed ? IDXPUB_OK : IDXPUB_FIRST_INDEX; +} + +static int index_publication_race_check(const char *repo, const char *cache) { + (void)cbm_setenv("CBM_CACHE_DIR", cache, 1); + cbm_unsetenv("CBM_INDEX_SUPERVISOR"); + cbm_index_supervisor_mark_host(); + cbm_config_t *config = cbm_config_open(cache); + cbm_mcp_server_t *server = cbm_mcp_server_new(NULL); + if (!config || !server) { + cbm_mcp_server_free(server); + cbm_config_close(config); + return IDXPUB_SETUP; + } + cbm_mcp_server_set_config(server, config); + index_publication_race_clock_t clock = {0}; + int result = index_publication_race_first_generation(server, repo, cache, &clock); + + /* A second generation to publish, the hold, and a 1 s duration limit. */ + char art_dir[CBM_SZ_1K]; + char art_path[CBM_SZ_1K]; + (void)snprintf(art_dir, sizeof(art_dir), "%s/.codebase-memory", repo); + (void)snprintf(art_path, sizeof(art_path), "%s/graph.db.zst", art_dir); + (void)snprintf(g_publication_race_fifo, sizeof(g_publication_race_fifo), "%s.tmp", art_path); + if (result == IDXPUB_OK && + (th_write_file(TH_PATH(repo, "extra.c"), "int extra(void) { return 1; }\n") != 0 || + mkdir(art_dir, 0700) != 0 || mkdir(art_path, 0700) != 0 || + mkfifo(g_publication_race_fifo, 0600) != 0 || + cbm_config_set(config, CBM_INDEX_CONFIG_MAX_DURATION_SECONDS, "1") != 0)) { + result = IDXPUB_HOLD_SETUP; + } + char *response = NULL; + if (result == IDXPUB_OK) { + char args[CBM_SZ_2K]; + (void)snprintf(args, sizeof(args), + "{\"repo_path\":\"%s\",\"mode\":\"fast\",\"persistence\":true}", repo); + cbm_log_set_level(CBM_LOG_WARN); + cbm_log_set_sink_ex(index_publication_race_sink, CBM_LOG_SINK_TEE); + cbm_index_supervisor_set_resource_hooks_for_testing(index_publication_race_clock, NULL, + &clock); + response = cbm_mcp_handle_tool(server, "index_repository", args); + cbm_index_supervisor_reset_resource_hooks_for_testing(); + cbm_log_set_sink(NULL); + } + struct stat live; + bool replaced = stat(clock.live_db, &live) == 0 && live.st_ino != clock.previous_inode; + if (result != IDXPUB_OK) { + /* setup failure already recorded */ + } else if (!response) { + result = IDXPUB_NO_RESPONSE; + } else if (!replaced) { + result = IDXPUB_NOT_REPLACED; + } else if (clock.over_limit_calls == 0) { + result = IDXPUB_NOT_CROSSED; + } else if (strstr(response, "resource_limit_exceeded")) { + result = IDXPUB_LIMIT_AFTER_PUBLICATION; + } else if (!g_publication_race_released) { + result = IDXPUB_NOT_RELEASED; + } else if (!strstr(response, "Index database was published")) { + result = IDXPUB_NOT_REPORTED_PUBLISHED; + } + if (result != IDXPUB_OK && response) { + printf(" second response: %.600s\n", response); + fflush(stdout); + } + free(response); + cbm_mcp_server_free(server); + cbm_config_close(config); + return result; +} +#endif + +TEST(index_worker_limit_never_reported_after_its_own_publication) { +#ifdef _WIN32 + SKIP_PLATFORM("fork + FIFO hold harness; the fence state machine is cross-platform and " + "covered by the index_supervisor suite"); +#else + /* th_mktempdir returns one static buffer: keep each path in its own copy. */ + const char *made = th_mktempdir("cbm_publication_race_repo"); + char *repo = made ? strdup(made) : NULL; + made = th_mktempdir("cbm_publication_race_cache"); + char *cache = made ? strdup(made) : NULL; + ASSERT_NOT_NULL(repo); + ASSERT_NOT_NULL(cache); + ASSERT_EQ(th_write_file(TH_PATH(repo, "main.c"), "int main(void) { return 0; }\n"), 0); + fflush(NULL); + pid_t pid = fork(); + if (pid == 0) { + /* Hang guard only: two supervised runs that each start an ASan worker. */ + alarm(180); + _exit(index_publication_race_check(repo, cache)); + } + int status = 0; + bool waited = waitpid(pid, &status, 0) == pid; + int exit_code = waited && WIFEXITED(status) ? WEXITSTATUS(status) : -1; + int signal_code = waited && WIFSIGNALED(status) ? WTERMSIG(status) : 0; + th_cleanup(repo); + th_cleanup(cache); + free(repo); + free(cache); + if (exit_code != IDXPUB_OK) { + printf(" publication race child exit=%d signal=%d\n", exit_code, signal_code); + } + ASSERT_TRUE(waited); + ASSERT_EQ(signal_code, 0); + ASSERT_EQ(exit_code, IDXPUB_OK); + PASS(); +#endif +} + +/* serving_index_preserved describes the request, not only its last attempt: + * an earlier attempt may have published before it crashed (recovery then runs + * another attempt), and an index that appeared during the request (artifact + * bootstrap) was not preserved. */ +TEST(index_worker_resource_response_reports_preservation_for_the_request) { + const char *made = th_mktempdir("cbm_request_facts_repo"); + char *repo = made ? strdup(made) : NULL; + made = th_mktempdir("cbm_request_facts_cache"); + char *cache = made ? strdup(made) : NULL; + ASSERT_NOT_NULL(repo); + ASSERT_NOT_NULL(cache); + ASSERT_EQ(th_write_file(TH_PATH(repo, "main.c"), "int main(void) { return 0; }\n"), 0); + const char *old_cache = getenv("CBM_CACHE_DIR"); + char *saved_cache = old_cache ? strdup(old_cache) : NULL; + (void)cbm_setenv("CBM_CACHE_DIR", cache, 1); + char args[CBM_SZ_2K]; + (void)snprintf(args, sizeof(args), "{\"repo_path\":\"%s\",\"name\":\"RequestFactsFixture\"}", + repo); + + cbm_mcp_index_request_facts_t before_index = {0}; + cbm_mcp_index_request_facts_begin(args, &before_index); + cbm_mcp_server_t *server = cbm_mcp_server_new(NULL); + char *indexed = server ? cbm_mcp_handle_tool(server, "index_repository", args) : NULL; + cbm_mcp_server_free(server); + cbm_mcp_index_request_facts_t facts = {0}; + cbm_mcp_index_request_facts_begin(args, &facts); + + cbm_index_worker_result_t limited; + memset(&limited, 0, sizeof(limited)); + limited.outcome = CBM_PROC_KILLED; + limited.tree_quiesced = true; + limited.resource_violation = (cbm_index_resource_violation_t){ + .resource = CBM_INDEX_RESOURCE_DURATION_MS, .observed = 1001, .limit = 1000}; + cbm_index_worker_result_t published_then_crashed; + memset(&published_then_crashed, 0, sizeof(published_then_crashed)); + published_then_crashed.outcome = CBM_PROC_CRASH; + published_then_crashed.publication_committed = true; + cbm_mcp_index_request_facts_t after_recovery = facts; + cbm_mcp_index_request_facts_record(&after_recovery, &limited); + bool limited_attempt_commits_nothing = !after_recovery.publication_committed; + cbm_mcp_index_request_facts_record(&after_recovery, &published_then_crashed); + + char *preserved = cbm_mcp_index_worker_resource_response(args, &limited, &facts); + char *earlier_published = + cbm_mcp_index_worker_resource_response(args, &limited, &after_recovery); + char *appeared = cbm_mcp_index_worker_resource_response(args, &limited, &before_index); + char *no_facts = cbm_mcp_index_worker_resource_response(args, &limited, NULL); + const char *kept = "\\\"serving_index_preserved\\\":true"; + const char *not_kept = "\\\"serving_index_preserved\\\":false"; + bool indexed_ok = indexed && strstr(indexed, "\\\"status\\\":\\\"indexed\\\""); + bool preserved_ok = preserved && strstr(preserved, kept); + bool earlier_ok = earlier_published && strstr(earlier_published, not_kept); + bool appeared_ok = appeared && strstr(appeared, not_kept); + bool no_facts_ok = no_facts && strstr(no_facts, not_kept); + + free(indexed); + free(preserved); + free(earlier_published); + free(appeared); + free(no_facts); + if (saved_cache) { + (void)cbm_setenv("CBM_CACHE_DIR", saved_cache, 1); + } else { + cbm_unsetenv("CBM_CACHE_DIR"); + } + free(saved_cache); + th_cleanup(repo); + th_cleanup(cache); + free(repo); + free(cache); + + ASSERT_FALSE(before_index.serving_index_existed); + ASSERT_TRUE(indexed_ok); + ASSERT_TRUE(facts.serving_index_existed); + ASSERT_FALSE(facts.publication_committed); + ASSERT_TRUE(limited_attempt_commits_nothing); + ASSERT_TRUE(after_recovery.publication_committed); + ASSERT_TRUE(preserved_ok); + ASSERT_TRUE(earlier_ok); + ASSERT_TRUE(appeared_ok); + ASSERT_TRUE(no_facts_ok); + PASS(); +} /* Child-side check: index a tiny fixture and verify it ran IN-PROCESS. * Distinct exit codes so the parent can report the exact failure mode. */ @@ -25238,6 +25521,8 @@ SUITE(mcp) { RUN_TEST(index_supervisor_start_failure_is_fail_closed_in_real_host); RUN_TEST(index_supervisor_clean_exit_without_response_response_issue1300); RUN_TEST(index_worker_resource_limit_is_trusted_structured_and_not_retried); + RUN_TEST(index_worker_limit_never_reported_after_its_own_publication); + RUN_TEST(index_worker_resource_response_reports_preservation_for_the_request); RUN_TEST(index_bg_paths_route_through_supervisor_issue832); RUN_TEST(sequential_service_edge_props_are_valid_json_issue898); RUN_TEST(index_second_inprocess_run_survives_issue773); diff --git a/tests/test_pipeline.c b/tests/test_pipeline.c index 2b80f51bd..2441c205e 100644 --- a/tests/test_pipeline.c +++ b/tests/test_pipeline.c @@ -21733,6 +21733,72 @@ TEST(cancelled_incremental_reindex_preserves_committed_db) { PASS(); } +typedef struct { + bool allow; + int calls; +} publish_gate_ctx_t; + +static bool publish_gate_decide(void *context) { + publish_gate_ctx_t *gate = context; + gate->calls++; + return gate->allow; +} + +/* The publication gate (a supervised worker's claim on its publication fence) + * is the last word before the serving database is touched: refused, the run + * stops with the previous generation's file in place; allowed, it publishes + * as before. Asked exactly once either way. */ +TEST(publish_gate_decides_before_the_committed_db_is_touched) { + if (setup_incremental_repo() != 0) { + FAIL("setup failed"); + } + cbm_pipeline_t *p = cbm_pipeline_new(g_incr_tmpdir, g_incr_dbpath, CBM_MODE_FULL); + ASSERT_NOT_NULL(p); + ASSERT_EQ(cbm_pipeline_run(p), 0); + char *project = strdup(cbm_pipeline_project_name(p)); + cbm_pipeline_free(p); + ASSERT_NOT_NULL(project); + struct stat baseline; + ASSERT_EQ(stat(g_incr_dbpath, &baseline), 0); + + ASSERT_TRUE(write_go_file(g_incr_tmpdir, "gate_candidate.go", + "package main\n\nfunc GateCandidate() int { return 7 }\n")); + publish_gate_ctx_t refused = {.allow = false}; + p = cbm_pipeline_new(g_incr_tmpdir, g_incr_dbpath, CBM_MODE_FULL); + ASSERT_NOT_NULL(p); + cbm_pipeline_set_publish_gate(p, publish_gate_decide, &refused); + int refused_rc = cbm_pipeline_run(p); + cbm_pipeline_free(p); + struct stat after_refusal; + bool same_file = + stat(g_incr_dbpath, &after_refusal) == 0 && after_refusal.st_ino == baseline.st_ino; + cbm_store_t *live = cbm_store_open_path(g_incr_dbpath); + int refused_candidates = live ? count_nodes_named(live, project, "GateCandidate") : -1; + cbm_store_close(live); + + publish_gate_ctx_t allowed = {.allow = true}; + p = cbm_pipeline_new(g_incr_tmpdir, g_incr_dbpath, CBM_MODE_FULL); + ASSERT_NOT_NULL(p); + cbm_pipeline_set_publish_gate(p, publish_gate_decide, &allowed); + int allowed_rc = cbm_pipeline_run(p); + cbm_pipeline_free(p); + live = cbm_store_open_path(g_incr_dbpath); + int allowed_candidates = live ? count_nodes_named(live, project, "GateCandidate") : -1; + cbm_store_close(live); + + free(project); + cleanup_incremental_repo(); + + ASSERT_EQ(refused.calls, 1); + ASSERT_EQ(refused_rc, CBM_PIPELINE_ABORT_PRESERVE_DB); + ASSERT_TRUE(same_file); + ASSERT_EQ(refused_candidates, 0); + ASSERT_EQ(allowed.calls, 1); + ASSERT_EQ(allowed_rc, 0); + ASSERT_GT(allowed_candidates, 0); + PASS(); +} + TEST(backup_failed_publish_failure_preserves_final_sidecars) { if (setup_incremental_repo() != 0) { FAIL("setup failed"); @@ -26421,6 +26487,7 @@ SUITE(pipeline) { RUN_TEST(incremental_new_file_added); RUN_TEST(cancelled_full_reindex_preserves_committed_db); RUN_TEST(cancelled_incremental_reindex_preserves_committed_db); + RUN_TEST(publish_gate_decides_before_the_committed_db_is_touched); RUN_TEST(backup_failed_publish_failure_preserves_final_sidecars); RUN_TEST(backup_failed_rename_failure_preserves_corrupt_main); #ifdef __linux__