diff --git a/src/daemon/application.c b/src/daemon/application.c index d005f9217..03ca80b37 100644 --- a/src/daemon/application.c +++ b/src/daemon/application.c @@ -181,7 +181,7 @@ struct cbm_daemon_application { cbm_daemon_application_update_ops_t update_ops; cbm_project_lock_manager_t *project_locks; size_t physical_job_limit; - size_t worker_memory_budget_bytes; + size_t aggregate_memory_budget_bytes; size_t active_mutations; size_t update_owners; cbm_daemon_application_update_worker_t update_worker; @@ -693,6 +693,8 @@ static bool application_truncate_file(const char *path) { } static bool application_job_cancel_requested(cbm_daemon_application_job_t *job); +static size_t application_worker_memory_slice_locked(cbm_daemon_application_t *application, + size_t *active_jobs_out); static char **application_read_suspects(cbm_daemon_application_job_t *job, const char *path, int *count_out, bool *cancelled_out) { @@ -1088,11 +1090,33 @@ static application_attempt_status_t application_job_run_attempt(cbm_daemon_appli return APPLICATION_ATTEMPT_CANCELLED; } + /* Decision 2 (#1997 #832): the slice is decided at spawn time from the jobs + * active right now (this job included), so a lone worker may use the whole + * aggregate and concurrent jobs split it. The divisor is logged so a + * fail-whole verdict can be read against the budget the worker really had. */ + size_t active_jobs = 0; + cbm_mutex_lock(&application->mutex); + size_t memory_budget_bytes = application_worker_memory_slice_locked(application, &active_jobs); + size_t aggregate_memory_budget_bytes = application->aggregate_memory_budget_bytes; + cbm_mutex_unlock(&application->mutex); + if (memory_budget_bytes > 0) { + char active_text[32]; + char aggregate_text[32]; + char slice_text[32]; + (void)snprintf(active_text, sizeof(active_text), "%zu", active_jobs); + (void)snprintf(aggregate_text, sizeof(aggregate_text), "%zu", + aggregate_memory_budget_bytes / (1024U * 1024U)); + (void)snprintf(slice_text, sizeof(slice_text), "%zu", + memory_budget_bytes / (1024U * 1024U)); + cbm_log_info("daemon.index.worker_budget", "project", job->project_key, "active_jobs", + active_text, "aggregate_mb", aggregate_text, "slice_mb", slice_text); + } + cbm_daemon_application_worker_t worker = NULL; application_tmp_lock(); - int start_result = application->worker_ops.start( - application->worker_ops.context, job->args_json, application->worker_memory_budget_bytes, - marker_path, quarantine_path, &worker); + int start_result = + application->worker_ops.start(application->worker_ops.context, job->args_json, + memory_budget_bytes, marker_path, quarantine_path, &worker); application_tmp_unlock(); if (start_result != 0 || !worker) { return application_job_cancel_requested(job) ? APPLICATION_ATTEMPT_CANCELLED @@ -1564,6 +1588,22 @@ static size_t application_active_job_count_locked(cbm_daemon_application_t *appl return count; } +/* Decision 2 (#1997 #832): a worker's memory slice is the aggregate budget + * divided by the jobs active at spawn time. The job being spawned is already + * in the table, so the divisor never drops below one; a zero aggregate means + * "no cap" and the worker falls back to its own RAM-fraction budget. */ +static size_t application_worker_memory_slice_locked(cbm_daemon_application_t *application, + size_t *active_jobs_out) { + size_t active = application_active_job_count_locked(application); + if (active == 0) { + active = 1; + } + if (active_jobs_out) { + *active_jobs_out = active; + } + return application->aggregate_memory_budget_bytes / active; +} + /* Compare the effective index request, not its JSON spelling. yyjson's deep * equality treats object member order as insignificant, while the small * normalization below removes values that the index handler interprets as @@ -2948,18 +2988,19 @@ cbm_daemon_application_t *cbm_daemon_application_new( application->ui_readiness_secret_set = true; } } - /* Equal fixed slices keep admission deterministic: starting fewer jobs does - * not let an early worker claim memory reserved for later concurrent jobs. - * The absurd sub-byte-per-slot case is made safe by reducing effective - * capacity before division; normal daemon budgets are many orders larger. */ + /* The per-worker slice is decided at spawn time (see + * application_worker_memory_slice_locked): the aggregate divided by the + * jobs active then, so a lone worker may use the whole aggregate and + * concurrent jobs split it (decision 2, #1997 #832) — the former fixed + * aggregate/limit slice starved a lone job on hosts like #1864. The absurd + * sub-byte-per-slot case is still made safe by reducing effective capacity + * so every admitted job can receive at least one byte; normal daemon + * budgets are many orders larger. */ if (aggregate_memory_budget_bytes > 0 && application->physical_job_limit > aggregate_memory_budget_bytes) { application->physical_job_limit = aggregate_memory_budget_bytes; } - if (aggregate_memory_budget_bytes > 0 && application->physical_job_limit > 0) { - application->worker_memory_budget_bytes = - aggregate_memory_budget_bytes / application->physical_job_limit; - } + application->aggregate_memory_budget_bytes = aggregate_memory_budget_bytes; if (!application->worker_ops.start) { application->worker_ops = (cbm_daemon_application_worker_ops_t){ .context = NULL, @@ -3558,7 +3599,7 @@ size_t cbm_daemon_application_worker_memory_budget_bytes(cbm_daemon_application_ return 0; } cbm_mutex_lock(&application->mutex); - size_t budget = application->worker_memory_budget_bytes; + size_t budget = application_worker_memory_slice_locked(application, NULL); cbm_mutex_unlock(&application->mutex); return budget; } diff --git a/src/mcp/mcp.c b/src/mcp/mcp.c index 7aa40605c..1b0bce021 100644 --- a/src/mcp/mcp.c +++ b/src/mcp/mcp.c @@ -11209,6 +11209,49 @@ static char *handle_index_repository(cbm_mcp_server_t *srv, const char *args) { if (cbm_pipeline_had_format_migration(p)) { yyjson_mut_obj_add_bool(doc, root, "format_migration", true); } + } else if (rc == CBM_PIPELINE_ABORT_OVER_BUDGET) { + /* Decision A (#1997 #832): resident memory stayed above the budget + * after back-pressure and one confirmation cycle, so the run stopped + * before publication. Name the cause, the numbers and the fact that + * the previous index still serves — the generic "check repo_path" + * hint below sent people debugging a path that was fine. */ + int budget_mb = (int)(cbm_mem_budget() / (1024 * 1024)); + int peak_rss_mb = (int)(cbm_mem_peak_rss() / (1024 * 1024)); + char budget_text[CBM_SZ_32]; + char peak_text[CBM_SZ_32]; + (void)snprintf(budget_text, sizeof(budget_text), "%d", budget_mb); + (void)snprintf(peak_text, sizeof(peak_text), "%d", peak_rss_mb); + cbm_log_error("index.abort", "reason", "over_memory_budget", "project", project_name, + "budget_mb", budget_text, "peak_rss_mb", peak_text); + yyjson_mut_obj_add_str(doc, root, "status", "error"); + yyjson_mut_obj_add_str(doc, root, "reason", "over_memory_budget"); + yyjson_mut_obj_add_str(doc, root, "previous_index", "preserved"); + yyjson_mut_obj_add_int(doc, root, "budget_mb", budget_mb); + yyjson_mut_obj_add_int(doc, root, "peak_rss_mb", peak_rss_mb); + yyjson_mut_obj_add_str(doc, root, "hint", + "Indexing stopped: resident memory stayed above the budget after " + "backpressure; no partial graph was published and the previous " + "index still serves. Raise CBM_MEM_BUDGET_MB, lower CBM_WORKERS, " + "or exclude large subtrees."); + } else if (rc == CBM_PIPELINE_ABORT_PRESERVE_DB) { + /* The truthful abort message (#2020): the old generic "check repo_path" + * hint sent people debugging a path that was fine, when the run + * aborted pre-publication (semantic inputs changed mid-run, or a + * discovery/manifest phase failed transiently) and the previous index + * is intact. A 99-test cascade on the Windows leg traced back to + * exactly this — the response said error, but not which kind. */ + yyjson_mut_obj_add_str(doc, root, "status", "aborted_previous_preserved"); + yyjson_mut_obj_add_str(doc, root, "hint", + "Indexing aborted before publication; the previous index is " + "intact and still serving. Causes: files changed while the run " + "was in flight, or a discovery/manifest phase failed " + "transiently. Retry; if it repeats, check the run log."); + } else if (rc == CBM_PIPELINE_PERSIST_FAILED) { + yyjson_mut_obj_add_str(doc, root, "status", "persist_failed"); + yyjson_mut_obj_add_str(doc, root, "hint", + "The validated staging database could not be published. Check " + "free disk space and permissions on the cache directory; the " + "previous index may have been rolled back."); } else { yyjson_mut_obj_add_str(doc, root, "status", "error"); yyjson_mut_obj_add_str(doc, root, "hint", diff --git a/src/pipeline/pass_parallel.c b/src/pipeline/pass_parallel.c index 9f572b2b7..759c199f6 100644 --- a/src/pipeline/pass_parallel.c +++ b/src/pipeline/pass_parallel.c @@ -30,7 +30,9 @@ enum { /* Extraction memory back-pressure: when the process is over its RSS budget, * a worker reclaims + naps before pulling another file so peers can finish * and return pages. Bounded spins avoid deadlock when the resident graph - * itself is near budget (then proceed with a soft overshoot). */ + * itself is near budget: then ONE confirmation cycle decides — drained + * means continue, still over means the attempt fails whole (decision A, + * #1997 #832) instead of the former open-ended soft overshoot. */ PP_BACKPRESSURE_MAX_SPINS = 40, PP_BACKPRESSURE_NAP_NS = 3000000, /* 3 ms */ }; @@ -667,6 +669,11 @@ typedef struct { * While set, pulls skip the nap (the designed soft overshoot); the cheap * over-budget probe re-arms the gate once RSS drains under budget. */ _Atomic int bp_futile; + /* Decision A (#1997 #832): set once futility was CONFIRMED by a second + * full cycle that still ended over budget. Every pull loop breaks on it + * and extract returns CBM_PIPELINE_ABORT_OVER_BUDGET. Deliberately not + * the shared cancel token — that one belongs to client cancellation. */ + _Atomic int over_budget_abort; const CBMMacroTable *macro_table; /* ObjectScript $$$macros (NULL if none) */ const CBMReturnTypeTable *return_type_table; /* ObjectScript return types (NULL if none) */ @@ -722,6 +729,39 @@ static void log_extract_done(int pos, uint64_t ms, int defs, const char *path) { } } +/* One back-pressure cycle: reclaim this thread's freed pages, then nap in + * bounded 3 ms steps while the process stays over budget. Counted for test + * observability. Returns true when the FULL cycle elapsed still over budget — + * the resident floor, not in-flight transients, holds the memory. */ +static bool pp_backpressure_cycle_still_over(extract_ctx_t *ec) { + cbm_mem_collect(); + atomic_fetch_add_explicit(&g_bp_nap_cycles, SKIP_ONE, memory_order_relaxed); + int bp = 0; + for (; bp < PP_BACKPRESSURE_MAX_SPINS && cbm_mem_over_budget() && + !atomic_load_explicit(ec->cancelled, memory_order_relaxed) && + !atomic_load_explicit(&ec->over_budget_abort, memory_order_relaxed); + bp++) { + struct timespec nap = {0, PP_BACKPRESSURE_NAP_NS}; + cbm_nanosleep(&nap, NULL); + } + return bp == PP_BACKPRESSURE_MAX_SPINS && cbm_mem_over_budget(); +} + +/* Decision A (#1997 #832): back-pressure was futile and the confirmation + * cycle still ended over budget. The budget stops being advisory here: flag + * the attempt so every worker stops pulling and extract returns + * CBM_PIPELINE_ABORT_OVER_BUDGET. Cooperative on purpose — the failure must + * travel as a clean-exit worker response, never a kill, or the supervisor + * would quarantine innocent files. */ +static void pp_fail_whole_over_budget(extract_ctx_t *ec) { + if (atomic_exchange_explicit(&ec->over_budget_abort, 1, memory_order_relaxed) == 0) { + cbm_log_error("mem.budget.exceeded", "rss_mb", + itoa_log((int)(cbm_mem_rss() / (1024 * 1024))), "budget_mb", + itoa_log((int)(cbm_mem_budget() / (1024 * 1024))), "phase", + "parallel_extract", "action", "fail_whole"); + } +} + static void extract_worker(int worker_id, void *ctx_ptr) { extract_ctx_t *ec = ctx_ptr; extract_worker_state_t *ws = &ec->workers[worker_id]; @@ -739,7 +779,8 @@ static void extract_worker(int worker_id, void *ctx_ptr) { break; } cbm_scale_tick(&ec->scale, sort_pos); - if (atomic_load_explicit(ec->cancelled, memory_order_relaxed)) { + if (atomic_load_explicit(ec->cancelled, memory_order_relaxed) || + atomic_load_explicit(&ec->over_budget_abort, memory_order_relaxed)) { break; } @@ -750,39 +791,41 @@ static void extract_worker(int worker_id, void *ctx_ptr) { * near the budget instead of letting all workers parse their biggest * files at once. Self-disabling when the budget is unset (tests) or RSS * is under budget; bounded spins avoid deadlock when the resident graph - * is itself near budget (then proceed with a soft overshoot). + * is itself near budget. * * Futility latch: when a FULL nap cycle ends still over budget, the * resident floor — not transients — holds the memory; napping again on * the next pull cannot reclaim it and only idles workers (linux kernel: - * one full cycle per pull ≈ 390 s at 79% avg CPU). Latch bp_futile and - * proceed with the soft overshoot; the over-budget probe below re-arms - * the gate as soon as RSS drains under budget. */ + * one full cycle per pull ≈ 390 s at 79% avg CPU). The worker that + * latches bp_futile (0→1) runs ONE confirmation cycle while its peers + * skip the nap. A confirmation that ends under budget re-arms the gate + * (RSS drained after all); one that still ends over budget fails the + * attempt whole (decision A, #1997 #832): every worker stops pulling, + * extract returns CBM_PIPELINE_ABORT_OVER_BUDGET, nothing is published + * and the previously serving index keeps answering. */ if (cbm_mem_budget() > 0) { bool over = cbm_mem_over_budget(); bool futile = atomic_load_explicit(&ec->bp_futile, memory_order_relaxed) != 0; if (over && !futile) { - cbm_mem_collect(); - atomic_fetch_add_explicit(&g_bp_nap_cycles, SKIP_ONE, memory_order_relaxed); - int bp = 0; - for (; bp < PP_BACKPRESSURE_MAX_SPINS && cbm_mem_over_budget() && - !atomic_load_explicit(ec->cancelled, memory_order_relaxed); - bp++) { - struct timespec nap = {0, PP_BACKPRESSURE_NAP_NS}; - cbm_nanosleep(&nap, NULL); - } - if (bp == PP_BACKPRESSURE_MAX_SPINS && cbm_mem_over_budget()) { - /* Log only the 0→1 transition: all workers race into the - * gate before anyone latches, so a plain store would WARN - * once per worker (12 lines per latch event). */ - if (atomic_exchange_explicit(&ec->bp_futile, 1, memory_order_relaxed) == 0) { - cbm_log_warn("mem.backpressure.futile", "action", "soft_overshoot"); + /* Act only on the 0→1 transition: all workers race into the + * gate before anyone latches, so a plain store would confirm + * (and WARN) once per worker. */ + if (pp_backpressure_cycle_still_over(ec) && + atomic_exchange_explicit(&ec->bp_futile, 1, memory_order_relaxed) == 0) { + cbm_log_warn("mem.backpressure.futile", "action", "confirm"); + if (pp_backpressure_cycle_still_over(ec)) { + pp_fail_whole_over_budget(ec); + } else { + atomic_store_explicit(&ec->bp_futile, 0, memory_order_relaxed); } } } else if (!over && futile) { atomic_store_explicit(&ec->bp_futile, 0, memory_order_relaxed); } } + if (atomic_load_explicit(&ec->over_budget_abort, memory_order_relaxed)) { + break; + } int file_idx = ec->sorted[sort_pos].idx; const cbm_file_info_t *fi = &ec->files[file_idx]; @@ -1129,6 +1172,7 @@ int cbm_parallel_extract_ex(cbm_pipeline_ctx_t *ctx, const cbm_file_info_t *file atomic_init(&ec.retain_cap_warned, 0); atomic_init(&ec.oversized_warned, 0); atomic_init(&ec.bp_futile, 0); + atomic_init(&ec.over_budget_abort, 0); /* Sub-phase: Dispatch workers (parse + extract per file, PARALLEL) */ CBM_PROF_START(t_dispatch); @@ -1176,6 +1220,12 @@ int cbm_parallel_extract_ex(cbm_pipeline_ctx_t *ctx, const cbm_file_info_t *file free(sorted); cbm_macro_table_free(pp_macro_table); /* ObjectScript macro table (NULL-safe) */ + /* The over-budget verdict outranks the cancel sentinel: a caller must be + * able to name the cause, and the orchestrator discards the staging DB on + * every non-zero code alike (the live generation is never touched). */ + if (atomic_load(&ec.over_budget_abort)) { + return CBM_PIPELINE_ABORT_OVER_BUDGET; + } if (atomic_load(ctx->cancelled)) { return CBM_NOT_FOUND; } diff --git a/src/pipeline/pipeline.h b/src/pipeline/pipeline.h index dfe81f726..954686849 100644 --- a/src/pipeline/pipeline.h +++ b/src/pipeline/pipeline.h @@ -63,9 +63,14 @@ void cbm_pipeline_free(cbm_pipeline_t *p); * need to know whether the PREVIOUS generation survived can distinguish the * failures by value: the run publishes by renaming a fully validated staging * database over the destination, so every abort before that rename leaves the - * existing database in place. Those codes (CBM_PIPELINE_ABORT_PRESERVE_DB and - * CBM_PIPELINE_PERSIST_FAILED) are defined in pipeline_internal.h alongside the - * stages that raise them. */ + * existing database in place. */ +#define CBM_PIPELINE_ABORT_PRESERVE_DB (-2) /* aborted pre-publication; previous DB intact */ +#define CBM_PIPELINE_PERSIST_FAILED (-4) /* staging/rollback failure during publication */ +/* Resident memory stayed above the budget after back-pressure and one + * confirmation cycle: the run stops before publication, no partial graph is + * written, the previous DB is intact (#1997 #832). Distinct from the cancel + * sentinel so callers can name the cause instead of "pipeline failed". */ +#define CBM_PIPELINE_ABORT_OVER_BUDGET (-5) int cbm_pipeline_run(cbm_pipeline_t *p); /* Request cancellation of a running pipeline (thread-safe). */ diff --git a/src/pipeline/pipeline_internal.h b/src/pipeline/pipeline_internal.h index d749c695a..3a6cc1e58 100644 --- a/src/pipeline/pipeline_internal.h +++ b/src/pipeline/pipeline_internal.h @@ -29,12 +29,12 @@ /* Route node QN buffer size (must fit __route__METHOD__/full/url/path) */ #define CBM_ROUTE_QN_SIZE 768 -/* Incremental integrity failure: abort the run and preserve the existing DB. - * Distinct from CBM_NOT_FOUND, which the orchestrator uses as the normal +/* CBM_PIPELINE_ABORT_PRESERVE_DB / CBM_PIPELINE_PERSIST_FAILED moved to + * pipeline.h — callers legitimately distinguish them (the header's contract + * always said so). FORCE_FULL_REINDEX never escapes the orchestrator and + * stays internal. All three are distinct from CBM_NOT_FOUND, the normal * "no incremental route; continue with a full index" sentinel. */ -#define CBM_PIPELINE_ABORT_PRESERVE_DB (-2) #define CBM_PIPELINE_FORCE_FULL_REINDEX (-3) -#define CBM_PIPELINE_PERSIST_FAILED (-4) /* Canonicalize route-path parameter placeholders (":id", "{id}", "", * "${...}") to a single "{}" token so that client call sites and server diff --git a/tests/test_daemon_application.c b/tests/test_daemon_application.c index cea698fda..f587940c2 100644 --- a/tests/test_daemon_application.c +++ b/tests/test_daemon_application.c @@ -5245,12 +5245,157 @@ TEST(daemon_application_default_limit_admits_four_and_rejects_fifth) { ASSERT_EQ(atomic_load(&fake.starts), DEFAULT_CAP_RUNNING); ASSERT_EQ(atomic_load(&fake.cancels), DEFAULT_CAP_RUNNING); ASSERT_EQ(atomic_load(&fake.destroys), DEFAULT_CAP_RUNNING); - size_t assigned_budget = 0; + /* Decision 2 (#1997 #832): each worker's slice is the aggregate divided by + * the jobs active when IT was spawned. Four requests race through + * admission, so which divisor each saw is scheduling-dependent — but every + * slice is one of aggregate/1..4 and never exceeds the aggregate. */ for (int i = 0; i < DEFAULT_CAP_RUNNING; i++) { - ASSERT_EQ(fake.memory_budgets[i], aggregate_budget / DEFAULT_CAP_RUNNING); - assigned_budget += fake.memory_budgets[i]; + bool valid_slice = false; + for (size_t divisor = 1; divisor <= DEFAULT_CAP_RUNNING; divisor++) { + valid_slice = valid_slice || fake.memory_budgets[i] == aggregate_budget / divisor; + } + ASSERT_TRUE(valid_slice); + ASSERT_TRUE(fake.memory_budgets[i] <= aggregate_budget); } - ASSERT_TRUE(assigned_budget <= aggregate_budget); + PASS(); +} + +/* Decision 2 (#1997 #832): the worker slice is aggregate / jobs ACTIVE at + * spawn time. Sequenced admission makes the divisor deterministic: the first + * job spawns alone and receives the whole aggregate; the second spawns while + * the first still runs and receives half. */ +TEST(daemon_application_worker_slice_is_aggregate_over_active_jobs) { + const size_t aggregate_budget = 4099; + app_fake_worker_context_t fake; + app_fake_worker_context_init(&fake); + cbm_daemon_application_worker_ops_t worker_ops = { + .context = &fake, + .start = app_fake_worker_start, + .poll = app_fake_worker_poll, + .cancel = app_fake_worker_cancel, + .log_path = app_fake_worker_log_path, + .destroy = app_fake_worker_destroy, + }; + cbm_daemon_application_config_t config = { + .worker_ops = &worker_ops, + .aggregate_memory_budget_bytes = aggregate_budget, + }; + cbm_daemon_application_t *application = cbm_daemon_application_new(&config); + char roots[2][APP_TEST_PATH_CAP]; + bool roots_ok = true; + for (int i = 0; i < 2; i++) { + (void)snprintf(roots[i], sizeof(roots[i]), "%s/cbm-app-slice-%d-XXXXXX", cbm_tmpdir(), i); + roots_ok = roots_ok && cbm_mkdtemp(roots[i]) != NULL; + } + app_index_thread_t requests[2] = { + {.application = application, .project = "slice-first", .root = roots[0], .result = -1}, + {.application = application, .project = "slice-second", .root = roots[1], .result = -1}, + }; + cbm_thread_t threads[2]; + bool first_started = application && roots_ok && + cbm_thread_create(&threads[0], 0, app_index_thread, &requests[0]) == 0; + bool first_spawned = first_started && app_wait_for_atomic_int(&fake.starts, 1); + bool second_started = + first_spawned && cbm_thread_create(&threads[1], 0, app_index_thread, &requests[1]) == 0; + bool second_spawned = second_started && app_wait_for_atomic_int(&fake.starts, 2); + atomic_store(&fake.allow_completion, true); + if (first_started) { + (void)cbm_thread_join(&threads[0]); + } + if (second_started) { + (void)cbm_thread_join(&threads[1]); + } + bool stopped = application && cbm_daemon_application_shutdown(application, APP_TEST_TIMEOUT_MS); + cbm_daemon_application_free(application); + for (int i = 0; i < 2; i++) { + (void)cbm_rmdir(roots[i]); + } + + ASSERT_TRUE(roots_ok); + ASSERT_TRUE(first_spawned); + ASSERT_TRUE(second_spawned); + ASSERT_EQ(requests[0].result, 0); + ASSERT_EQ(requests[1].result, 0); + ASSERT_TRUE(stopped); + ASSERT_EQ(fake.memory_budgets[0], aggregate_budget); + ASSERT_EQ(fake.memory_budgets[1], aggregate_budget / 2); + PASS(); +} + +/* Decision A (#1997 #832): the over-budget verdict travels as a CLEAN worker + * exit carrying an error response. The daemon passes that response through + * verbatim on the first attempt — no crash/hang recovery loop, no quarantine + * of innocent files — because a healthy process reported an honest failure. */ +TEST(daemon_application_over_budget_response_passes_through_without_recovery) { + static const char over_budget_response[] = + "{\"content\":[{\"type\":\"text\",\"text\":\"{\\\"project\\\":\\\"budget\\\"," + "\\\"status\\\":\\\"error\\\",\\\"reason\\\":\\\"over_memory_budget\\\"," + "\\\"previous_index\\\":\\\"preserved\\\"}\"}],\"isError\":true}"; + app_fake_worker_context_t fake; + app_fake_worker_context_init(&fake); + atomic_store(&fake.scripted, true); + fake.outcomes[0] = CBM_PROC_CLEAN; + fake.responses[0] = over_budget_response; + fake.outcomes[1] = CBM_PROC_CLEAN; /* a second start would be the recovery loop */ + cbm_daemon_application_worker_ops_t worker_ops = { + .context = &fake, + .start = app_fake_worker_start, + .poll = app_fake_worker_poll, + .cancel = app_fake_worker_cancel, + .log_path = app_fake_worker_log_path, + .destroy = app_fake_worker_destroy, + }; + cbm_daemon_application_config_t config = {.worker_ops = &worker_ops}; + cbm_daemon_application_t *application = cbm_daemon_application_new(&config); + cbm_daemon_runtime_application_callbacks_t callbacks = + cbm_daemon_application_runtime_callbacks(application); + cbm_daemon_runtime_application_session_t *session = + application ? app_test_open(&callbacks, 71) : NULL; + char root[APP_TEST_PATH_CAP]; + snprintf(root, sizeof(root), "%s/cbm-app-over-budget-XXXXXX", cbm_tmpdir()); + bool root_ok = cbm_mkdtemp(root) != NULL; + uint8_t *context = NULL; + uint32_t context_length = 0; + char args[APP_TEST_PATH_CAP + 32]; + snprintf(args, sizeof(args), "{\"repo_path\":\"%s\"}", root); + uint8_t *tool = NULL; + uint32_t tool_length = 0; + bool setup = application && session && root_ok && + app_test_context_request(root, root, &context, &context_length) && + app_test_tool_request("index_repository", args, &tool, &tool_length); + uint8_t *response = NULL; + uint32_t response_length = 0; + if (setup) { + setup = app_test_request(&callbacks, session, context, context_length, &response, + &response_length) == CBM_DAEMON_RUNTIME_APPLICATION_OK; + free(response); + response = NULL; + response_length = 0; + } + cbm_daemon_runtime_application_status_t status = + setup + ? app_test_request(&callbacks, session, tool, tool_length, &response, &response_length) + : CBM_DAEMON_RUNTIME_APPLICATION_HANDLER_ERROR; + bool passed_through = response && response_length > 0 && + strstr((char *)response, "over_memory_budget") != NULL && + strstr((char *)response, "\"isError\":true") != NULL; + free(response); + free(context); + free(tool); + if (session) { + callbacks.session_cancel(callbacks.context, session); + callbacks.session_close(callbacks.context, session); + } + bool stopped = application && cbm_daemon_application_shutdown(application, APP_TEST_TIMEOUT_MS); + cbm_daemon_application_free(application); + (void)cbm_rmdir(root); + + ASSERT_TRUE(setup); + ASSERT_EQ(status, CBM_DAEMON_RUNTIME_APPLICATION_OK); + ASSERT_TRUE(passed_through); + ASSERT_EQ(atomic_load(&fake.starts), 1); + ASSERT_EQ(atomic_load(&fake.destroys), 1); + ASSERT_TRUE(stopped); PASS(); } @@ -5458,6 +5603,8 @@ SUITE(daemon_application) { RUN_TEST(daemon_application_thread_start_failure_rolls_back_job_reservation); RUN_TEST(daemon_application_queues_explicit_index_behind_physical_job_limit); RUN_TEST(daemon_application_default_limit_admits_four_and_rejects_fifth); + RUN_TEST(daemon_application_worker_slice_is_aggregate_over_active_jobs); + RUN_TEST(daemon_application_over_budget_response_passes_through_without_recovery); RUN_TEST(daemon_application_free_reports_retained_live_ownership); RUN_TEST(daemon_application_rejects_clean_exit_when_process_tree_is_not_contained); } diff --git a/tests/test_incremental.c b/tests/test_incremental.c index 498d912b9..4374e97f3 100644 --- a/tests/test_incremental.c +++ b/tests/test_incremental.c @@ -107,7 +107,20 @@ static int reformat_files(const char *subdir, int max_files) { static char *index_repo(void) { char args[512]; snprintf(args, sizeof(args), "{\"repo_path\":\"%s\"}", g_repodir); - return cbm_mcp_handle_tool(g_srv, "index_repository", args); + char *resp = cbm_mcp_handle_tool(g_srv, "index_repository", args); + /* Fail-closed at the source: every caller asserts resp != NULL, but an + * error response (e.g. a transient pre-publication abort that preserves + * the previous index) used to slip through and cascade into dozens of + * mysterious count/query failures downstream — a 99-test pile-up on the + * Windows leg traced back to exactly this. Surface the real cause once, + * here, and let the existing non-NULL asserts fail loudly. */ + if (resp && strstr(resp, "\"isError\":true")) { + fprintf(stderr, " index_repo: index_repository returned an error response:\n %.400s\n", + resp); + free(resp); + return NULL; + } + return resp; } /* Timed index: returns response, sets *elapsed_ms and *peak_rss_mb */ diff --git a/tests/test_mcp.c b/tests/test_mcp.c index 6fbaaf66c..10d5b9e6e 100644 --- a/tests/test_mcp.c +++ b/tests/test_mcp.c @@ -9,6 +9,7 @@ #include "../src/foundation/constants.h" #include "../src/foundation/log.h" #include "../src/foundation/platform.h" /* cbm_file_size */ +#include "../src/foundation/mem.h" /* cbm_mem_set_budget_for_tests — over-budget seam */ #include "../src/foundation/subprocess.h" #include "../src/foundation/workspace.h" #include "../src/mcp/compact_out.h" @@ -17670,6 +17671,161 @@ static int idx823_supervised_name_override_check(const char *repo_dir, const cha } #endif +/* Write the 64-file Go fixture (above MIN_FILES_FOR_PARALLEL, so the gated + * parallel extract path runs). generation 2 gives every file a second + * definition so a re-run must re-extract all of them — an unchanged repo is an + * incremental no-op and never reaches the memory gate. */ +static bool budget_fixture_write(const char *repo, int generation) { + for (int i = 0; i < 64; i++) { + char path[CBM_SZ_1K]; + char body[CBM_SZ_256]; + snprintf(path, sizeof(path), "%s/f%02d.go", repo, i); + if (generation == 1) { + snprintf(body, sizeof(body), "package main\n\nfunc F%02d() int {\n\treturn %d\n}\n", i, + i); + } else { + snprintf(body, sizeof(body), + "package main\n\nfunc F%02d() int {\n\treturn %d\n}\n\n" + "func G%02d() int {\n\treturn F%02d() + 1\n}\n", + i, i, i, i); + } + if (th_write_file(path, body) != 0) { + return false; + } + } + return true; +} + +/* Parse the JSON object inside an MCP tool result. NULL when absent. */ +static yyjson_doc *budget_result_doc(const char *result) { + char *text = extract_text_content(result); + yyjson_doc *doc = text ? yyjson_read(text, strlen(text), 0) : NULL; + free(text); + return doc; +} + +static const char *budget_doc_str(yyjson_doc *doc, const char *key) { + return doc ? yyjson_get_str(yyjson_obj_get(yyjson_doc_get_root(doc), key)) : NULL; +} + +static int budget_doc_int(yyjson_doc *doc, const char *key, int missing) { + yyjson_val *val = doc ? yyjson_obj_get(yyjson_doc_get_root(doc), key) : NULL; + return val && yyjson_is_int(val) ? (int)yyjson_get_int(val) : missing; +} + +/* Decision A (#1997 #832): an attempt that stays over the memory budget after + * back-pressure fails WHOLE and the response says so — reason, numbers, hint, + * and that the previous index still serves — instead of the generic "check + * repo_path" error. In-process: the runner is never a marked supervisor host, + * and the budget seam stands in for a 1 MiB machine. */ +TEST(index_repository_over_budget_reports_named_reason) { + char repo[CBM_SZ_1K]; + char cache[CBM_SZ_1K]; + snprintf(repo, sizeof(repo), "%s/cbm-mcp-budget-repo-XXXXXX", cbm_tmpdir()); + snprintf(cache, sizeof(cache), "%s/cbm-mcp-budget-cache-XXXXXX", cbm_tmpdir()); + if (!cbm_mkdtemp(repo)) { + FAIL("cbm_mkdtemp repo failed"); + } + if (!cbm_mkdtemp(cache)) { + th_rmtree(repo); + FAIL("cbm_mkdtemp cache failed"); + } + const char *saved_cache = getenv("CBM_CACHE_DIR"); + char *saved_cache_copy = saved_cache ? strdup(saved_cache) : NULL; + const char *saved_workers = getenv("CBM_WORKERS"); + char *saved_workers_copy = saved_workers ? strdup(saved_workers) : NULL; + cbm_setenv("CBM_CACHE_DIR", cache, 1); + cbm_setenv("CBM_WORKERS", "4", 1); + + bool files_ok = budget_fixture_write(repo, 1); + cbm_mcp_server_t *srv = cbm_mcp_server_new(NULL); + bool had_server = srv != NULL; + char args[CBM_SZ_2K]; + snprintf(args, sizeof(args), "{\"repo_path\":\"%s\"}", repo); + + /* Generation 1 at the process budget publishes. */ + char *first = srv && files_ok ? cbm_mcp_handle_tool(srv, "index_repository", args) : NULL; + yyjson_doc *first_doc = budget_result_doc(first); + const char *first_status = budget_doc_str(first_doc, "status"); + bool first_indexed = first_status && strcmp(first_status, "indexed") == 0; + int nodes_first = budget_doc_int(first_doc, "nodes", -1); + const char *first_project = budget_doc_str(first_doc, "project"); + char *project = first_project ? strdup(first_project) : NULL; + yyjson_doc_free(first_doc); + free(first); + char db_path[CBM_SZ_2K]; + snprintf(db_path, sizeof(db_path), "%s/%s.db", cache, project ? project : "missing"); + long db_size_before = (long)cbm_file_size(db_path); + + /* Generation 2 at 1 MiB over a changed repo fails whole. */ + files_ok = files_ok && budget_fixture_write(repo, 2); + size_t saved_budget = cbm_mem_budget(); + cbm_mem_set_budget_for_tests((size_t)1024 * 1024); + char *second = srv && files_ok ? cbm_mcp_handle_tool(srv, "index_repository", args) : NULL; + cbm_mem_set_budget_for_tests(saved_budget); + bool second_is_error = second && strstr(second, "\"isError\":true") != NULL; + yyjson_doc *second_doc = budget_result_doc(second); + const char *second_status = budget_doc_str(second_doc, "status"); + const char *second_reason = budget_doc_str(second_doc, "reason"); + const char *second_previous = budget_doc_str(second_doc, "previous_index"); + const char *second_hint = budget_doc_str(second_doc, "hint"); + bool status_error = second_status && strcmp(second_status, "error") == 0; + bool reason_named = second_reason && strcmp(second_reason, "over_memory_budget") == 0; + bool previous_preserved = second_previous && strcmp(second_previous, "preserved") == 0; + bool hint_names_knob = second_hint && strstr(second_hint, "CBM_MEM_BUDGET_MB") != NULL; + int budget_mb = budget_doc_int(second_doc, "budget_mb", -1); + int peak_rss_mb = budget_doc_int(second_doc, "peak_rss_mb", -1); + yyjson_doc_free(second_doc); + free(second); + long db_size_after = (long)cbm_file_size(db_path); + + /* The previous generation still answers with its full graph. */ + char status_args[CBM_SZ_2K]; + snprintf(status_args, sizeof(status_args), "{\"project\":\"%s\"}", project ? project : ""); + char *status = srv && project ? cbm_mcp_handle_tool(srv, "index_status", status_args) : NULL; + /* index_status answers in the tree format ("nodes: " on its own line). */ + char *status_text = extract_text_content(status); + const char *nodes_line = status_text ? strstr(status_text, "\nnodes: ") : NULL; + int nodes_after = nodes_line ? (int)strtol(nodes_line + strlen("\nnodes: "), NULL, 10) : -1; + if (nodes_after != nodes_first) { + printf(" index_status after the failed attempt: %.400s\n", + status_text ? status_text : "(null)"); + } + free(status_text); + free(status); + + cbm_mcp_server_free(srv); + cleanup_project_db(cache, project); + free(project); + if (saved_workers_copy) { + cbm_setenv("CBM_WORKERS", saved_workers_copy, 1); + } else { + cbm_unsetenv("CBM_WORKERS"); + } + free(saved_workers_copy); + restore_cache_dir(saved_cache_copy); + free(saved_cache_copy); + th_rmtree(cache); + th_rmtree(repo); + + ASSERT_TRUE(files_ok); + ASSERT_TRUE(had_server); + ASSERT_TRUE(first_indexed); + ASSERT_GT(nodes_first, 0); + ASSERT_TRUE(second_is_error); + ASSERT_TRUE(status_error); + ASSERT_TRUE(reason_named); + ASSERT_TRUE(previous_preserved); + ASSERT_TRUE(hint_names_knob); + ASSERT_EQ(budget_mb, 1); + ASSERT_GT(peak_rss_mb, budget_mb); + /* Preserved on disk and still served: same file size, same node count. */ + ASSERT_GT(db_size_before, 0L); + ASSERT_EQ(db_size_after, db_size_before); + ASSERT_EQ(nodes_after, nodes_first); + PASS(); +} + TEST(index_repository_cli_name_override_issue823) { #ifdef _WIN32 SKIP_PLATFORM("POSIX fork harness required to isolate supervisor host mark"); @@ -20238,6 +20394,7 @@ SUITE(mcp) { RUN_TEST(index_repository_relative_path_uses_explicit_session_root); RUN_TEST(index_repository_supervisor_uses_canonical_session_path); RUN_TEST(index_repository_cli_name_override_issue823); + RUN_TEST(index_repository_over_budget_reports_named_reason); RUN_TEST(index_supervisor_unsafe_clean_is_never_fallback_or_recovery); RUN_TEST(index_supervisor_gate_requires_marked_host_issue845); RUN_TEST(index_supervisor_start_failure_is_fail_closed_in_real_host); diff --git a/tests/test_pipeline.c b/tests/test_pipeline.c index 0f6332cb2..738445fdf 100644 --- a/tests/test_pipeline.c +++ b/tests/test_pipeline.c @@ -12871,23 +12871,82 @@ TEST(pipeline_committed_counts_match_persisted) { * MIN_FILES_FOR_PARALLEL (50) — else the run routes sequential, the gate never * fires, and the test would pass vacuously (cycles==0). The engagement assert * below (cycles >= 1) is a hard guard against that regressing silently. */ -TEST(pipeline_backpressure_futile_nap_disengages) { - /* 64 tiny files: > MIN_FILES_FOR_PARALLEL (50) so the parallel path (and its - * back-pressure gate) actually runs; old-code cycles (~64) >> the bound. */ +/* Shared fixture for the back-pressure tests: 64 tiny Go files, above + * MIN_FILES_FOR_PARALLEL (50) so the parallel extract path — the only phase + * with a memory gate — is the one that runs. */ +static bool write_backpressure_fixture(void) { snprintf(g_tmpdir, sizeof(g_tmpdir), "/tmp/cbm_test_XXXXXX"); if (!cbm_mkdtemp(g_tmpdir)) { - FAIL("failed to create temp dir"); + return false; } for (int i = 0; i < 64; i++) { char path[512]; snprintf(path, sizeof(path), "%s/f%02d.go", g_tmpdir, i); FILE *f = fopen(path, "w"); if (!f) { - FAIL("failed to create fixture file"); + return false; } fprintf(f, "package main\n\nfunc F%02d() int {\n\treturn %d\n}\n", i, i); fclose(f); } + return true; +} + +/* Rewrite every fixture file with a second definition. An unchanged repo + * routes incremental_manifest → incremental.noop (nothing is re-extracted, so + * no gate can fire); changing all 64 files makes the next run re-extract them + * all through the parallel path. */ +static bool mutate_backpressure_fixture(void) { + for (int i = 0; i < 64; i++) { + char path[512]; + snprintf(path, sizeof(path), "%s/f%02d.go", g_tmpdir, i); + FILE *f = fopen(path, "w"); + if (!f) { + return false; + } + fprintf(f, + "package main\n\nfunc F%02d() int {\n\treturn %d\n}\n\n" + "func G%02d() int {\n\treturn F%02d() + 1\n}\n", + i, i, i, i); + fclose(f); + } + return true; +} + +/* Whole-file read for byte-identity assertions on a published database. */ +static unsigned char *read_file_bytes(const char *path, size_t *len_out) { + *len_out = 0; + FILE *f = fopen(path, "rb"); + if (!f) { + return NULL; + } + long size = fseek(f, 0, SEEK_END) == 0 ? ftell(f) : -1; + if (size < 0) { + fclose(f); + return NULL; + } + rewind(f); + unsigned char *buf = malloc((size_t)size + 1); + if (!buf) { + fclose(f); + return NULL; + } + size_t got = fread(buf, 1, (size_t)size, f); + fclose(f); + if (got != (size_t)size) { + free(buf); + return NULL; + } + *len_out = got; + return buf; +} + +TEST(pipeline_backpressure_futile_nap_disengages) { + /* 64 tiny files: > MIN_FILES_FOR_PARALLEL (50) so the parallel path (and its + * back-pressure gate) actually runs; old-code cycles (~64) >> the bound. */ + if (!write_backpressure_fixture()) { + FAIL("failed to create fixture"); + } /* 1 MB budget: over-budget on every pull, unreclaimable by napping. * Set via the test hook, NOT setenv + cbm_mem_init: the init-once guard @@ -12942,7 +13001,10 @@ TEST(pipeline_backpressure_futile_nap_disengages) { teardown_test_repo(); ASSERT_EQ(restore_workers_rc, 0); - ASSERT_EQ(rc, 0); + /* Decision A (#1997 #832): a budget that never drains is not a soft + * overshoot any more — after the latch and ONE confirmation cycle the + * attempt fails whole with the named code (rc==0 was the advisory era). */ + ASSERT_EQ(rc, CBM_PIPELINE_ABORT_OVER_BUDGET); /* Engagement guard (anti-vacuous): the gate must have actually run — the * parallel path taken and the 1 MB budget exceeded on every pull. cycles==0 * means the fixture routed sequential (or the gate was compiled out) and @@ -12951,8 +13013,9 @@ TEST(pipeline_backpressure_futile_nap_disengages) { FAIL("back-pressure gate never engaged (cycles==0) — fixture routed sequential?"); } /* Futile napping must disengage: at most one in-flight cycle per worker - * plus a small margin, never one per file (64). */ - long bound = TEST_WORKERS + 2; + * plus the single confirmation cycle and a small margin, never one per + * file (64). */ + long bound = TEST_WORKERS + 3; if (cycles > bound) { char msg[128]; snprintf(msg, sizeof(msg), "nap cycles %ld > bound %ld (gate re-paid per pull)", cycles, @@ -12962,6 +13025,102 @@ TEST(pipeline_backpressure_futile_nap_disengages) { PASS(); } +/* Decision A (#1997 #832): once back-pressure is futile AND one confirmation + * cycle still ends over budget, the attempt fails WHOLE with a named code — + * no partial graph is published, no stage residue remains, and the previously + * serving generation is byte-identical. */ +TEST(pipeline_over_budget_after_futility_fails_whole_and_preserves_db) { + if (!write_backpressure_fixture()) { + FAIL("failed to create fixture"); + } + enum { TEST_WORKERS = 4 }; + const char *old_workers = getenv("CBM_WORKERS"); + char *saved_workers = old_workers ? strdup(old_workers) : NULL; + if ((old_workers && !saved_workers) || cbm_setenv("CBM_WORKERS", "4", 1) != 0) { + free(saved_workers); + teardown_test_repo(); + FAIL("failed to pin CBM_WORKERS"); + } + + char db_path[512]; + snprintf(db_path, sizeof(db_path), "%s/budget.db", g_tmpdir); + + /* Generation 1 at the process budget: publishes normally. */ + cbm_pipeline_t *p = cbm_pipeline_new(g_tmpdir, db_path, CBM_MODE_FULL); + int rc_first = p ? cbm_pipeline_run(p) : -1; + char *project = p ? strdup(cbm_pipeline_project_name(p)) : NULL; + cbm_pipeline_free(p); + int nodes_before = -1; + bool valid_before = false; + cbm_store_t *live = rc_first == 0 && project ? cbm_store_open_path(db_path) : NULL; + if (live) { + valid_before = cbm_store_check_integrity(live); + nodes_before = cbm_store_count_nodes(live, project); + cbm_store_close(live); + } + size_t before_len = 0; + unsigned char *before = read_file_bytes(db_path, &before_len); + + /* Generation 2 at a 1 MiB budget over a CHANGED repo (every file gains a + * definition, so the run must re-extract all 64): over budget on every + * probe and unreclaimable by napping, so futility latches and the + * confirmation cycle still ends over budget. */ + bool mutated = mutate_backpressure_fixture(); + size_t saved_budget = cbm_mem_budget(); + cbm_mem_set_budget_for_tests((size_t)1024 * 1024); + bool over_at_start = cbm_mem_over_budget(); + cbm_pp_bp_nap_cycles_reset(); + p = cbm_pipeline_new(g_tmpdir, db_path, CBM_MODE_FULL); + int rc_second = p ? cbm_pipeline_run(p) : -1; + long cycles = cbm_pp_bp_nap_cycles(); + /* Restore the caller-visible budget BEFORE any assertion. */ + cbm_mem_set_budget_for_tests(saved_budget); + cbm_pipeline_free(p); + + size_t after_len = 0; + unsigned char *after = read_file_bytes(db_path, &after_len); + int stage_count = count_generation_stage_artifacts(g_tmpdir, "budget.db"); + int nodes_after = -1; + bool valid_after = false; + live = cbm_store_open_path(db_path); + if (live) { + valid_after = cbm_store_check_integrity(live); + nodes_after = project ? cbm_store_count_nodes(live, project) : -1; + cbm_store_close(live); + } + bool bytes_identical = + before && after && before_len == after_len && memcmp(before, after, before_len) == 0; + free(before); + free(after); + free(project); + int restore_workers_rc = + saved_workers ? cbm_setenv("CBM_WORKERS", saved_workers, 1) : cbm_unsetenv("CBM_WORKERS"); + free(saved_workers); + teardown_test_repo(); + + ASSERT_EQ(restore_workers_rc, 0); + ASSERT_EQ(rc_first, 0); + ASSERT_TRUE(valid_before); + ASSERT_GT(nodes_before, 0); + ASSERT_TRUE(before_len > 0); + ASSERT_TRUE(mutated); + ASSERT_TRUE(over_at_start); + /* The named failure — not the cancel sentinel, not success. */ + ASSERT_EQ(rc_second, CBM_PIPELINE_ABORT_OVER_BUDGET); + /* Engagement guard (anti-vacuous): fail-whole needs the latching cycle AND + * the confirmation cycle; fewer means the gate never decided anything. */ + if (cycles < 2) { + FAIL("back-pressure gate never confirmed futility (cycles<2) — fixture routed sequential?"); + } + /* The serving generation is untouched: same bytes, still valid, same + * graph; and the failed attempt left no stage residue behind. */ + ASSERT_TRUE(bytes_identical); + ASSERT_TRUE(valid_after); + ASSERT_EQ(nodes_after, nodes_before); + ASSERT_EQ(stage_count, 0); + PASS(); +} + /* TS cross-registry test hooks (ts_lsp.c) — extern to avoid pulling the * tree-sitter-typed ts_lsp.h into this store-level test. */ extern long cbm_ts_full_registry_builds(void); @@ -13774,6 +13933,7 @@ SUITE(pipeline) { RUN_TEST(pipeline_run_null); /* Extraction back-pressure */ RUN_TEST(pipeline_backpressure_futile_nap_disengages); + RUN_TEST(pipeline_over_budget_after_futility_fails_whole_and_preserves_db); /* Sequential cross-LSP shared registry (ms-typescript quadratic) */ RUN_TEST(pipeline_seq_ts_cross_uses_shared_registry); /* File persistence */