From 7acb5eac839f4745a73df79ce2d1e9374bb956c1 Mon Sep 17 00:00:00 2001 From: Martin Vogel Date: Sat, 26 Sep 2026 07:14:58 +0200 Subject: [PATCH] fix(index): name a worker that exits cleanly without a response (#1300) The supervisor creates the worker's response file before spawning it. A worker that exited 0 without writing that file left it empty, and slurp_worker_response read the empty file as "" with status OK. The run then counted as a SUCCESS with an empty response (the daemon marked the job successful). The clean-exit branch of worker_terminal_log also deleted the worker log, the only record of where the run stopped. Only an oversized response was treated as a failure. - index_supervisor: an empty response file is WORKER_RESPONSE_READ_EMPTY. A clean exit with no readable response sets response_missing, keeps the worker log, records its path, and records the last structured event the log holds (text or JSON format, last 64 KiB) as last_phase. It is logged as index.supervisor.no_response with the log path and the last phase. The outcome stays CLEAN, so the disposition is FALLBACK: an explicit error, with no in-process fallback and no recovery re-run of a worker that did not crash. - mcp: cbm_mcp_index_worker_no_response_failure builds the error (reason "no_response", last_phase, worker_log, hint). It is used for the first attempt and for a recovery run. - daemon application: the job failure message says the worker exited cleanly without writing a response, and gives the last phase and the log to inspect. - pipeline_incremental: the legacy per-file purge loop now logs incremental.purge.start (changed/deleted counts) and incremental.purge.progress. Progress lines carry counts only, use a ceiling stride (at most ~20 lines plus the final one) and never log one line per file. Before this, nothing was logged between incremental.edge_snapshot and incremental.purge. Tests (RED before, green after, RED on revert): - index_supervisor_clean_exit_without_response_is_named_failure_issue1300 (a real worker subprocess logs a phase and exits 0 without a response) - daemon_application_clean_exit_without_response_is_named_failure_issue1300 - incr_purge_reports_bounded_progress_issue1300 - index_supervisor_clean_exit_without_response_response_issue1300 (the contract of the response builder) Signed-off-by: Martin Vogel --- src/daemon/application.c | 9 ++++ src/mcp/index_supervisor.c | 84 +++++++++++++++++++++++++++++ src/mcp/index_supervisor.h | 8 +++ src/mcp/mcp.c | 52 ++++++++++++++++++ src/mcp/mcp.h | 6 +++ src/pipeline/pipeline_incremental.c | 24 +++++++++ tests/test_daemon_application.c | 84 ++++++++++++++++++++++++++++- tests/test_incremental.c | 69 ++++++++++++++++++++++++ tests/test_index_supervisor.c | 79 +++++++++++++++++++++++++++ tests/test_main.c | 8 +++ tests/test_mcp.c | 40 ++++++++++++++ 11 files changed, 462 insertions(+), 1 deletion(-) diff --git a/src/daemon/application.c b/src/daemon/application.c index 748b8bb75..d37b3e1c2 100644 --- a/src/daemon/application.c +++ b/src/daemon/application.c @@ -1192,6 +1192,15 @@ static char *application_job_failure_response(const cbm_index_worker_result_t *r (void)snprintf(message, sizeof(message), "index worker containment failed (%s); inspect log: %s", cbm_proc_outcome_str(result->outcome), log_path ? log_path : "unavailable"); + } else if (result && result->response_missing) { + /* #1300: name the silent clean exit, where it stopped, and the log. */ + const char *log = log_path ? log_path + : result->worker_log[0] ? result->worker_log + : "unavailable"; + (void)snprintf(message, sizeof(message), + "index worker exited cleanly (exit 0) without writing a response; the " + "index was not published; last phase reached: %s; inspect log: %s", + result->last_phase[0] ? result->last_phase : "unknown", log); } else if (result) { (void)snprintf(message, sizeof(message), "index worker ended with %s (exit=%d, signal=%d); inspect log: %s", diff --git a/src/mcp/index_supervisor.c b/src/mcp/index_supervisor.c index 5e15a7eed..223945d3c 100644 --- a/src/mcp/index_supervisor.c +++ b/src/mcp/index_supervisor.c @@ -359,6 +359,7 @@ typedef enum { WORKER_RESPONSE_READ_OK = 0, WORKER_RESPONSE_READ_ERROR, WORKER_RESPONSE_READ_TOO_LARGE, + WORKER_RESPONSE_READ_EMPTY, /* the pre-created file was never written (#1300) */ } worker_response_read_status_t; /* Read at most one daemon-application payload. The worker is already reaped, @@ -384,6 +385,14 @@ static char *slurp_worker_response(const char *path, worker_response_read_status (void)fclose(f); return NULL; } + if (n == 0) { + /* The supervisor creates the response file before spawning, so an + * empty file is a worker that never wrote its result. Reading it as + * "" made that a successful run with an empty response (#1300). */ + *status_out = WORKER_RESPONSE_READ_EMPTY; + (void)fclose(f); + return NULL; + } if (fseek(f, 0, SEEK_SET) != 0) { (void)fclose(f); return NULL; @@ -405,6 +414,67 @@ static char *slurp_worker_response(const char *path, worker_response_read_status return buf; } +enum { + WORKER_LOG_TAIL_BYTES = 64 * 1024, /* last-phase scan window: bounded, never the whole log */ + WORKER_LOG_LINE_CAP = 1024, +}; + +/* Copy the event name of one structured log line ("level=... msg= ..." + * or {"level":...,"event":"",...}) into out. False for any other line + * (plain stderr text, a truncated fragment). */ +static bool worker_log_line_event(const char *line, char *out, size_t out_size) { + static const char text_key[] = " msg="; + static const char json_key[] = "\"event\":\""; + const char *start = NULL; + char stop = ' '; + if (strncmp(line, "level=", strlen("level=")) == 0) { + const char *msg = strstr(line, text_key); + start = msg ? msg + strlen(text_key) : NULL; + } else if (strncmp(line, "{\"level\":", strlen("{\"level\":")) == 0) { + const char *event = strstr(line, json_key); + start = event ? event + strlen(json_key) : NULL; + stop = '"'; + } + if (!start) { + return false; + } + size_t length = 0; + while (start[length] && start[length] != stop && start[length] != '\n' && + start[length] != '\r') { + length++; + } + if (length == 0 || length >= out_size) { + return false; + } + memcpy(out, start, length); + out[length] = '\0'; + return true; +} + +/* #1300: the last structured event the worker logged before it exited — the + * phase it reached. Scans only the log's final WORKER_LOG_TAIL_BYTES. */ +static void worker_log_last_phase(const char *log_path, char *out, size_t out_size) { + (void)snprintf(out, out_size, "%s", "unknown"); + FILE *log = cbm_fopen(log_path, "rb"); + if (!log) { + return; + } + if (fseek(log, 0, SEEK_END) == 0) { + long size = ftell(log); + long from = size > WORKER_LOG_TAIL_BYTES ? size - WORKER_LOG_TAIL_BYTES : 0; + if (size >= 0 && fseek(log, from, SEEK_SET) == 0) { + char line[WORKER_LOG_LINE_CAP]; + char event[CBM_SZ_128]; + while (fgets(line, sizeof(line), log)) { + if (worker_log_line_event(line, event, sizeof(event))) { + (void)snprintf(out, out_size, "%s", event); + } + } + } + } + (void)fclose(log); +} + enum { INDEX_WORKER_PATH_CAP = CBM_SZ_4K, INDEX_WORKER_ARGV_CAP = 17, @@ -634,6 +704,10 @@ static void worker_terminal_log(cbm_index_worker_handle_t *handle) { } else if (handle->result.cancellation_requested) { cbm_log_warn("index.supervisor.worker_cancelled", "outcome", cbm_proc_outcome_str(handle->result.outcome), "log", handle->log_path); + } else if (handle->result.response_missing) { + /* #1300: keep and name the log — it is the only record of the run. */ + cbm_log_error("index.supervisor.no_response", "exit_code", exit_text, "last_phase", + handle->result.last_phase, "log", handle->log_path); } else if (handle->result.outcome == CBM_PROC_CLEAN && !cbm_profile_active) { (void)cbm_unlink(handle->log_path); } else if (handle->result.outcome == CBM_PROC_CLEAN) { @@ -828,6 +902,16 @@ cbm_index_worker_poll_t cbm_index_worker_poll(cbm_index_worker_handle_t *handle, handle->result.response_rejected = true; handle->result.outcome = CBM_PROC_EXIT_NONZERO; handle->result.exit_code = -1; + } else if (!handle->result.response) { + /* #1300: a clean exit that wrote no response. The outcome stays + * CLEAN (the disposition is FALLBACK: an explicit error, no + * in-process fallback and no recovery re-run of a worker that did + * not crash); the log and the last phase it reached are kept. */ + handle->result.response_missing = true; + worker_log_last_phase(handle->log_path, handle->result.last_phase, + sizeof(handle->result.last_phase)); + (void)snprintf(handle->result.worker_log, sizeof(handle->result.worker_log), "%s", + handle->log_path); } } (void)cbm_unlink(handle->response_path); diff --git a/src/mcp/index_supervisor.h b/src/mcp/index_supervisor.h index 42b6bd74a..c231a563e 100644 --- a/src/mcp/index_supervisor.h +++ b/src/mcp/index_supervisor.h @@ -147,6 +147,14 @@ typedef struct { bool forced; bool tree_quiesced; bool supervision_failed; + /* #1300: the worker exited CLEAN but wrote no response (an empty or + * unreadable response file). A contained failure, never a success: the + * response stays NULL, the worker log is retained and named, and + * last_phase is the last structured event that log recorded ("unknown" + * when it holds none), so the report says where the run stopped. */ + bool response_missing; + char last_phase[128]; + char worker_log[4096]; bool response_rejected; /* clean worker exceeded the bounded response protocol */ char *response; /* worker result only after a contained, uncancelled CLEAN exit; * borrowed for async polls, caller-owned from the sync wrapper */ diff --git a/src/mcp/mcp.c b/src/mcp/mcp.c index d6c7bd468..c627999aa 100644 --- a/src/mcp/mcp.c +++ b/src/mcp/mcp.c @@ -10728,6 +10728,44 @@ int cbm_index_restart_cap_for_testing(void) { return index_restart_cap(); } +char *cbm_mcp_index_worker_no_response_failure(const char *args, + const cbm_index_worker_result_t *worker_result) { + const char *phase = + worker_result && worker_result->last_phase[0] ? worker_result->last_phase : "unknown"; + const char *log = + worker_result && worker_result->worker_log[0] ? worker_result->worker_log : "unavailable"; + /* Borrow repo_path from a parsed view and write into a stack buffer: no + * raw heap strings to hand back (memory-core ratchet). */ + yyjson_doc *args_doc = args ? yyjson_read(args, strlen(args), 0) : NULL; + const char *repo_path = + yyjson_get_str(yyjson_obj_get(yyjson_doc_get_root(args_doc), "repo_path")); + yyjson_mut_doc *doc = yyjson_mut_doc_new(NULL); + yyjson_mut_val *root = yyjson_mut_obj(doc); + yyjson_mut_doc_set_root(doc, root); + yyjson_mut_obj_add_str(doc, root, "status", "error"); + yyjson_mut_obj_add_str(doc, root, "outcome", cbm_proc_outcome_str(CBM_PROC_CLEAN)); + yyjson_mut_obj_add_str(doc, root, "reason", "no_response"); + yyjson_mut_obj_add_strcpy(doc, root, "last_phase", phase); + yyjson_mut_obj_add_strcpy(doc, root, "worker_log", log); + char hint[CBM_SZ_1K]; + (void)snprintf(hint, sizeof(hint), + "Indexing worker exited cleanly (exit 0) without writing a response; the " + "index was not published. Last phase reached: %s. Inspect the worker log: %s", + phase, log); + yyjson_mut_obj_add_strcpy(doc, root, "hint", hint); + if (repo_path) { + yyjson_mut_obj_add_strcpy(doc, root, "repo_path", repo_path); + } + yyjson_doc_free(args_doc); + char json[CBM_SZ_16K]; + bool written = yy_mut_normalize_output_text(doc, root, true) && + yyjson_mut_write_buf(json, sizeof(json), doc, 0, NULL) > 0; + yyjson_mut_doc_free(doc); + /* The fields are bounded (hint 1 KiB, log 4 KiB, phase 128 B), so the + * plain hint is only the fallback for a pathological repo_path. */ + return cbm_mcp_text_result(written ? json : hint, true); +} + static char *index_run_supervised(cbm_mcp_server_t *srv, const char *args) { invalidate_cached_store(srv); @@ -10739,6 +10777,13 @@ static char *index_run_supervised(cbm_mcp_server_t *srv, const char *args) { cbm_mcp_supervised_result_disposition_t disposition = cbm_mcp_supervised_result_disposition(rc, &wr); + if (wr.response_missing) { + /* #1300: a clean exit without a response is a named failure. */ + char *failure = cbm_mcp_index_worker_no_response_failure(args, &wr); + cbm_index_worker_result_free(&wr); + invalidate_cached_store(srv); + return failure; + } if (disposition == CBM_MCP_SUPERVISED_RESULT_FALLBACK) { cbm_proc_outcome_t outcome = wr.outcome; cbm_index_worker_result_free(&wr); @@ -10807,6 +10852,13 @@ static char *index_run_supervised(cbm_mcp_server_t *srv, const char *args) { srv ? &srv->pipeline_cancel_requested : NULL, &wr2); cbm_mcp_supervised_result_disposition_t recovery_disposition = cbm_mcp_supervised_result_disposition(rc2, &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. */ + resp = cbm_mcp_index_worker_no_response_failure(args, &wr2); + cbm_index_worker_result_free(&wr2); + break; + } if (recovery_disposition == CBM_MCP_SUPERVISED_RESULT_FALLBACK) { /* Keep the original worker failure if recovery setup fails. */ cbm_log_error("index.supervisor.recovery_start_failed", "original_outcome", diff --git a/src/mcp/mcp.h b/src/mcp/mcp.h index 09befb780..eaad2c74d 100644 --- a/src/mcp/mcp.h +++ b/src/mcp/mcp.h @@ -217,6 +217,12 @@ char *cbm_mcp_handle_tool(cbm_mcp_server_t *srv, const char *tool_name, const ch /* ── Supervised background index (RSS isolation, #832) ────────── */ +/* #1300: the explicit error response for a worker that exited cleanly without + * writing a response (worker_result->response_missing). Names the retained + * worker log and the last phase it reached; caller frees. */ +char *cbm_mcp_index_worker_no_response_failure(const char *args, + const cbm_index_worker_result_t *worker_result); + /* One shared policy gate for supervised-index callers. Only an explicitly safe * terminal can yield a result or enter crash/hang recovery; cancellation, * failed tree containment, and unavailable worker startup are terminal. The diff --git a/src/pipeline/pipeline_incremental.c b/src/pipeline/pipeline_incremental.c index 154518658..99195cc2c 100644 --- a/src/pipeline/pipeline_incremental.c +++ b/src/pipeline/pipeline_incremental.c @@ -102,6 +102,25 @@ static const char *itoa_buf(int v) { return buf[idx]; } +/* #1300: the purge of stale per-file nodes can run for a long time on a large + * persisted graph (reported: hundreds of files against ~268k stored hashes), + * and it logged nothing between incremental.edge_snapshot and + * incremental.purge — a stall there and a clean exit left no trace of where + * the run was. Progress is counts only: a ceiling stride keeps it to at most + * INCR_PURGE_PROGRESS_STEPS lines plus the final one, never one per file. */ +enum { INCR_PURGE_PROGRESS_STEPS = 20 }; + +static void incr_purge_progress(int done, int total, struct timespec start) { + int stride = (total + INCR_PURGE_PROGRESS_STEPS - SKIP_ONE) / INCR_PURGE_PROGRESS_STEPS; + if (stride < SKIP_ONE) { + stride = SKIP_ONE; + } + if (done % stride == 0 || done == total) { + cbm_log_info("incremental.purge.progress", "files_done", itoa_buf(done), "files_total", + itoa_buf(total), "elapsed_ms", itoa_buf((int)elapsed_ms(start))); + } +} + static const char *incr_mode_name(int mode) { switch (mode) { case CBM_MODE_FULL: @@ -2688,12 +2707,17 @@ int cbm_pipeline_run_incremental(cbm_pipeline_t *p, const char *db_path, cbm_fil /* Step 2: Purge stale nodes */ cbm_clock_gettime(CLOCK_MONOTONIC, &t); + int purge_total = ci + deleted_count; + cbm_log_info("incremental.purge.start", "changed", itoa_buf(ci), "deleted", + itoa_buf(deleted_count)); for (int i = 0; i < ci; i++) { cbm_gbuf_delete_by_file(existing, changed_files[i].rel_path); + incr_purge_progress(i + SKIP_ONE, purge_total, t); } for (int i = 0; i < deleted_count; i++) { cbm_gbuf_delete_by_file(existing, deleted[i]); free(deleted[i]); + incr_purge_progress(ci + i + SKIP_ONE, purge_total, t); } free(deleted); cbm_log_info("incremental.purge", "elapsed_ms", itoa_buf((int)elapsed_ms(t))); diff --git a/tests/test_daemon_application.c b/tests/test_daemon_application.c index 0108e8ec5..7647d3351 100644 --- a/tests/test_daemon_application.c +++ b/tests/test_daemon_application.c @@ -1349,6 +1349,8 @@ TEST(daemon_application_prune_clears_logical_watch_for_reregistration) { enum { APP_FAKE_MAX_ATTEMPTS = 16 }; enum { APP_FAKE_ARGS_CAP = 2048 }; +/* Scripted response slot: a CLEAN exit that wrote no response (#1300). */ +static const char APP_FAKE_NO_RESPONSE[] = ""; typedef struct { atomic_int starts; @@ -1534,7 +1536,16 @@ static cbm_index_worker_poll_t app_fake_worker_poll(void *opaque, ? worker->context->responses[worker->attempt] : "{\"content\":[{\"type\":\"text\",\"text\":\"{" "\\\"status\\\":\\\"indexed\\\"}\"}]}"; - worker->result.response = outcome == CBM_PROC_CLEAN ? cbm_strdup(response) : NULL; + bool silent = response == APP_FAKE_NO_RESPONSE; + worker->result.response = + outcome == CBM_PROC_CLEAN && !silent ? cbm_strdup(response) : NULL; + if (silent) { + /* What the supervisor reports for a clean exit without a + * response (#1300). */ + worker->result.response_missing = true; + (void)snprintf(worker->result.last_phase, sizeof(worker->result.last_phase), "%s", + "incremental.purge.progress"); + } *result_out = &worker->result; return CBM_INDEX_WORKER_POLL_TERMINAL; } @@ -3516,6 +3527,76 @@ TEST(daemon_application_disconnect_before_request_callback_is_sticky) { PASS(); } +/* #1300: a worker that exits cleanly without a response must reach the + * client as a named failure: said to be a clean exit with no response, with + * the last phase it reached and the worker log to inspect — not the generic + * "ended with clean" line. No recovery re-run for a worker that did not + * crash. */ +TEST(daemon_application_clean_exit_without_response_is_named_failure_issue1300) { + 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] = APP_FAKE_NO_RESPONSE; + 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 = app_test_open(&callbacks, 1300); + char root[APP_TEST_PATH_CAP]; + snprintf(root, sizeof(root), "%s/cbm-app-no-response-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; + 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; + if (session) { + callbacks.session_close(callbacks.context, session); + } + bool stopped = application && cbm_daemon_application_shutdown(application, APP_TEST_TIMEOUT_MS); + cbm_daemon_application_free(application); + + const char *text = response ? (const char *)response : ""; + ASSERT_TRUE(setup); + ASSERT_EQ(status, CBM_DAEMON_RUNTIME_APPLICATION_OK); + ASSERT_TRUE(strstr(text, "without writing a response") != NULL); + ASSERT_TRUE(strstr(text, "incremental.purge.progress") != NULL); + ASSERT_TRUE(strstr(text, "/tmp/cbm-fake-worker.log") != NULL); + ASSERT_TRUE(strstr(text, "indexed") == NULL); + ASSERT_EQ(atomic_load(&fake.starts), 1); + ASSERT_TRUE(stopped); + + free(response); + free(context); + free(tool); + (void)cbm_rmdir(root); + PASS(); +} + TEST(daemon_application_request_cancel_preserves_persistent_watch_and_session) { app_watch_race_fixture_t fixture; bool fixture_ready = app_watch_race_fixture_init(&fixture, 3021); @@ -5691,4 +5772,5 @@ SUITE(daemon_application) { 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); + RUN_TEST(daemon_application_clean_exit_without_response_is_named_failure_issue1300); } diff --git a/tests/test_incremental.c b/tests/test_incremental.c index 4374e97f3..b3ac8b473 100644 --- a/tests/test_incremental.c +++ b/tests/test_incremental.c @@ -16,6 +16,7 @@ #include #include #include +#include #include #include #include @@ -546,6 +547,71 @@ TEST(incr_formatter_run) { PASS(); } +static char g_purge_log[8192]; +static int g_purge_progress_lines; + +static void purge_log_sink(const char *line) { + if (!line || !strstr(line, "msg=incremental.purge")) { + return; + } + if (strstr(line, "msg=incremental.purge.progress")) { + g_purge_progress_lines++; + } + size_t used = strlen(g_purge_log); + (void)snprintf(g_purge_log + used, sizeof(g_purge_log) - used, "%s\n", line); +} + +/* #1300: the per-file purge after incremental.edge_snapshot logged nothing + * until it finished, so a stall there left no trace of where the run was. It + * now announces its counts and reports bounded progress — counts only, at + * most 20 progress lines however many files, ending at files_done == + * files_total. */ +TEST(incr_purge_reports_bounded_progress_issue1300) { + char *resp = index_repo(); + ASSERT(resp != NULL); + free(resp); + reformat_files("fastapi", 30); + /* The loop instrumented for #1300 sits on the legacy partial route (the + * production closure route purges in SQL and logs delta.snapshot_purge). */ + cbm_pipeline_incremental_test_force_legacy_partial_once(); + + CBMLogLevel saved_level = cbm_log_get_level(); + CBMLogFormat saved_format = cbm_log_get_format(); + cbm_log_set_level(CBM_LOG_INFO); + cbm_log_set_format(CBM_LOG_FORMAT_TEXT); + g_purge_log[0] = '\0'; + g_purge_progress_lines = 0; + cbm_log_set_sink_ex(purge_log_sink, CBM_LOG_SINK_TEE); + resp = index_repo(); + cbm_log_set_sink(NULL); + cbm_log_set_level(saved_level); + cbm_log_set_format(saved_format); + ASSERT(resp != NULL); + ASSERT(strstr(resp, "indexed") != NULL); + free(resp); + ASSERT_EQ(cbm_pipeline_incremental_test_last_route(), CBM_INCREMENTAL_ROUTE_LEGACY_PARTIAL); + + const char *start = strstr(g_purge_log, "msg=incremental.purge.start"); + ASSERT(start != NULL); + int changed = -1; + int deleted = -1; + const char *changed_at = strstr(start, "changed="); + const char *deleted_at = strstr(start, "deleted="); + ASSERT(changed_at != NULL && deleted_at != NULL); + changed = atoi(changed_at + strlen("changed=")); + deleted = atoi(deleted_at + strlen("deleted=")); + int total = changed + deleted; + ASSERT_GT(total, 20); /* enough files that one line per file would exceed the bound */ + ASSERT_GT(g_purge_progress_lines, 0); + ASSERT(g_purge_progress_lines <= 21); + ASSERT(g_purge_progress_lines < total); /* never one line per file */ + char last[64]; + (void)snprintf(last, sizeof(last), "files_done=%d files_total=%d", total, total); + ASSERT(strstr(g_purge_log, last) != NULL); + ASSERT(strstr(g_purge_log, "msg=incremental.purge elapsed_ms=") != NULL); + PASS(); +} + TEST(incr_add_file) { int nodes_before = get_node_count(); @@ -3349,6 +3415,9 @@ SUITE(incremental) { RUN_TEST(tool_err_ingest_bad_project); RUN_TEST(tool_err_ingest_no_traces); + /* #1300: legacy-partial purge progress (the full rebuild below resets the graph) */ + RUN_TEST(incr_purge_reports_bounded_progress_issue1300); + /* Phase 40: index_repository mode=full + arch empty aspects */ RUN_TEST(tool_index_mode_full); RUN_TEST(tool_arch_empty_aspects); diff --git a/tests/test_index_supervisor.c b/tests/test_index_supervisor.c index d69354d76..1861c8076 100644 --- a/tests/test_index_supervisor.c +++ b/tests/test_index_supervisor.c @@ -9,7 +9,9 @@ #include "foundation/log.h" #include "foundation/platform.h" #include "foundation/profile.h" +#include "foundation/constants.h" #include "mcp/index_supervisor.h" +#include "mcp/mcp.h" #include #include @@ -838,6 +840,82 @@ TEST(index_supervisor_oversized_response_is_contained_and_log_is_retained) { PASS(); } +static char g_index_supervisor_no_response_log[CBM_SZ_2K]; + +static void index_supervisor_test_no_response_sink(const char *line) { + if (strstr(line, "index.supervisor.no_response")) { + (void)snprintf(g_index_supervisor_no_response_log, + sizeof(g_index_supervisor_no_response_log), "%s", line); + } +} + +/* #1300: a worker that exits 0 without writing its response. The supervisor + * pre-creates the response file, so the old read returned "" and the run was + * a SUCCESS with an empty response; the clean-exit branch then deleted the + * worker log, the only record of where the run stopped. It must instead be a + * named failure: no response, the log retained, and the last phase the log + * recorded (incremental.edge_snapshot, not the plain stderr line after it). */ +TEST(index_supervisor_clean_exit_without_response_is_named_failure_issue1300) { + char cache[INDEX_SUPERVISOR_TEST_PATH_CAP]; + (void)snprintf(cache, sizeof(cache), "%s/cbm-index-silent-XXXXXX", cbm_tmpdir()); + ASSERT_NOT_NULL(cbm_mkdtemp(cache)); + const char *old_cache = getenv("CBM_CACHE_DIR"); + char *saved_cache = old_cache ? cbm_strdup(old_cache) : NULL; + (void)cbm_setenv("CBM_CACHE_DIR", cache, 1); + CBMLogLevel saved_level = cbm_log_get_level(); + CBMLogFormat saved_format = cbm_log_get_format(); + cbm_log_set_level(CBM_LOG_INFO); + cbm_log_set_format(CBM_LOG_FORMAT_TEXT); + g_index_supervisor_no_response_log[0] = '\0'; + cbm_log_set_sink_ex(index_supervisor_test_no_response_sink, CBM_LOG_SINK_TEE); + + cbm_index_worker_handle_t *handle = NULL; + int start_rc = cbm_index_worker_start("{\"__cbm_test_worker\":\"silent-exit\"}", 0, false, NULL, + NULL, &handle); + char log_path[INDEX_SUPERVISOR_TEST_PATH_CAP] = {0}; + if (handle) { + (void)snprintf(log_path, sizeof(log_path), "%s", cbm_index_worker_log_path(handle)); + } + const cbm_index_worker_result_t *result = NULL; + bool terminal = handle && index_supervisor_test_poll_terminal( + handle, INDEX_SUPERVISOR_TEST_TERMINAL_MS, &result); + bool clean_exit = terminal && result && result->outcome == CBM_PROC_CLEAN && + result->exit_code == 0 && result->tree_quiesced; + bool no_response = terminal && result && result->response == NULL; + bool named = terminal && result && result->response_missing && + strcmp(result->last_phase, "incremental.edge_snapshot") == 0 && + strcmp(result->worker_log, log_path) == 0; + bool not_success = + terminal && result && + cbm_mcp_supervised_result_disposition(0, result) != CBM_MCP_SUPERVISED_RESULT_SUCCESS; + bool log_retained = log_path[0] && cbm_file_size(log_path) > 0; + if (terminal) { + cbm_index_worker_destroy(handle); + } else { + index_supervisor_test_dump("silent-exit worker log", log_path); + index_supervisor_test_cleanup_handle(handle); + } + cbm_log_set_sink(NULL); + cbm_log_set_level(saved_level); + cbm_log_set_format(saved_format); + bool event_named = strstr(g_index_supervisor_no_response_log, + "last_phase=incremental.edge_snapshot") != NULL && + log_path[0] && strstr(g_index_supervisor_no_response_log, log_path) != NULL; + (void)cbm_unlink(log_path); + index_supervisor_test_restore_env("CBM_CACHE_DIR", saved_cache); + (void)th_rmtree(cache); + + ASSERT_EQ(start_rc, 0); + ASSERT_TRUE(terminal); + ASSERT_TRUE(clean_exit); + ASSERT_TRUE(no_response); + ASSERT_TRUE(not_success); + ASSERT_TRUE(named); + ASSERT_TRUE(log_retained); + ASSERT_TRUE(event_named); + PASS(); +} + /* #1070, #1130, #1132, #1133, #1145, #1450: six reports of an indexing worker * that died leaving "the worker log file is completely empty (0 KB)". Nothing * was ever flushed, so not one of them is reproducible or attributable — the @@ -950,5 +1028,6 @@ SUITE(index_supervisor) { RUN_TEST(index_supervisor_worker_keeps_default_info_liveness_heartbeat); RUN_TEST(index_supervisor_drains_terminal_backlog_into_request_progress_callback); RUN_TEST(index_supervisor_oversized_response_is_contained_and_log_is_retained); + RUN_TEST(index_supervisor_clean_exit_without_response_is_named_failure_issue1300); RUN_TEST(index_supervisor_killed_worker_log_is_never_empty_and_names_the_run); } diff --git a/tests/test_main.c b/tests/test_main.c index db3748288..f7d2cfc08 100644 --- a/tests/test_main.c +++ b/tests/test_main.c @@ -270,6 +270,14 @@ static void tf_index_worker_probe(const char *args_json, const char *response_ou fflush(NULL); _Exit(response ? 0 : 1); } + if (strstr(args_json, "\"silent-exit\"")) { + /* #1300: reach a phase, then exit 0 without writing the response. The + * trailing plain-text line must not be mistaken for a phase. */ + cbm_log_info("incremental.edge_snapshot", "captured", "3"); + (void)fprintf(stderr, "async worker silent-exit probe\n"); + fflush(NULL); + _Exit(0); + } if (strstr(args_json, "\"crash\"")) { (void)fprintf(stderr, "async worker crash probe\n"); fflush(NULL); diff --git a/tests/test_mcp.c b/tests/test_mcp.c index ffa9ea3a3..9dadff44a 100644 --- a/tests/test_mcp.c +++ b/tests/test_mcp.c @@ -18109,6 +18109,45 @@ TEST(index_supervisor_unsafe_clean_is_never_fallback_or_recovery) { PASS(); } +/* #1300: a clean worker exit without a response is never a success, and its + * error response names the reason, the last phase and the worker log. */ +TEST(index_supervisor_clean_exit_without_response_response_issue1300) { + cbm_index_worker_result_t result = { + .outcome = CBM_PROC_CLEAN, + .exit_code = 0, + .tree_quiesced = true, + .response_missing = true, + }; + (void)snprintf(result.last_phase, sizeof(result.last_phase), "%s", "incremental.edge_snapshot"); + (void)snprintf(result.worker_log, sizeof(result.worker_log), "%s", + "/cache/logs/.worker-log-abc123"); + ASSERT_EQ(cbm_mcp_supervised_result_disposition(0, &result), + CBM_MCP_SUPERVISED_RESULT_FALLBACK); + + char *response = cbm_mcp_index_worker_no_response_failure( + "{\"repo_path\":\"/work/repo\",\"mode\":\"full\"}", &result); + ASSERT_NOT_NULL(response); + ASSERT_TRUE(response_contains_json_fragment(response, "\"isError\":true")); + ASSERT_TRUE(response_contains_json_fragment(response, "\"status\":\"error\"")); + ASSERT_TRUE(response_contains_json_fragment(response, "\"reason\":\"no_response\"")); + ASSERT_TRUE( + response_contains_json_fragment(response, "\"last_phase\":\"incremental.edge_snapshot\"")); + ASSERT_TRUE(response_contains_json_fragment( + response, "\"worker_log\":\"/cache/logs/.worker-log-abc123\"")); + ASSERT_TRUE(response_contains_json_fragment(response, "\"repo_path\":\"/work/repo\"")); + free(response); + + /* No phase recorded, no log known: still named, never blank. */ + result.last_phase[0] = '\0'; + result.worker_log[0] = '\0'; + response = cbm_mcp_index_worker_no_response_failure(NULL, &result); + ASSERT_NOT_NULL(response); + ASSERT_TRUE(response_contains_json_fragment(response, "\"last_phase\":\"unknown\"")); + ASSERT_TRUE(response_contains_json_fragment(response, "\"worker_log\":\"unavailable\"")); + free(response); + 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. */ enum { @@ -20684,6 +20723,7 @@ SUITE(mcp) { 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); + RUN_TEST(index_supervisor_clean_exit_without_response_response_issue1300); 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);