diff --git a/src/daemon/application.c b/src/daemon/application.c index 0026e88392..daca0e5a29 100644 --- a/src/daemon/application.c +++ b/src/daemon/application.c @@ -2126,6 +2126,9 @@ static bool application_index_args_normalize_defaults(yyjson_mut_val *root) { if (name && (!yyjson_mut_is_str(name) || yyjson_mut_get_len(name) == 0)) { (void)yyjson_mut_obj_remove_key(root, "name"); } + /* Automatic jobs carry _cbm_background, which picks the worker count and + * never what is indexed, so an explicit request joins them. */ + (void)yyjson_mut_obj_remove_key(root, "_cbm_background"); return true; } @@ -2539,10 +2542,10 @@ static char *application_auto_index_args(cbm_daemon_application_t *application, return NULL; } yyjson_mut_doc_set_root(document, root); - char *args = yyjson_mut_obj_add_strcpy(document, root, "repo_path", root_path) && - application_index_args_add_policy(application, document, root) - ? yyjson_mut_write(document, 0, NULL) - : NULL; + bool encoded = yyjson_mut_obj_add_strcpy(document, root, "repo_path", root_path) && + application_index_args_add_policy(application, document, root) && + yyjson_mut_obj_add_bool(document, root, "_cbm_background", true); + char *args = encoded ? yyjson_mut_write(document, 0, NULL) : NULL; yyjson_mut_doc_free(document); return args; } @@ -4366,7 +4369,8 @@ static int application_background_index(cbm_daemon_application_t *application, } yyjson_mut_doc_set_root(document, root); bool encoded = yyjson_mut_obj_add_strcpy(document, root, "repo_path", canonical_root) && - application_index_args_add_policy(application, document, root); + application_index_args_add_policy(application, document, root) && + yyjson_mut_obj_add_bool(document, root, "_cbm_background", true); char *default_project = cbm_project_name_from_path(canonical_root); bool custom_project = project_name[0] && (!default_project || strcmp(default_project, project_name) != 0); diff --git a/src/mcp/mcp.c b/src/mcp/mcp.c index 1fc5f45888..58e0302759 100644 --- a/src/mcp/mcp.c +++ b/src/mcp/mcp.c @@ -12435,7 +12435,8 @@ static char *index_run_supervised_path(cbm_mcp_server_t *srv, const char *root_p * worker reads no config); unset, the worker's index mode decides. */ if (!yyjson_mut_obj_add_strcpy(doc, root, "repo_path", root_path) || !cbm_mcp_index_policy_add_to_args(doc, root, &policy) || - !cbm_mcp_index_doc_candidates_add_to_args(doc, root, srv ? srv->config : NULL)) { + !cbm_mcp_index_doc_candidates_add_to_args(doc, root, srv ? srv->config : NULL) || + !yyjson_mut_obj_add_bool(doc, root, "_cbm_background", true)) { yyjson_mut_doc_free(doc); return NULL; } @@ -12738,6 +12739,22 @@ static bool index_root_owner_resolve(const char *repo_path, char **owner_out, ch return false; } +/* Background CPU headroom is for a refresh of a graph that already serves. + * A project's first index keeps every core, even when it was started + * automatically: the caller is waiting on a graph that does not exist yet. + * An empty or unreadable file is not a committed index. Both sites that + * honour automatic indexing — handle_index_repository's _cbm_background + * flag and the in-process autoindex_thread — go through this so they + * cannot drift. */ +static bool project_has_committed_index(const char *project) { + if (!project || !project[0]) { + return false; + } + char db_path[CBM_SZ_1K]; + project_db_path(project, db_path, sizeof(db_path)); + return db_path[0] && project_db_is_servable(project, db_path); +} + /* The three heap strings handle_index_repository owns from * cbm_mcp_get_string_arg / resolved_repo_path_from_project_arg. One release * point keeps the dozen early-return paths in step; free(NULL) is a no-op, so @@ -12867,6 +12884,7 @@ static char *handle_index_repository(cbm_mcp_server_t *srv, const char *args) { char *repo_path = cbm_mcp_get_string_arg(args, "repo_path"); char *mode_str = cbm_mcp_get_string_arg(args, "mode"); char *name_override = cbm_mcp_get_string_arg(args, "name"); + bool background = cbm_mcp_get_bool_arg(args, "_cbm_background"); cbm_normalize_path_sep(repo_path); if (!repo_path) { @@ -13039,6 +13057,10 @@ static char *handle_index_repository(cbm_mcp_server_t *srv, const char *args) { return cbm_mcp_text_result("invalid project name", true); } free(name_override); + /* Headroom follows the final project name, decided before artifact + * bootstrap can create the first local copy of this index. */ + cbm_pipeline_set_background(p, background && + project_has_committed_index(cbm_pipeline_project_name(p))); cbm_pipeline_set_persistence(p, persistence); cbm_pipeline_set_resource_policy(p, &resource_policy); index_doc_candidates_apply(p, doc_candidates); @@ -21388,6 +21410,7 @@ static void *autoindex_thread(void *arg) { return NULL; } index_doc_candidates_apply(p, index_doc_candidates_choice(srv, NULL)); + cbm_pipeline_set_background(p, project_has_committed_index(cbm_pipeline_project_name(p))); /* Block until any concurrent pipeline finishes */ cbm_pipeline_lock(); diff --git a/src/pipeline/lsp_surface.c b/src/pipeline/lsp_surface.c index f8d43ff695..62c35f3249 100644 --- a/src/pipeline/lsp_surface.c +++ b/src/pipeline/lsp_surface.c @@ -486,10 +486,12 @@ typedef struct { const CBMLSPDef *all_defs; const int *def_starts; cbm_lsp_surface_row_t *rows; + int file_count; + _Atomic int next_file; _Atomic bool failed; } surface_job_t; -static void surface_row_one(int i, void *arg) { +static void surface_row_build(int i, void *arg) { surface_job_t *job = (surface_job_t *)arg; if (atomic_load_explicit(&job->failed, memory_order_relaxed)) { return; @@ -529,6 +531,45 @@ static void surface_row_one(int i, void *arg) { } } +#if defined(CBM_ENABLE_TEST_SEAMS) && CBM_ENABLE_TEST_SEAMS +static cbm_lsp_surface_row_observer_fn g_row_observer; +static void *g_row_observer_ctx; + +void cbm_lsp_surface_test_set_row_observer(cbm_lsp_surface_row_observer_fn observer, void *ctx) { + g_row_observer = observer; + g_row_observer_ctx = ctx; +} +#endif + +static void surface_row_one(int i, void *arg) { +#if defined(CBM_ENABLE_TEST_SEAMS) && CBM_ENABLE_TEST_SEAMS + if (g_row_observer) { + g_row_observer(i, true, g_row_observer_ctx); + } +#endif + surface_row_build(i, arg); +#if defined(CBM_ENABLE_TEST_SEAMS) && CBM_ENABLE_TEST_SEAMS + if (g_row_observer) { + g_row_observer(i, false, g_row_observer_ctx); + } +#endif +} + +/* The pool also runs work on its caller. Dispatch at most one coarse job per + * allowed participant, each stealing files, so the total active row builders + * cannot exceed the pipeline budget even when the caller takes a job. */ +static void surface_rows_worker(int worker, void *arg) { + (void)worker; + surface_job_t *job = arg; + for (;;) { + int i = atomic_fetch_add_explicit(&job->next_file, 1, memory_order_relaxed); + if (i >= job->file_count) { + break; + } + surface_row_one(i, job); + } +} + int cbm_lsp_surface_build_rows(const cbm_pipeline_ctx_t *ctx, const char *project, CBMFileResult **cache, const cbm_file_info_t *files, int file_count, const CBMLSPDef *all_defs, const int *def_starts, @@ -550,10 +591,16 @@ int cbm_lsp_surface_build_rows(const cbm_pipeline_ctx_t *ctx, const char *projec .all_defs = all_defs, .def_starts = def_starts, .rows = rows, + .file_count = file_count, }; + atomic_init(&job.next_file, 0); atomic_init(&job.failed, false); - cbm_parallel_for(file_count, surface_row_one, &job, - (cbm_parallel_for_opts_t){.max_workers = 0, .force_pthreads = false}); + int workers = cbm_pipeline_worker_count(ctx ? ctx->pipeline : NULL); + if (workers > file_count) { + workers = file_count; + } + cbm_parallel_for(workers, surface_rows_worker, &job, + (cbm_parallel_for_opts_t){.max_workers = workers, .force_pthreads = false}); if (atomic_load_explicit(&job.failed, memory_order_relaxed)) { cbm_store_free_lsp_surfaces(rows, file_count); /* untouched rows are all NULL */ return -1; diff --git a/src/pipeline/lsp_surface.h b/src/pipeline/lsp_surface.h index dc198bb654..1a73b34274 100644 --- a/src/pipeline/lsp_surface.h +++ b/src/pipeline/lsp_surface.h @@ -68,4 +68,11 @@ int cbm_lsp_surface_python_namespace_from_json(CBMArena *arena, const char *json */ int cbm_lsp_surface_python_change(const char *old_json, const char *new_json); +#if defined(CBM_ENABLE_TEST_SEAMS) && CBM_ENABLE_TEST_SEAMS +/* Bracket each file callback, including skipped/error rows. Install only with + * no surface build running; clear after the observed build returns. */ +typedef void (*cbm_lsp_surface_row_observer_fn)(int index, bool entering, void *ctx); +void cbm_lsp_surface_test_set_row_observer(cbm_lsp_surface_row_observer_fn observer, void *ctx); +#endif + #endif /* CBM_PIPELINE_LSP_SURFACE_H */ diff --git a/src/pipeline/pipeline.c b/src/pipeline/pipeline.c index 2727e41d19..657e25cd4f 100644 --- a/src/pipeline/pipeline.c +++ b/src/pipeline/pipeline.c @@ -81,6 +81,18 @@ static inline void *intptr_to_ptr(intptr_t v) { * Atomic spinlock: 0 = free, 1 = locked. */ static atomic_int g_pipeline_busy = 0; +#if defined(CBM_INCREMENTAL_TEST_API) && CBM_INCREMENTAL_TEST_API +static atomic_int g_last_worker_count = 0; + +void cbm_pipeline_worker_count_test_reset(void) { + atomic_store(&g_last_worker_count, 0); +} + +int cbm_pipeline_worker_count_test_last(void) { + return atomic_load(&g_last_worker_count); +} +#endif + #if defined(CBM_INCREMENTAL_TEST_API) && CBM_INCREMENTAL_TEST_API static atomic_bool g_persist_test_fail_after_stage_dump = false; static atomic_bool g_persist_test_cancel_after_predump = false; @@ -210,6 +222,7 @@ struct cbm_pipeline { bool persistence; /* write .codebase-memory/graph.db.zst after indexing */ cbm_index_resource_policy_t resource_policy; cbm_index_resource_violation_t resource_violation; + bool background; /* reserve CPU headroom when no user is waiting */ /* Snapshot of the artifact export failure of THIS run (set only by * export_after_publish failure, zeroed at run start, cleared on success). @@ -965,6 +978,12 @@ const char *cbm_pipeline_export_error(const cbm_pipeline_t *p) { return p ? p->export_error : ""; } +void cbm_pipeline_set_background(cbm_pipeline_t *p, bool background) { + if (p) { + p->background = background; + } +} + bool cbm_pipeline_set_project_name(cbm_pipeline_t *p, const char *name) { if (!p || !name || !name[0]) { return false; @@ -1829,17 +1848,42 @@ void cbm_pipeline_set_committed_counts(cbm_pipeline_t *p, int nodes, int edges) } } +/* Two workers is the smallest width that takes the parallel route. */ +enum { PL_PARALLEL_MIN_WORKERS = 2 }; + +/* Background refreshes leave CPU headroom, but never drop below two workers + * while the foreground width is two or more. WHY: one worker sends the indexer + * down the sequential route, which still builds a different graph than the + * parallel route (fewer Route and HANDLES edges, churned CALLS). Until the two + * routes agree, an automatic refresh must not change the graph only because + * the host has two or fewer performance cores. */ +int cbm_pipeline_background_worker_width(int foreground, int headroom) { + if (foreground >= PL_PARALLEL_MIN_WORKERS && headroom < PL_PARALLEL_MIN_WORKERS) { + return PL_PARALLEL_MIN_WORKERS; + } + return headroom; +} + /* Effective worker count. The crash supervisor re-runs its worker single- * threaded (CBM_INDEX_SINGLE_THREAD=1) so a per-file marker can pin the EXACT * crasher; a parallel re-run would race the marker. Honour that override * everywhere the worker count drives the parallel/sequential decision, so the * whole extraction phase collapses to the deterministic sequential path. */ -static int effective_worker_count(bool initial) { +int cbm_pipeline_worker_count(const cbm_pipeline_t *p) { const char *st = getenv("CBM_INDEX_SINGLE_THREAD"); + int workers; if (st && st[0] == '1') { - return 1; + workers = 1; + } else if (p && p->background) { + workers = cbm_pipeline_background_worker_width(cbm_default_worker_count(true), + cbm_default_worker_count(false)); + } else { + workers = cbm_default_worker_count(true); } - return cbm_default_worker_count(initial); +#if defined(CBM_INCREMENTAL_TEST_API) && CBM_INCREMENTAL_TEST_API + atomic_store(&g_last_worker_count, workers); +#endif + return workers; } /* Resolve the DB path for this pipeline. Caller must free(). */ @@ -4186,7 +4230,7 @@ static int run_githistory(cbm_pipeline_t *p, cbm_pipeline_ctx_t *ctx) { gh_compute_arg_t gh_arg = {.repo_path = ctx->repo_path, .result = &gh_result}; if (p->mode != CBM_MODE_FAST) { - if (effective_worker_count(true) > SKIP_ONE) { + if (cbm_pipeline_worker_count(p) > SKIP_ONE) { if (cbm_thread_create(&gh_thread, 0, gh_compute_thread_fn, &gh_arg) == 0) { gh_threaded = true; } @@ -4286,7 +4330,7 @@ static int run_structure_and_extraction(cbm_pipeline_t *p, cbm_pipeline_ctx_t *c return CBM_NOT_FOUND; } - int worker_count = p->frozen ? 1 : effective_worker_count(true); + int worker_count = p->frozen ? 1 : cbm_pipeline_worker_count(p); if (p->frozen) frozen_observe(p, 0, 1); CBM_PROF_START(t_extract_total); @@ -4417,7 +4461,7 @@ static int cbm_pipeline_run_staged(cbm_pipeline_t *p) { : mode_promoted ? cbm_pipeline_build_fresh_semantic_manifest(p, p->project_name, &baseline_manifest, &baseline_count) - : cbm_pipeline_build_semantic_manifest(p->project_name, p->repo_path, files, + : cbm_pipeline_build_semantic_manifest(p, p->project_name, p->repo_path, files, file_count, p->excluded_dirs, p->excluded_count, &p->git_ctx, p->userconfig, &baseline_manifest, &baseline_count); diff --git a/src/pipeline/pipeline.h b/src/pipeline/pipeline.h index 9fd6946f1b..2748298ed2 100644 --- a/src/pipeline/pipeline.h +++ b/src/pipeline/pipeline.h @@ -71,6 +71,23 @@ bool cbm_pipeline_doc_candidates_enabled(const cbm_pipeline_t *p); /* Copy the exact discovery violation from the most recent run. */ void cbm_pipeline_get_resource_violation(const cbm_pipeline_t *p, cbm_index_resource_violation_t *violation); +/* Mark work that runs without a waiting user. Background pipelines reserve + * CPU headroom; foreground pipelines retain the initial-index all-core policy. */ +void cbm_pipeline_set_background(cbm_pipeline_t *p, bool background); + +/* Resolve the worker policy for this pipeline, including environment and + * crash-recovery overrides. */ +int cbm_pipeline_worker_count(const cbm_pipeline_t *p); + +/* Background width from the foreground width and the headroom width: the + * headroom, floored at two whenever the foreground width is two or more so an + * automatic refresh keeps the parallel route. */ +int cbm_pipeline_background_worker_width(int foreground, int headroom); + +#if defined(CBM_INCREMENTAL_TEST_API) && CBM_INCREMENTAL_TEST_API +void cbm_pipeline_worker_count_test_reset(void); +int cbm_pipeline_worker_count_test_last(void); +#endif /* Snapshot of the artifact export failure of the last cbm_pipeline_run, or "" * when the run succeeded / did not reach post-publish export. Used to diff --git a/src/pipeline/pipeline_incremental.c b/src/pipeline/pipeline_incremental.c index d15bd7349a..6e8b398312 100644 --- a/src/pipeline/pipeline_incremental.c +++ b/src/pipeline/pipeline_incremental.c @@ -479,9 +479,9 @@ static void *manifest_hash_worker(void *arg) { enum { MANIFEST_PARALLEL_MIN_FILES = 64 }; -int cbm_pipeline_build_semantic_manifest(const char *project, const char *repo_path, - const cbm_file_info_t *files, int file_count, - char **excluded_dirs, int excluded_count, +int cbm_pipeline_build_semantic_manifest(const cbm_pipeline_t *p, const char *project, + const char *repo_path, const cbm_file_info_t *files, + int file_count, char **excluded_dirs, int excluded_count, const cbm_git_context_t *git_ctx, const cbm_userconfig_t *userconfig, cbm_file_hash_t **out, int *out_count) { @@ -522,7 +522,7 @@ int cbm_pipeline_build_semantic_manifest(const char *project, const char *repo_p } struct timespec t_hash; cbm_clock_gettime(CLOCK_MONOTONIC, &t_hash); - int hash_workers = cbm_default_worker_count(true); + int hash_workers = cbm_pipeline_worker_count(p); if (rc == 0 && file_count >= MANIFEST_PARALLEL_MIN_FILES && hash_workers > SKIP_ONE) { char (*shas)[CBM_SHA256_HEX_LEN + 1] = malloc((size_t)file_count * sizeof(*shas)); int64_t *mtimes = calloc((size_t)file_count, sizeof(int64_t)); @@ -869,9 +869,9 @@ int cbm_pipeline_build_fresh_semantic_manifest(cbm_pipeline_t *p, const char *pr &fresh_ignored_total); } if (rc == 0) { - rc = cbm_pipeline_build_semantic_manifest(project, repo_path, fresh_files, fresh_file_count, - fresh_excluded, fresh_excluded_count, - &fresh_git_ctx, fresh_userconfig, out, out_count); + rc = cbm_pipeline_build_semantic_manifest( + p, project, repo_path, fresh_files, fresh_file_count, fresh_excluded, + fresh_excluded_count, &fresh_git_ctx, fresh_userconfig, out, out_count); } cbm_set_user_lang_config(previous_userconfig); cbm_git_context_free(&fresh_git_ctx); @@ -1849,7 +1849,8 @@ static int run_extract_resolve(cbm_pipeline_ctx_t *ctx, cbm_file_info_t *changed * a full build takes, which is what makes its output converge. */ #define MIN_FILES_FOR_PARALLEL_INCR 50 - int worker_count = cbm_pipeline_is_frozen(ctx->pipeline) ? 1 : cbm_default_worker_count(true); + int worker_count = + cbm_pipeline_is_frozen(ctx->pipeline) ? 1 : cbm_pipeline_worker_count(ctx->pipeline); bool use_parallel = closure != NULL || (worker_count > SKIP_ONE && ci > MIN_FILES_FOR_PARALLEL_INCR); @@ -2335,7 +2336,7 @@ static int closure_probe_surfaces(cbm_pipeline_t *p, const char *project, _Atomic int64_t probe_ids; atomic_init(&probe_ids, cbm_gbuf_next_id(probe_gbuf)); rc = cbm_parallel_extract(&probe_ctx, probe_files, probe_count, cache, &probe_ids, - cbm_default_worker_count(true)); + cbm_pipeline_worker_count(p)); } if (rc == 0) { char **def_modules = @@ -2349,10 +2350,14 @@ static int closure_probe_surfaces(cbm_pipeline_t *p, const char *project, if (def_modules && def_starts) { defs = cbm_pxc_collect_all_defs(NULL, &probe_arena, cache, probe_files, probe_count, project, def_modules, &def_count, def_starts); - rc = def_count < 0 - ? CBM_PIPELINE_ABORT_PRESERVE_DB - : cbm_lsp_surface_build_rows(NULL, project, cache, probe_files, probe_count, - defs, def_starts, out_rows, out_count); + /* Extract left pipeline NULL so file errors are not recorded twice. + * Surface rows only borrow it for the background worker policy. */ + probe_ctx.pipeline = p; + rc = def_count < 0 ? CBM_PIPELINE_ABORT_PRESERVE_DB + : cbm_lsp_surface_build_rows(&probe_ctx, project, cache, probe_files, + probe_count, defs, def_starts, out_rows, + out_count); + probe_ctx.pipeline = NULL; } else { rc = -1; } @@ -3478,7 +3483,7 @@ static int run_closure_delta(cbm_pipeline_t *p, const char *db_path, const char elig[elig_count++] = row; } } - int workers = cbm_default_worker_count(true); + int workers = cbm_pipeline_worker_count(p); if (workers < 1) { workers = 1; } diff --git a/src/pipeline/pipeline_internal.h b/src/pipeline/pipeline_internal.h index 80485c7e89..04198cda48 100644 --- a/src/pipeline/pipeline_internal.h +++ b/src/pipeline/pipeline_internal.h @@ -1571,9 +1571,9 @@ bool cbm_pipeline_incremental_generation_exact(cbm_pipeline_t *p, cbm_store_t *s #define CBM_SEMANTIC_INPUT_GLOBAL_CONFIG CBM_SEMANTIC_INPUT_PREFIX "global-extension-config-v1" #define CBM_SEMANTIC_INPUT_PROJECT_CONFIG CBM_SEMANTIC_INPUT_PREFIX "project-extension-config-v1" -int cbm_pipeline_build_semantic_manifest(const char *project, const char *repo_path, - const cbm_file_info_t *files, int file_count, - char **excluded_dirs, int excluded_count, +int cbm_pipeline_build_semantic_manifest(const cbm_pipeline_t *p, const char *project, + const char *repo_path, const cbm_file_info_t *files, + int file_count, char **excluded_dirs, int excluded_count, const cbm_git_context_t *git_ctx, const cbm_userconfig_t *userconfig, cbm_file_hash_t **out, int *out_count); diff --git a/src/pipeline/worker_pool.c b/src/pipeline/worker_pool.c index 2943fd746a..96611bf8dd 100644 --- a/src/pipeline/worker_pool.c +++ b/src/pipeline/worker_pool.c @@ -22,9 +22,28 @@ enum { WP_TRUE = 1, WP_MIN = 1, WP_STEP = 1 }; /* CBM_WORKER_STACK_SIZE (8 MB per worker) lives in worker_pool.h. */ +#if defined(CBM_ENABLE_TEST_SEAMS) && CBM_ENABLE_TEST_SEAMS +static cbm_parallel_spawn_observer_fn g_spawn_observer; +static void *g_spawn_observer_ctx; + +void cbm_parallel_test_set_spawn_observer(cbm_parallel_spawn_observer_fn observer, void *ctx) { + g_spawn_observer = observer; + g_spawn_observer_ctx = ctx; +} + +static void observe_spawn(int count, int helpers) { + if (g_spawn_observer) { + g_spawn_observer(count, helpers, g_spawn_observer_ctx); + } +} +#else +#define observe_spawn(count, helpers) ((void)0) +#endif + /* ── Serial fallback ─────────────────────────────────────────────── */ static void run_serial(int count, cbm_parallel_fn fn, void *ctx) { + observe_spawn(count, 0); for (int i = 0; i < count; i++) { fn(i, ctx); } @@ -103,6 +122,7 @@ static void run_pthreads_counted(int count, cbm_parallel_fn fn, void *ctx, int n } started++; } + observe_spawn(count, started); int main_done = 0; uint64_t main_ops_before = cbm_memev_thread_ops(); while (WP_TRUE) { @@ -164,6 +184,7 @@ static void run_pthreads(int count, cbm_parallel_fn fn, void *ctx, int nworkers) } } + observe_spawn(count, nworkers); /* Main thread also participates */ while (WP_TRUE) { int idx = atomic_fetch_add_explicit(&next_idx, WP_STEP, memory_order_relaxed); diff --git a/src/pipeline/worker_pool.h b/src/pipeline/worker_pool.h index 91697b961d..49b3efd1f1 100644 --- a/src/pipeline/worker_pool.h +++ b/src/pipeline/worker_pool.h @@ -36,4 +36,11 @@ typedef struct { * If count <= 1 or workers <= 1, runs single-threaded. */ void cbm_parallel_for(int count, cbm_parallel_fn fn, void *ctx, cbm_parallel_for_opts_t opts); +#if defined(CBM_ENABLE_TEST_SEAMS) && CBM_ENABLE_TEST_SEAMS +/* Observe the completed spawn attempt before the caller takes work. Install + * only while no dispatch is running; clear after the observed dispatch joins. */ +typedef void (*cbm_parallel_spawn_observer_fn)(int count, int helpers, void *ctx); +void cbm_parallel_test_set_spawn_observer(cbm_parallel_spawn_observer_fn observer, void *ctx); +#endif + #endif /* CBM_WORKER_POOL_H */ diff --git a/tests/test_daemon_application.c b/tests/test_daemon_application.c index 5bb032a5cd..7103fce7f7 100644 --- a/tests/test_daemon_application.c +++ b/tests/test_daemon_application.c @@ -1662,6 +1662,7 @@ typedef struct { * itself instead, which is the edge a reader on another thread needs. */ atomic_bool args_captured[APP_FAKE_MAX_ATTEMPTS]; size_t memory_budgets[APP_FAKE_MAX_ATTEMPTS]; + bool background_requests[APP_FAKE_MAX_ATTEMPTS]; } app_fake_worker_context_t; typedef struct { @@ -1729,6 +1730,8 @@ static int app_fake_worker_start(void *opaque, const char *args_json, size_t mem args_json ? args_json : ""); atomic_store(&context->args_captured[worker->attempt], true); context->memory_budgets[worker->attempt] = memory_budget_bytes; + context->background_requests[worker->attempt] = + cbm_mcp_get_bool_arg(args_json, "_cbm_background"); if (marker_file) { (void)snprintf(context->marker_paths[worker->attempt], APP_TEST_PATH_CAP, "%s", marker_file); @@ -2546,6 +2549,7 @@ TEST(daemon_application_initialize_coalesces_auto_index_for_full_sessions) { ASSERT_TRUE(first_initialized); ASSERT_TRUE(first_owned); ASSERT_TRUE(auto_policy_propagated); + ASSERT_TRUE(fake.background_requests[0]); ASSERT_TRUE(second_initialized); ASSERT_TRUE(coalesced); ASSERT_TRUE(restricted_disconnect_kept_job); @@ -4169,6 +4173,7 @@ TEST(daemon_application_request_cancel_preserves_persistent_watch_and_session) { ASSERT_TRUE(thread_started); ASSERT_TRUE(subscribed); ASSERT_TRUE(worker_started); + ASSERT_FALSE(fixture.fake.background_requests[0]); ASSERT_TRUE(request_returned); ASSERT_TRUE(thread_joined); ASSERT_EQ(request.status, CBM_DAEMON_RUNTIME_APPLICATION_CANCELLED); @@ -4466,6 +4471,7 @@ TEST(daemon_application_final_cancel_drains_admitted_watcher_job) { ASSERT_TRUE(fixture_ready); ASSERT_TRUE(thread_started); ASSERT_TRUE(worker_started); + ASSERT_TRUE(fixture.fake.background_requests[0]); ASSERT_TRUE(job_active); ASSERT_EQ(watches_after_cancel, 0); ASSERT_TRUE(thread_joined); @@ -6928,8 +6934,9 @@ typedef struct { uint32_t context_length; } app_async_fixture_t; -static bool app_async_fixture_init(app_async_fixture_t *fixture, const char *label, - bool permanent) { +/* stored_config is optional; a test that needs auto_index passes its own. */ +static bool app_async_fixture_init_with_config(app_async_fixture_t *fixture, const char *label, + bool permanent, cbm_config_t *stored_config) { memset(fixture, 0, sizeof(*fixture)); app_fake_worker_context_init(&fixture->fake); fixture->ops = (cbm_daemon_application_worker_ops_t){ @@ -6940,7 +6947,8 @@ static bool app_async_fixture_init(app_async_fixture_t *fixture, const char *lab .log_path = app_fake_worker_log_path, .destroy = app_fake_worker_destroy, }; - cbm_daemon_application_config_t config = {.worker_ops = &fixture->ops}; + cbm_daemon_application_config_t config = {.config = stored_config, + .worker_ops = &fixture->ops}; fixture->application = cbm_daemon_application_new(&config); if (!fixture->application) { return false; @@ -6961,6 +6969,11 @@ static bool app_async_fixture_init(app_async_fixture_t *fixture, const char *lab &fixture->context_length); } +static bool app_async_fixture_init(app_async_fixture_t *fixture, const char *label, + bool permanent) { + return app_async_fixture_init_with_config(fixture, label, permanent, NULL); +} + static cbm_daemon_runtime_application_session_t *app_async_session(app_async_fixture_t *fixture, cbm_daemon_client_id_t id) { cbm_daemon_runtime_application_session_t *session = app_test_open(&fixture->callbacks, id); @@ -7338,6 +7351,78 @@ TEST(daemon_application_cut_short_sync_index_advice_reaches_next_call) { * the daemon's only live session is refused async (the daemon would stop and * cancel the job as soon as the command exits), while status stays allowed and * a long-lived MCP session on the same temporary daemon remains a valid host. */ +/* Automatic indexing (session auto-index, watcher re-index) marks its job + * _cbm_background so the worker leaves CPU headroom. The marker picks the + * worker count, never what is indexed, so an explicit index_repository for the + * same root still JOINS the automatic job already running for it, the contract + * the repo_path fold above restored, instead of being refused as an options + * conflict. The worker is held, so the join is observed while the job runs. */ +TEST(daemon_application_explicit_index_joins_running_auto_index_job) { + app_env_backup_t cache_environment; + bool cache_saved = app_env_backup_capture(&cache_environment, "CBM_CACHE_DIR"); + char cache[APP_TEST_PATH_CAP]; + (void)snprintf(cache, sizeof(cache), "%s/cbm-app-auto-join-cache-XXXXXX", cbm_tmpdir()); + bool cache_ready = + cache_saved && cbm_mkdtemp(cache) != NULL && cbm_setenv("CBM_CACHE_DIR", cache, 1) == 0; + cbm_config_t *stored_config = cache_ready ? cbm_config_open(cache) : NULL; + bool config_ready = stored_config && + cbm_config_set(stored_config, CBM_CONFIG_AUTO_INDEX, "true") == 0 && + cbm_config_set(stored_config, CBM_CONFIG_AUTO_WATCH, "false") == 0; + app_async_fixture_t fixture; + bool setup = config_ready && + app_async_fixture_init_with_config(&fixture, "auto-join", true, stored_config); + + cbm_daemon_runtime_application_session_t *automatic = + setup ? app_test_open(&fixture.callbacks, 7801) : NULL; + bool auto_started = automatic && + app_test_initialize_profile(&fixture.callbacks, automatic, fixture.root, + CBM_MCP_TOOL_PROFILE_ALL, NULL, NULL) && + app_wait_for_atomic_int(&fixture.fake.starts, 1); + /* args_json is published by the args_captured release store; the + * background_requests slot is written after it, so read the args. */ + bool auto_marked = auto_started && + app_wait_for_atomic_bool(&fixture.fake.args_captured[0], true) && + cbm_mcp_get_bool_arg(fixture.fake.args_json[0], "_cbm_background"); + + cbm_daemon_runtime_application_session_t *explicit_session = + auto_marked ? app_async_session(&fixture, 7802) : NULL; + char *joined = + explicit_session ? app_async_call_now(&fixture, explicit_session, ",\"async\":true") : NULL; + int starts_while_running = atomic_load(&fixture.fake.starts); + size_t active_while_running = + setup ? cbm_daemon_application_active_jobs(fixture.application) : 0; + + if (explicit_session) { + fixture.callbacks.session_close(fixture.callbacks.context, explicit_session); + } + if (automatic) { + fixture.callbacks.session_cancel(fixture.callbacks.context, automatic); + fixture.callbacks.session_close(fixture.callbacks.context, automatic); + } + bool stopped = !setup || app_async_fixture_finish(&fixture); + cbm_config_close(stored_config); + bool cleaned = !cache_ready || th_rmtree(cache) == 0; + bool cache_restored = app_env_backup_restore(&cache_environment); + + ASSERT_TRUE(cache_ready); + ASSERT_TRUE(config_ready); + ASSERT_TRUE(setup); + ASSERT_TRUE(auto_started); + ASSERT_TRUE(auto_marked); + ASSERT_NOT_NULL(explicit_session); + ASSERT_NOT_NULL(joined); + ASSERT_FALSE(app_async_contains(joined, "different options")); + ASSERT_FALSE(app_async_contains(joined, "\"isError\":true")); + ASSERT_TRUE(app_async_contains(joined, "\"joined\":true")); + ASSERT_EQ(starts_while_running, 1); + ASSERT_EQ(active_while_running, 1); + ASSERT_TRUE(stopped); + ASSERT_TRUE(cleaned); + ASSERT_TRUE(cache_restored); + free(joined); + PASS(); +} + TEST(daemon_application_async_refused_for_lone_one_shot_client_of_temporary_daemon) { app_async_fixture_t fixture; bool setup = app_async_fixture_init(&fixture, "async-refuse", false); @@ -7666,6 +7751,7 @@ SUITE(daemon_application) { RUN_TEST(daemon_application_async_and_status_validation_errors); RUN_TEST(daemon_application_cancelled_sync_index_reply_offers_async); RUN_TEST(daemon_application_cut_short_sync_index_advice_reaches_next_call); + RUN_TEST(daemon_application_explicit_index_joins_running_auto_index_job); RUN_TEST(daemon_application_async_refused_for_lone_one_shot_client_of_temporary_daemon); RUN_TEST(daemon_application_unnamed_status_finds_async_job_of_root_owner_issue2134); RUN_TEST(daemon_application_unnamed_status_keeps_path_derived_key_issue2134); diff --git a/tests/test_mcp.c b/tests/test_mcp.c index bb3227131c..3b5c62e2aa 100644 --- a/tests/test_mcp.c +++ b/tests/test_mcp.c @@ -24153,6 +24153,161 @@ TEST(autoindex_limit_guards_git_root_issue713) { PASS(); } +/* A servable project row is a committed index. user_version stays 0, so a + * later run cannot take the incremental no-op and skip recording workers. */ +static bool mcp_publish_committed_index(const char *root) { + char *project = root ? cbm_project_name_from_path(root) : NULL; + cbm_store_t *store = project ? cbm_store_open(project) : NULL; + bool published = store && cbm_store_upsert_project(store, project, root) == CBM_STORE_OK; + cbm_store_close(store); + free(project); + return published; +} + +#ifdef CBM_ENABLE_TEST_SEAMS +typedef struct { + const char *root; + bool published; +} mcp_committed_seed_t; + +/* maybe_auto_index returns before the thread when .db already + * exists. Publish the committed index from the count hook, which runs after + * that check and before autoindex_thread chooses a width. */ +static void mcp_autoindex_publish_committed(void *context) { + mcp_committed_seed_t *seed = context; + if (seed) { + seed->published = mcp_publish_committed_index(seed->root); + } +} +#endif + +/* First index, including one started automatically, uses every core. Headroom + * applies only once a servable project database already exists. + * automatic: in-process auto-index (initialize → autoindex_thread). + * Otherwise index_repository with _cbm_background, the daemon/watcher path. */ +static int mcp_assert_inprocess_worker_policy(bool committed, bool automatic) { + char cache[256]; + char repo[512]; + (void)snprintf(cache, sizeof(cache), "%s/cbm-worker-policy-%d%d-XXXXXX", cbm_tmpdir(), + committed ? 1 : 0, automatic ? 1 : 0); + bool cache_ready = cbm_mkdtemp(cache) != NULL; + (void)snprintf(repo, sizeof(repo), "%s/repo", cache); + char source[640]; + (void)snprintf(source, sizeof(source), "%s/main.py", repo); + bool repo_ready = cache_ready && th_mkdir_p(repo) == 0 && + th_write_file(source, "def background_index():\n return True\n") == 0; + + mcp_test_env_backup_t environment[] = { + {.name = "CBM_CACHE_DIR"}, + {.name = "CBM_WORKERS"}, + {.name = "CBM_INDEX_SINGLE_THREAD"}, + }; + bool environment_saved = true; + for (size_t i = 0; i < sizeof(environment) / sizeof(environment[0]); i++) { + const char *value = getenv(environment[i].name); + environment[i].present = value != NULL; + environment[i].value = value ? strdup(value) : NULL; + environment_saved = environment_saved && (!value || environment[i].value); + } + bool environment_ready = environment_saved && cbm_setenv("CBM_CACHE_DIR", cache, 1) == 0 && + cbm_unsetenv("CBM_WORKERS") == 0 && + cbm_unsetenv("CBM_INDEX_SINGLE_THREAD") == 0; + + bool seeded = !committed; + char old_cwd[CBM_SZ_4K] = {0}; + bool cwd_ready = true; + cbm_config_t *config = NULL; + bool config_ready = true; + cbm_mcp_server_t *server = NULL; + char *response = NULL; + if (automatic) { + cwd_ready = repo_ready && environment_ready && cbm_getcwd(old_cwd, sizeof(old_cwd)) && + cbm_chdir(repo) == 0; + config = cwd_ready ? cbm_config_open(cache) : NULL; + config_ready = config && cbm_config_set(config, CBM_CONFIG_AUTO_INDEX, "true") == 0 && + cbm_config_set(config, CBM_CONFIG_AUTO_WATCH, "false") == 0; + cbm_pipeline_worker_count_test_reset(); + server = config_ready ? cbm_mcp_server_new(NULL) : NULL; + if (server) { + cbm_mcp_server_set_config(server, config); +#ifdef CBM_ENABLE_TEST_SEAMS + mcp_committed_seed_t seed = {.root = repo, .published = !committed}; + if (committed) { + cbm_mcp_server_set_auto_index_count_test_hook( + server, mcp_autoindex_publish_committed, &seed); + } +#endif + response = cbm_mcp_server_handle( + server, "{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"initialize\",\"params\":{}}"); +#ifdef CBM_ENABLE_TEST_SEAMS + seeded = seed.published; +#endif + cbm_mcp_server_free(server); /* joins the automatic index thread */ + } + } else { + if (committed && repo_ready && environment_ready) { + seeded = mcp_publish_committed_index(repo); + } + cbm_pipeline_worker_count_test_reset(); + server = repo_ready && environment_ready && seeded ? cbm_mcp_server_new(NULL) : NULL; + if (server) { + char repo_json[512]; + char args[CBM_SZ_1K]; + (void)snprintf(repo_json, sizeof(repo_json), "%s", repo); + cbm_normalize_path_sep(repo_json); + (void)snprintf(args, sizeof(args), "{\"_cbm_background\":true,\"repo_path\":\"%s\"}", + repo_json); + response = cbm_mcp_handle_tool(server, "index_repository", args); + cbm_mcp_server_free(server); + } + } + int selected_workers = cbm_pipeline_worker_count_test_last(); + bool server_ready = server != NULL; + bool response_ready = response != NULL; + + free(response); + cbm_config_close(config); + if (automatic && cwd_ready) { + (void)cbm_chdir(old_cwd); + } + /* Capture while CBM_WORKERS is still unset. Restoring first makes the + * expected count follow the lane override instead of this policy. */ + int expected = committed ? cbm_pipeline_background_worker_width(cbm_default_worker_count(true), + cbm_default_worker_count(false)) + : cbm_default_worker_count(true); + mcp_test_restore_env(environment, sizeof(environment) / sizeof(environment[0])); + bool cleaned = !cache_ready || th_rmtree(cache) == 0; + + ASSERT_TRUE(cache_ready); + ASSERT_TRUE(repo_ready); + ASSERT_TRUE(environment_saved); + ASSERT_TRUE(environment_ready); + ASSERT_TRUE(cwd_ready); + ASSERT_TRUE(config_ready); + ASSERT_TRUE(seeded); + ASSERT_TRUE(server_ready); + ASSERT_TRUE(response_ready); + ASSERT_EQ(selected_workers, expected); + ASSERT_TRUE(cleaned); + PASS(); +} + +TEST(mcp_auto_index_in_process_fresh_project_uses_full_width) { + return mcp_assert_inprocess_worker_policy(false, true); +} + +TEST(mcp_auto_index_in_process_existing_index_keeps_headroom) { + return mcp_assert_inprocess_worker_policy(true, true); +} + +TEST(mcp_background_flag_fresh_project_uses_full_width) { + return mcp_assert_inprocess_worker_policy(false, false); +} + +TEST(mcp_background_flag_existing_index_keeps_headroom) { + return mcp_assert_inprocess_worker_policy(true, false); +} + /* ══════════════════════════════════════════════════════════════════ * #853 — auto_watch=false must ALSO gate the SUPERVISED fresh-index * watcher registration (keystone × #849 merge interaction) @@ -26501,6 +26656,10 @@ SUITE(mcp) { RUN_TEST(autoindex_limit_guards_non_git_root_issue713); RUN_TEST(autoindex_limit_admits_non_git_root_under_limit_issue713); RUN_TEST(autoindex_limit_guards_git_root_issue713); + RUN_TEST(mcp_auto_index_in_process_fresh_project_uses_full_width); + RUN_TEST(mcp_auto_index_in_process_existing_index_keeps_headroom); + RUN_TEST(mcp_background_flag_fresh_project_uses_full_width); + RUN_TEST(mcp_background_flag_existing_index_keeps_headroom); RUN_TEST(search_code_context_window_is_bounded); RUN_TEST(search_code_context_extreme_values_answer_normally); RUN_TEST(tool_query_graph_max_rows_above_ceiling_is_capped); diff --git a/tests/test_pipeline.c b/tests/test_pipeline.c index 992f5ae78a..59b8bf9d79 100644 --- a/tests/test_pipeline.c +++ b/tests/test_pipeline.c @@ -13,6 +13,7 @@ #include "lsp/c_lsp.h" #include "lsp/py_lsp.h" #include "pipeline/lsp_surface.h" +#include "pipeline/worker_pool.h" #include "pipeline/pass_lsp_cross.h" #include "pipeline/pipeline_internal.h" #include "pipeline/helm_template.h" @@ -122,6 +123,269 @@ TEST(pipeline_create_free) { PASS(); } +TEST(pipeline_background_worker_policy_preserves_overrides) { + const char *old_workers = getenv("CBM_WORKERS"); + char *saved_workers = old_workers ? strdup(old_workers) : NULL; + const char *old_single = getenv("CBM_INDEX_SINGLE_THREAD"); + char *saved_single = old_single ? strdup(old_single) : NULL; + cbm_unsetenv("CBM_WORKERS"); + cbm_unsetenv("CBM_INDEX_SINGLE_THREAD"); + + cbm_pipeline_t *foreground = cbm_pipeline_new("/some/path", NULL, CBM_MODE_FULL); + cbm_pipeline_t *background = cbm_pipeline_new("/some/path", NULL, CBM_MODE_FULL); + ASSERT_NOT_NULL(foreground); + ASSERT_NOT_NULL(background); + cbm_pipeline_set_background(background, true); + + int foreground_default = cbm_pipeline_worker_count(foreground); + int background_default = cbm_pipeline_worker_count(background); + bool defaults_match = foreground_default == cbm_default_worker_count(true) && + background_default == cbm_pipeline_background_worker_width( + cbm_default_worker_count(true), + cbm_default_worker_count(false)) && + foreground_default >= 1 && background_default >= 1 && + (foreground_default < 2 || background_default >= 2); + + /* A pinned single worker leaves no parallel route to preserve. */ + cbm_setenv("CBM_WORKERS", "1", 1); + bool env_override = + cbm_pipeline_worker_count(foreground) == 1 && cbm_pipeline_worker_count(background) == 1; + cbm_setenv("CBM_WORKERS", "3", 1); + env_override = env_override && cbm_pipeline_worker_count(foreground) == 3 && + cbm_pipeline_worker_count(background) == 3; + + cbm_setenv("CBM_INDEX_SINGLE_THREAD", "1", 1); + bool recovery_override = cbm_pipeline_worker_count(foreground) == 1 && + cbm_pipeline_worker_count(background) == 1; + + cbm_pipeline_free(foreground); + cbm_pipeline_free(background); + saved_workers ? cbm_setenv("CBM_WORKERS", saved_workers, 1) : cbm_unsetenv("CBM_WORKERS"); + saved_single ? cbm_setenv("CBM_INDEX_SINGLE_THREAD", saved_single, 1) + : cbm_unsetenv("CBM_INDEX_SINGLE_THREAD"); + free(saved_workers); + free(saved_single); + + ASSERT_TRUE(defaults_match); + ASSERT_TRUE(env_override); + ASSERT_TRUE(recovery_override); + PASS(); +} + +/* An automatic refresh keeps CPU headroom but never falls to one worker, the + * sequential route, while the foreground width could run in parallel. Each + * host shape is given as the two widths cbm_default_worker_count() derives + * from it (foreground: every core; headroom: performance cores minus one, at + * least one), so the verdict does not depend on the machine running it. */ +TEST(pipeline_background_worker_width_keeps_parallel_route) { + /* One core: foreground 1, headroom 1. Nothing to preserve. */ + ASSERT_EQ(cbm_pipeline_background_worker_width(1, 1), 1); + /* Two performance cores: headroom alone would be 1. */ + ASSERT_EQ(cbm_pipeline_background_worker_width(2, 1), 2); + /* Two performance cores beside efficiency cores: same floor. */ + ASSERT_EQ(cbm_pipeline_background_worker_width(8, 1), 2); + /* Larger hosts keep their headroom unchanged. */ + ASSERT_EQ(cbm_pipeline_background_worker_width(3, 2), 2); + ASSERT_EQ(cbm_pipeline_background_worker_width(4, 3), 3); + ASSERT_EQ(cbm_pipeline_background_worker_width(18, 5), 5); + PASS(); +} + +#if defined(CBM_ENABLE_TEST_SEAMS) && CBM_ENABLE_TEST_SEAMS +enum { SURFACE_BUDGET_FILES = 5 }; + +typedef struct { + atomic_int expected; + atomic_int arrivals; + atomic_int live; + atomic_int peak; + atomic_bool released; + atomic_int visits[SURFACE_BUDGET_FILES]; + int helpers; +} surface_budget_probe_t; + +static void surface_budget_spawned(int count, int helpers, void *arg) { + surface_budget_probe_t *probe = arg; + probe->helpers = helpers; + int participants = helpers + 1; + atomic_store(&probe->expected, count < participants ? count : participants); +} + +static void surface_budget_row(int index, bool entering, void *arg) { + surface_budget_probe_t *probe = arg; + if (!entering) { + atomic_fetch_sub(&probe->live, 1); + return; + } + atomic_fetch_add(&probe->visits[index], 1); + int live = atomic_fetch_add(&probe->live, 1) + 1; + int peak = atomic_load(&probe->peak); + while (peak < live && !atomic_compare_exchange_weak(&probe->peak, &peak, live)) {} + atomic_fetch_add(&probe->arrivals, 1); + /* No timing window: hold the first wave until every possible participant + * has entered. The post-spawn observation accounts for partial thread + * creation and serial fallback, so neither can strand this gate. The test + * runner's watchdog bounds a broken dispatch without making time a verdict. */ + while (!atomic_load(&probe->released)) { + int expected = atomic_load(&probe->expected); + if (expected > 0 && atomic_load(&probe->arrivals) >= expected) { + atomic_store(&probe->released, true); + } + } +} + +TEST(pipeline_surface_respects_total_worker_budget) { + CBMFileResult results[SURFACE_BUDGET_FILES] = {0}; + CBMFileResult *cache[SURFACE_BUDGET_FILES]; + cbm_file_info_t files[SURFACE_BUDGET_FILES] = { + {.rel_path = "a.go", .language = CBM_LANG_GO}, + {.rel_path = "b.go", .language = CBM_LANG_GO}, + {.rel_path = "c.go", .language = CBM_LANG_GO}, + {.rel_path = "d.go", .language = CBM_LANG_GO}, + {.rel_path = "e.go", .language = CBM_LANG_GO}, + }; + surface_budget_probe_t probe = {0}; + for (int i = 0; i < SURFACE_BUDGET_FILES; i++) { + cache[i] = &results[i]; + atomic_init(&probe.visits[i], 0); + } + cache[1] = NULL; /* The skipped row must not change stable file ordering. */ + atomic_init(&probe.expected, 0); + atomic_init(&probe.arrivals, 0); + atomic_init(&probe.live, 0); + atomic_init(&probe.peak, 0); + atomic_init(&probe.released, false); + const char *old_workers = getenv("CBM_WORKERS"); + char *saved_workers = old_workers ? strdup(old_workers) : NULL; + const char *old_single = getenv("CBM_INDEX_SINGLE_THREAD"); + char *saved_single = old_single ? strdup(old_single) : NULL; + cbm_setenv("CBM_WORKERS", "2", 1); + cbm_unsetenv("CBM_INDEX_SINGLE_THREAD"); + cbm_pipeline_t *pipeline = cbm_pipeline_new("/some/path", NULL, CBM_MODE_FULL); + cbm_pipeline_set_background(pipeline, true); + cbm_pipeline_ctx_t ctx = {.pipeline = pipeline}; + cbm_lsp_surface_row_t *rows = NULL; + int count = 0; + cbm_parallel_test_set_spawn_observer(surface_budget_spawned, &probe); + cbm_lsp_surface_test_set_row_observer(surface_budget_row, &probe); + int rc = cbm_lsp_surface_build_rows(&ctx, "budget", cache, files, SURFACE_BUDGET_FILES, NULL, + NULL, &rows, &count); + int parallel_peak = atomic_load(&probe.peak); + int parallel_helpers = probe.helpers; + atomic_store(&probe.expected, 0); + atomic_store(&probe.arrivals, 0); + atomic_store(&probe.peak, 0); + atomic_store(&probe.released, false); + cbm_setenv("CBM_INDEX_SINGLE_THREAD", "1", 1); + cbm_lsp_surface_row_t *serial_rows = NULL; + int serial_count = 0; + int serial_rc = cbm_lsp_surface_build_rows(&ctx, "budget", cache, files, SURFACE_BUDGET_FILES, + NULL, NULL, &serial_rows, &serial_count); + cbm_lsp_surface_test_set_row_observer(NULL, NULL); + cbm_parallel_test_set_spawn_observer(NULL, NULL); + bool same = rc == 0 && serial_rc == 0 && rows && serial_rows && + count == SURFACE_BUDGET_FILES - 1 && count == serial_count; + for (int i = 0; same && i < count; i++) { + same = strcmp(rows[i].rel_path, files[i < 1 ? i : i + 1].rel_path) == 0 && + strcmp(rows[i].defs_json, serial_rows[i].defs_json) == 0 && + strcmp(rows[i].surface_sha, serial_rows[i].surface_sha) == 0; + } + cbm_store_free_lsp_surfaces(rows, count); + cbm_store_free_lsp_surfaces(serial_rows, serial_count); + cbm_pipeline_free(pipeline); + saved_workers ? cbm_setenv("CBM_WORKERS", saved_workers, 1) : cbm_unsetenv("CBM_WORKERS"); + saved_single ? cbm_setenv("CBM_INDEX_SINGLE_THREAD", saved_single, 1) + : cbm_unsetenv("CBM_INDEX_SINGLE_THREAD"); + free(saved_workers); + free(saved_single); + + ASSERT_EQ(rc, 0); + ASSERT_EQ(serial_rc, 0); + ASSERT_TRUE(same); + ASSERT_GTE(parallel_helpers, 1); /* A serial setup is not a concurrency repro. */ + ASSERT_EQ(probe.helpers, 0); + ASSERT_EQ(atomic_load(&probe.live), 0); + for (int i = 0; i < SURFACE_BUDGET_FILES; i++) { + ASSERT_EQ(atomic_load(&probe.visits[i]), 2); + } + ASSERT_EQ(atomic_load(&probe.peak), 1); + ASSERT_EQ(parallel_peak, 2); + PASS(); +} +#endif + +TEST(pipeline_background_policy_preserves_index_results) { + ASSERT_EQ(setup_test_repo(), 0); + /* 64 fillers plus setup_test_repo's three files exceed MIN_FILES_FOR_PARALLEL + * (50). Below that threshold both policies take the sequential path, so the + * comparison stays green even with the background policy removed. */ + for (int i = 0; i < 64; i++) { + char name[32]; + char body[96]; + (void)snprintf(name, sizeof(name), "pad%02d.go", i); + (void)snprintf(body, sizeof(body), "package main\n\nfunc Pad%02d() int {\n\treturn %d\n}\n", + i, i); + if (th_write_file(TH_PATH(g_tmpdir, name), body) != 0) { + teardown_test_repo(); + FAIL("failed to pad fixture past MIN_FILES_FOR_PARALLEL"); + } + } + + /* An inherited CBM_WORKERS pin makes both runs use the same count. Drop it + * so this compares an all-core index against a headroom index. */ + const char *old_workers = getenv("CBM_WORKERS"); + char *saved_workers = old_workers ? strdup(old_workers) : NULL; + const char *old_single = getenv("CBM_INDEX_SINGLE_THREAD"); + char *saved_single = old_single ? strdup(old_single) : NULL; + if ((old_workers && !saved_workers) || (old_single && !saved_single)) { + free(saved_workers); + free(saved_single); + teardown_test_repo(); + FAIL("failed to save worker env"); + } + cbm_unsetenv("CBM_WORKERS"); + cbm_unsetenv("CBM_INDEX_SINGLE_THREAD"); + + char foreground_db[512]; + char background_db[512]; + (void)snprintf(foreground_db, sizeof(foreground_db), "%s/foreground.db", g_tmpdir); + (void)snprintf(background_db, sizeof(background_db), "%s/background.db", g_tmpdir); + + cbm_pipeline_t *foreground = cbm_pipeline_new(g_tmpdir, foreground_db, CBM_MODE_FULL); + cbm_pipeline_t *background = cbm_pipeline_new(g_tmpdir, background_db, CBM_MODE_FULL); + int foreground_rc = -1; + int background_rc = -1; + int foreground_nodes = -1; + int foreground_edges = -1; + int background_nodes = -1; + int background_edges = -1; + bool created = foreground && background; + if (created) { + cbm_pipeline_set_background(background, true); + foreground_rc = cbm_pipeline_run(foreground); + background_rc = cbm_pipeline_run(background); + cbm_pipeline_get_committed_counts(foreground, &foreground_nodes, &foreground_edges); + cbm_pipeline_get_committed_counts(background, &background_nodes, &background_edges); + } + + cbm_pipeline_free(foreground); + cbm_pipeline_free(background); + teardown_test_repo(); + saved_workers ? cbm_setenv("CBM_WORKERS", saved_workers, 1) : cbm_unsetenv("CBM_WORKERS"); + saved_single ? cbm_setenv("CBM_INDEX_SINGLE_THREAD", saved_single, 1) + : cbm_unsetenv("CBM_INDEX_SINGLE_THREAD"); + free(saved_workers); + free(saved_single); + + ASSERT_TRUE(created); + ASSERT_EQ(foreground_rc, 0); + ASSERT_EQ(background_rc, 0); + ASSERT_GT(foreground_nodes, 0); + ASSERT_EQ(background_nodes, foreground_nodes); + ASSERT_EQ(background_edges, foreground_edges); + PASS(); +} + TEST(pipeline_null_repo) { cbm_pipeline_t *p = cbm_pipeline_new(NULL, NULL, CBM_MODE_FULL); ASSERT_NULL(p); @@ -8944,8 +9208,8 @@ TEST(pipeline_semantic_manifest_rejects_non_directory_root) { cbm_file_hash_t *manifest = NULL; int manifest_count = -1; - int rc = cbm_pipeline_build_semantic_manifest("manifest-fail-closed", root_path, NULL, 0, NULL, - 0, NULL, NULL, &manifest, &manifest_count); + int rc = cbm_pipeline_build_semantic_manifest(NULL, "manifest-fail-closed", root_path, NULL, 0, + NULL, 0, NULL, NULL, &manifest, &manifest_count); cbm_pipeline_free_semantic_manifest(manifest, manifest_count > 0 ? manifest_count : 0); th_rmtree(tmp); @@ -8973,7 +9237,7 @@ TEST(pipeline_semantic_manifest_skips_unreadable_subdirectory) { cbm_file_hash_t *manifest = NULL; int manifest_count = -1; - int rc = cbm_pipeline_build_semantic_manifest("manifest-skip-locked", tmp, NULL, 0, NULL, 0, + int rc = cbm_pipeline_build_semantic_manifest(NULL, "manifest-skip-locked", tmp, NULL, 0, NULL, 0, NULL, NULL, &manifest, &manifest_count); bool found_sibling_gitignore = false; @@ -28932,6 +29196,12 @@ SUITE(pipeline) { RUN_TEST(pipeline_lock_release_allows_contender); /* Lifecycle */ RUN_TEST(pipeline_create_free); + RUN_TEST(pipeline_background_worker_policy_preserves_overrides); + RUN_TEST(pipeline_background_worker_width_keeps_parallel_route); +#if defined(CBM_ENABLE_TEST_SEAMS) && CBM_ENABLE_TEST_SEAMS + RUN_TEST(pipeline_surface_respects_total_worker_budget); +#endif + RUN_TEST(pipeline_background_policy_preserves_index_results); RUN_TEST(pipeline_null_repo); RUN_TEST(pipeline_free_null); RUN_TEST(pipeline_cancel);