Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
82 changes: 77 additions & 5 deletions src/pipeline/pipeline.c
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@ enum { CBM_DIR_PERMS = 0755, PL_RING = 4, PL_RING_MASK = 3, PL_SEQ_PASSES = 6, P
#include "foundation/compat_thread.h"
#include "foundation/profile.h"
#include "foundation/mem.h"
#include "yyjson/yyjson.h"

#include <stdint.h>
#include <stdio.h>
Expand Down Expand Up @@ -151,6 +152,64 @@ static const char *itoa_buf(int val) {
return bufs[i];
}

static const char *index_mode_name(cbm_index_mode_t mode) {
switch (mode) {
case CBM_MODE_FULL:
return "full";
case CBM_MODE_MODERATE:
return "moderate";
case CBM_MODE_FAST:
return "fast";
default:
return "unknown";
}
}

static bool parse_index_mode(const char *properties_json, cbm_index_mode_t *out_mode) {
if (!properties_json || !out_mode) {
return false;
}

yyjson_doc *doc = yyjson_read(properties_json, strlen(properties_json), 0);
if (!doc) {
return false;
}

yyjson_val *root = yyjson_doc_get_root(doc);
yyjson_val *mode_value = yyjson_is_obj(root) ? yyjson_obj_get(root, "index_mode") : NULL;
bool parsed = true;
if (yyjson_equals_str(mode_value, "full")) {
*out_mode = CBM_MODE_FULL;
} else if (yyjson_equals_str(mode_value, "moderate")) {
*out_mode = CBM_MODE_MODERATE;
} else if (yyjson_equals_str(mode_value, "fast")) {
*out_mode = CBM_MODE_FAST;
} else {
parsed = false;
}

yyjson_doc_free(doc);
return parsed;
}

static bool index_mode_covers(cbm_index_mode_t stored_mode, cbm_index_mode_t requested_mode) {
return stored_mode >= CBM_MODE_FULL && stored_mode <= CBM_MODE_FAST &&
requested_mode >= CBM_MODE_FULL && requested_mode <= CBM_MODE_FAST &&
stored_mode <= requested_mode;
}

static bool read_stored_index_mode(cbm_store_t *store, const char *project,
cbm_index_mode_t *out_mode) {
cbm_node_t project_node = {0};
if (cbm_store_find_node_by_qn(store, project, project, &project_node) != CBM_STORE_OK) {
return false;
}

bool parsed = parse_index_mode(project_node.properties_json, out_mode);
cbm_node_free_fields(&project_node);
return parsed;
}

/* Log current + peak RSS at a pipeline phase boundary (memory profiling). */
static void log_phase_mem(const char *phase) {
enum { PL_BYTES_PER_MB = 1024 * 1024 };
Expand Down Expand Up @@ -446,7 +505,11 @@ static int pass_structure(cbm_pipeline_t *p, const cbm_file_info_t *files, int f
cbm_log_info("pass.start", "pass", "structure", "files", itoa_buf(file_count));

/* Project node */
cbm_gbuf_upsert_node(p->gbuf, "Project", p->project_name, p->project_name, NULL, 0, 0, "{}");
char project_props[CBM_SZ_64];
snprintf(project_props, sizeof(project_props), "{\"index_mode\":\"%s\"}",
index_mode_name(p->mode));
cbm_gbuf_upsert_node(p->gbuf, "Project", p->project_name, p->project_name, NULL, 0, 0,
project_props);
const char *branch_qn = p->branch_qn ? p->branch_qn : p->project_name;
const char *branch_name = p->git_ctx.branch ? p->git_ctx.branch : "working-tree";
char branch_props[CBM_SZ_2K];
Expand Down Expand Up @@ -998,17 +1061,26 @@ static int try_incremental_or_delete_db(cbm_pipeline_t *p, cbm_file_info_t *file
if (check_store && cbm_store_check_integrity(check_store)) {
cbm_file_hash_t *hashes = NULL;
int hash_count = 0;
cbm_index_mode_t stored_mode = CBM_MODE_FAST;
bool stored_mode_known = read_stored_index_mode(check_store, p->project_name, &stored_mode);
cbm_store_get_file_hashes(check_store, p->project_name, &hashes, &hash_count);
cbm_store_free_file_hashes(hashes, hash_count);
cbm_store_close(check_store);
if (hash_count > 0 && file_count <= hash_count + (hash_count / PAIR_LEN)) {
bool mode_covered = stored_mode_known && index_mode_covers(stored_mode, p->mode);
if (hash_count > 0 && mode_covered && file_count <= hash_count + (hash_count / PAIR_LEN)) {
cbm_log_info("pipeline.route", "path", "incremental", "stored_hashes",
itoa_buf(hash_count));
int rc = cbm_pipeline_run_incremental(p, db_path, files, file_count);
itoa_buf(hash_count), "stored_mode", index_mode_name(stored_mode),
"requested_mode", index_mode_name(p->mode), "effective_mode",
index_mode_name(stored_mode));
int rc = cbm_pipeline_run_incremental(p, db_path, files, file_count, stored_mode);
free(db_path);
return rc;
}
if (hash_count > 0) {
if (hash_count > 0 && !mode_covered) {
cbm_log_info("pipeline.route", "path", "mode_upgrade_reindex", "stored_mode",
stored_mode_known ? index_mode_name(stored_mode) : "unknown",
"requested_mode", index_mode_name(p->mode));
} else if (hash_count > 0) {
cbm_log_info("pipeline.route", "path", "mode_change_reindex", "stored_hashes",
itoa_buf(hash_count), "discovered", itoa_buf(file_count));
}
Expand Down
10 changes: 8 additions & 2 deletions src/pipeline/pipeline_incremental.c
Original file line number Diff line number Diff line change
Expand Up @@ -684,11 +684,12 @@ static void dump_and_persist(cbm_gbuf_t *gbuf, const char *db_path, const char *
/* ── Incremental pipeline entry point ────────────────────────────── */

int cbm_pipeline_run_incremental(cbm_pipeline_t *p, const char *db_path, cbm_file_info_t *files,
int file_count) {
int file_count, cbm_index_mode_t effective_mode) {
struct timespec t0;
cbm_clock_gettime(CLOCK_MONOTONIC, &t0);

const char *project = cbm_pipeline_project_name(p);
cbm_index_mode_t requested_mode = (cbm_index_mode_t)cbm_pipeline_get_mode(p);

/* Open existing disk DB */
cbm_store_t *store = cbm_store_open_path(db_path);
Expand Down Expand Up @@ -839,7 +840,7 @@ int cbm_pipeline_run_incremental(cbm_pipeline_t *p, const char *db_path, cbm_fil
.registry = registry,
.cancelled = cbm_pipeline_cancelled_ptr(p),
.pipeline = p, /* so passes can record per-file skips (Track B) */
.mode = cbm_pipeline_get_mode(p),
.mode = effective_mode,
.path_aliases = path_aliases,
.excluded_dirs = excluded_dirs,
.excluded_count = excluded_count,
Expand All @@ -854,7 +855,12 @@ int cbm_pipeline_run_incremental(cbm_pipeline_t *p, const char *db_path, cbm_fil
}
}

/* Discovery already used requested_mode. Re-extraction must preserve the
* stronger capabilities recorded in Project.index_mode, including full-only
* C/C++ Macro nodes. Restore the process-wide gate immediately afterwards. */
cbm_set_macro_extraction(effective_mode == CBM_MODE_FULL);
run_extract_resolve(&ctx, changed_files, ci);
cbm_set_macro_extraction(requested_mode == CBM_MODE_FULL);
cbm_pipeline_pass_k8s(&ctx, changed_files, ci);
run_postpasses(&ctx, changed_files, ci, project);

Expand Down
6 changes: 4 additions & 2 deletions src/pipeline/pipeline_internal.h
Original file line number Diff line number Diff line change
Expand Up @@ -607,9 +607,11 @@ int cbm_scan_project_env_urls_excluded(const char *root_path, cbm_env_binding_t

/* Run incremental re-index on an existing disk DB.
* Classifies files by mtime+size, deletes changed nodes, re-parses changed
* files, merges into disk DB. Returns 0 on success. */
* files with effective_mode capabilities, then merges into disk DB. The
* pipeline's requested mode still owns discovery/exclusions. Returns 0 on
* success. */
int cbm_pipeline_run_incremental(cbm_pipeline_t *p, const char *db_path, cbm_file_info_t *files,
int file_count);
int file_count, cbm_index_mode_t effective_mode);

/* Pipeline accessors for incremental use */
const char *cbm_pipeline_repo_path(const cbm_pipeline_t *p);
Expand Down
Loading
Loading