mirror of
https://github.com/DeusData/codebase-memory-mcp.git
synced 2026-11-04 10:48:20 +00:00
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 <martin.vogel.tech@gmail.com>
This commit is contained in:
1 parent
64c23fab6c
commit
7acb5eac83
11 files changed
+462
-1
No files matched your search
@@ -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",
|
||||
|
||||
@@ -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=<event> ..."
|
||||
* or {"level":...,"event":"<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);
|
||||
|
||||
@@ -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 */
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)));
|
||||
|
||||
@@ -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[] = "<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);
|
||||
}
|
||||
@@ -16,6 +16,7 @@
|
||||
#include <mcp/mcp.h>
|
||||
#include <store/store.h>
|
||||
#include <pipeline/pipeline.h>
|
||||
#include <pipeline/pipeline_internal.h>
|
||||
#include <foundation/log.h>
|
||||
#include <foundation/mem.h>
|
||||
#include <foundation/platform.h>
|
||||
@@ -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);
|
||||
|
||||
@@ -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 <stdbool.h>
|
||||
#include <stdint.h>
|
||||
@@ -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);
|
||||
}
|
||||
@@ -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);
|
||||
|
||||
@@ -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);
|
||||
|
||||
Reference in new issue
Block a user