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