From 6ec9e944e535b56b0f7df3ccafff07a549ea1b10 Mon Sep 17 00:00:00 2001 From: Martin Vogel Date: Mon, 3 Aug 2026 21:31:46 +0200 Subject: [PATCH] =?UTF-8?q?perf(pipeline):=20delta=20fixes=20from=20scale?= =?UTF-8?q?=20testing=20=E2=80=94=20global=20watermark,=20full=20preseed,?= =?UTF-8?q?=20in-place=20surfaces?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three defects the kernel and django corpora exposed in the delta executor, each caught by its own fail-closed design and each fixed at the root: Global id watermark. MAX(id) was project-scoped while node ids are one keyspace for the whole database; a fresh node collided with a row outside the project filter (UNIQUE nodes.id on django). The watermark now clears every row. Full preseed. The label-filtered proxy set immediately met its counterexamples: synthetic Decorator nodes on django failed the patch via the QN constraint, and Macro -- six million of the kernel's 8.5M nodes -- was absent entirely, which would have silently dropped cross-file macro edges rather than failing. Curating an edge-endpoint-label list is guessing; every project node is now a proxy. The load stays edge-free and property-free, which is where the old full load actually spent its time (preseed measures 14.1s against the 38.3s gbuf load it replaces, plus that load's 16.5M edges). In-place surfaces. publish rewrote every lsp_surface row on each delta (delete-all plus re-upsert of ~89k serialized def sets); the patch now deletes exactly the purged files' rows and upserts the repaired files' fresh ones inside its own transaction, and publish skips the wholesale rewrite behind generation->surfaces_in_place. Measured 0.12s for the whole write block at kernel scale. publish_staged gains per-block timing logs; they located the next optimization targets precisely (23.8s in the meta/coverage section at kernel scale, integrity and seal effectively free). Measured end-to-end on the kernel corpus: one-file warm reindex 306s (binary routing) -> 223s (closure via dump) -> 121.7s (delta), peak RSS 33.6 -> 21.5 -> 13.6GB. django delta repair: 1.6s worker time. Convergence suite and 437 pipeline/incremental tests green throughout. Signed-off-by: Martin Vogel --- src/pipeline/pipeline.c | 17 +++++++- src/pipeline/pipeline_delta.c | 61 +++++++++++++++-------------- src/pipeline/pipeline_incremental.c | 47 ++++++++++++---------- src/pipeline/pipeline_internal.h | 4 ++ 4 files changed, 77 insertions(+), 52 deletions(-) diff --git a/src/pipeline/pipeline.c b/src/pipeline/pipeline.c index 8116c0012..814ca0784 100644 --- a/src/pipeline/pipeline.c +++ b/src/pipeline/pipeline.c @@ -1686,6 +1686,8 @@ int cbm_pipeline_publish_generation(const cbm_pipeline_generation_t *generation) * the nodes it created. */ int cbm_pipeline_publish_staged(char *stage_path, const cbm_pipeline_generation_t *generation, bool fts_wholesale) { + struct timespec t_pub; + cbm_clock_gettime(CLOCK_MONOTONIC, &t_pub); cbm_store_t *store = cbm_store_open_path(stage_path); if (!store) { discard_generation_stage(stage_path); @@ -1700,7 +1702,11 @@ int cbm_pipeline_publish_staged(char *stage_path, const cbm_pipeline_generation_ * store, before the atomic rename, so graph and surface data can never * publish separately. The delete guards the incremental path, whose * staging DB starts as a copy of the previous generation. */ - if (ok) { + /* surfaces_in_place: the delta patch already upserted exactly the + * repaired files' rows and deleted the purged ones inside its own + * transaction; rewriting every row here would be the single largest + * block of a delta publish at scale. */ + if (ok && !generation->surfaces_in_place) { ok = cbm_store_delete_lsp_surfaces(store, generation->project) == CBM_STORE_OK && cbm_store_upsert_lsp_surface_batch(store, generation->surface_rows, generation->surface_row_count) == CBM_STORE_OK; @@ -1709,6 +1715,9 @@ int cbm_pipeline_publish_staged(char *stage_path, const cbm_pipeline_generation_ ok = cbm_store_adr_store(store, generation->project, generation->adr_content) == CBM_STORE_OK; } + cbm_log_info("publish.timing", "block", "writes", "elapsed_ms", + itoa_buf((int)elapsed_ms(t_pub))); + cbm_clock_gettime(CLOCK_MONOTONIC, &t_pub); cbm_project_t project_info = {0}; bool have_project_info = @@ -1728,12 +1737,18 @@ int cbm_pipeline_publish_staged(char *stage_path, const cbm_pipeline_generation_ if (fts_wholesale && generation_rebuild_fts(store) != CBM_STORE_OK) { ok = false; } + cbm_log_info("publish.timing", "block", "fts", "elapsed_ms", itoa_buf((int)elapsed_ms(t_pub))); + cbm_clock_gettime(CLOCK_MONOTONIC, &t_pub); if (ok && !cbm_store_check_integrity(store)) { ok = false; } + cbm_log_info("publish.timing", "block", "integrity", "elapsed_ms", + itoa_buf((int)elapsed_ms(t_pub))); + cbm_clock_gettime(CLOCK_MONOTONIC, &t_pub); if (ok && cbm_store_seal_for_atomic_publish(store) != CBM_STORE_OK) { ok = false; } + cbm_log_info("publish.timing", "block", "seal", "elapsed_ms", itoa_buf((int)elapsed_ms(t_pub))); cbm_store_close(store); if (!ok) { discard_generation_stage(stage_path); diff --git a/src/pipeline/pipeline_delta.c b/src/pipeline/pipeline_delta.c index 6f8271f9b..74a92a0c4 100644 --- a/src/pipeline/pipeline_delta.c +++ b/src/pipeline/pipeline_delta.c @@ -219,17 +219,13 @@ int cbm_delta_purge(cbm_store_t *store, const char *project, const char *const * return 0; } -/* Labels whose nodes resolution or the restricted post-passes may reference - * as edge endpoints. Everything else stays on disk; an unexpected reference - * to an unseeded QN surfaces as a fresh node whose INSERT then violates the - * UNIQUE(project, qualified_name) constraint, failing the patch — and the - * orchestration's fallback to a full rebuild self-heals. Fail-closed, never - * silently wrong. */ -static const char *const DELTA_PRESEED_LABELS[] = { - "Function", "Method", "Class", "Struct", "Interface", "Enum", "Type", - "Trait", "Protocol", "Variable", "Field", "File", "Module", "Package", - "Folder", "Project", "EnvVar", "Dependency", "Route", -}; +/* Every project node becomes a proxy. The first cut filtered by label and + * the fail-closed patch immediately found the counterexamples (synthetic + * Decorator nodes on django; Macro is six of the kernel's 8.5M nodes) — any + * node the resolvers or restricted post-passes can target as an edge + * endpoint must carry its real id in RAM, and curating that set by label is + * guessing. The proxy load stays edge-free and property-free, which is + * where the old full load actually spent its time. */ int64_t cbm_delta_preseed(cbm_store_t *store, const char *project, cbm_gbuf_t *gbuf) { sqlite3 *db = cbm_store_get_db(store); @@ -239,36 +235,26 @@ int64_t cbm_delta_preseed(cbm_store_t *store, const char *project, cbm_gbuf_t *g int64_t max_id = 0; { sqlite3_stmt *stmt = NULL; - if (sqlite3_prepare_v2(db, "SELECT COALESCE(MAX(id), 0) FROM nodes WHERE project = ?1", - CBM_NOT_FOUND, &stmt, NULL) != SQLITE_OK) { + /* Global MAX, not project-scoped: node ids are one keyspace for the + * whole database, and the watermark must clear every row in it. */ + if (sqlite3_prepare_v2(db, "SELECT COALESCE(MAX(id), 0) FROM nodes", CBM_NOT_FOUND, &stmt, + NULL) != SQLITE_OK) { return -1; } - sqlite3_bind_text(stmt, 1, project, CBM_NOT_FOUND, SQLITE_TRANSIENT); if (sqlite3_step(stmt) == SQLITE_ROW) { max_id = sqlite3_column_int64(stmt, 0); } sqlite3_finalize(stmt); } - enum { PRESEED_LABEL_COUNT = sizeof(DELTA_PRESEED_LABELS) / sizeof(DELTA_PRESEED_LABELS[0]) }; - char ph[2 * PRESEED_LABEL_COUNT + 1]; - delta_placeholders(ph, PRESEED_LABEL_COUNT); - char sql[CBM_SZ_1K]; - int n = snprintf(sql, sizeof(sql), - "SELECT id, label, name, qualified_name, file_path FROM nodes" - " WHERE project = ?1 AND label IN (%s) ORDER BY id", - ph); - if (n < 0 || (size_t)n >= sizeof(sql)) { - return -1; - } sqlite3_stmt *stmt = NULL; - if (sqlite3_prepare_v2(db, sql, CBM_NOT_FOUND, &stmt, NULL) != SQLITE_OK) { + if (sqlite3_prepare_v2(db, + "SELECT id, label, name, qualified_name, file_path FROM nodes" + " WHERE project = ?1 ORDER BY id", + CBM_NOT_FOUND, &stmt, NULL) != SQLITE_OK) { return -1; } sqlite3_bind_text(stmt, 1, project, CBM_NOT_FOUND, SQLITE_TRANSIENT); - for (int i = 0; i < PRESEED_LABEL_COUNT; i++) { - sqlite3_bind_text(stmt, 2 + i, DELTA_PRESEED_LABELS[i], CBM_NOT_FOUND, SQLITE_STATIC); - } int step_rc; int64_t seeded = 0; while ((step_rc = sqlite3_step(stmt)) == SQLITE_ROW) { @@ -331,8 +317,23 @@ static void delta_patch_node(const cbm_gbuf_node_t *node, void *userdata) { CBM_NOT_FOUND, SQLITE_TRANSIENT); if (sqlite3_step(ctx->node_stmt) != SQLITE_DONE) { ctx->failed = true; + char id_buf[32]; + snprintf(id_buf, sizeof(id_buf), "%lld", (long long)node->id); cbm_log_error("delta.patch_node_failed", "qn", - node->qualified_name ? node->qualified_name : ""); + node->qualified_name ? node->qualified_name : "", "err", + sqlite3_errmsg(ctx->db)); + cbm_log_error("delta.patch_node_id", "id", id_buf); + sqlite3_stmt *who = NULL; + if (sqlite3_prepare_v2(ctx->db, "SELECT label, qualified_name FROM nodes WHERE id = ?1", + CBM_NOT_FOUND, &who, NULL) == SQLITE_OK) { + sqlite3_bind_int64(who, 1, node->id); + if (sqlite3_step(who) == SQLITE_ROW) { + cbm_log_error("delta.patch_id_owner", "label", + (const char *)sqlite3_column_text(who, 0), "qn", + (const char *)sqlite3_column_text(who, 1)); + } + sqlite3_finalize(who); + } return; } ctx->nodes++; diff --git a/src/pipeline/pipeline_incremental.c b/src/pipeline/pipeline_incremental.c index 36ba7b9b5..b9b58bba7 100644 --- a/src/pipeline/pipeline_incremental.c +++ b/src/pipeline/pipeline_incremental.c @@ -18,6 +18,7 @@ enum { INCR_RING_BUF = 4, INCR_RING_MASK = 3, INCR_TS_BUF = 24 }; #include "pipeline/artifact.h" #include "pipeline/lsp_surface.h" #include "pipeline/pass_lsp_cross.h" +#include "sqlite3.h" #include "yyjson/yyjson.h" #include "pipeline/pipeline_internal.h" #include "store/store.h" @@ -1907,8 +1908,6 @@ static int run_closure_delta(cbm_pipeline_t *p, const char *db_path, const char const char **purge_paths = NULL; cbm_file_hash_t *manifest = NULL; int manifest_count = 0; - cbm_lsp_surface_row_t *publish_rows = NULL; - int publish_row_count = 0; cbm_coverage_row_t *cov = NULL; int cov_n = 0; @@ -2140,25 +2139,31 @@ static int run_closure_delta(cbm_pipeline_t *p, const char *db_path, const char goto out; } - /* Surviving previous surfaces + this run's fresh ones. */ + /* Surfaces in place: delete the purged files' rows and upsert only + * the repaired files' fresh ones inside the staging store. The other + * ~all rows are already correct in the clone; rewriting them was the + * largest single block of a kernel-scale delta publish. */ { - int cap = plan->stored_count + cr.fresh_count; - publish_rows = - (cbm_lsp_surface_row_t *)calloc(cap ? (size_t)cap : 1, sizeof(*publish_rows)); - if (!publish_rows) { - goto out; - } - for (int i = 0; i < plan->stored_count; i++) { - cbm_lsp_surface_row_t *row = &plan->stored_rows[i]; - if (!row->rel_path || cbm_ht_get(stale_surface_set, row->rel_path)) { - continue; + sqlite3 *sdb = cbm_store_get_db(staging); + sqlite3_stmt *del = NULL; + bool surf_ok = sdb != NULL; + if (surf_ok && sqlite3_prepare_v2(sdb, + "DELETE FROM lsp_surface WHERE project = ?1" + " AND rel_path = ?2", + CBM_NOT_FOUND, &del, NULL) == SQLITE_OK) { + for (int i = 0; surf_ok && i < ci + deleted_count; i++) { + sqlite3_reset(del); + sqlite3_bind_text(del, 1, project, CBM_NOT_FOUND, SQLITE_TRANSIENT); + sqlite3_bind_text(del, 2, purge_paths[i], CBM_NOT_FOUND, SQLITE_TRANSIENT); + surf_ok = sqlite3_step(del) == SQLITE_DONE; } - publish_rows[publish_row_count++] = *row; - memset(row, 0, sizeof(*row)); + sqlite3_finalize(del); + } else { + surf_ok = false; } - for (int i = 0; i < cr.fresh_count; i++) { - publish_rows[publish_row_count++] = cr.fresh_rows[i]; - memset(&cr.fresh_rows[i], 0, sizeof(cr.fresh_rows[i])); + if (!surf_ok || cbm_store_upsert_lsp_surface_batch(staging, cr.fresh_rows, + cr.fresh_count) != CBM_STORE_OK) { + goto out; } } @@ -2190,8 +2195,9 @@ static int run_closure_delta(cbm_pipeline_t *p, const char *db_path, const char .coverage_version = CBM_SEMANTIC_INDEX_VERSION, .hash_records_complete = true, }, - .surface_rows = publish_rows, - .surface_row_count = publish_row_count, + .surface_rows = NULL, + .surface_row_count = 0, + .surfaces_in_place = true, }; cbm_store_close(staging); staging = NULL; @@ -2224,7 +2230,6 @@ out: free(cr.def_modules); cbm_arena_destroy(&cr.arena); } - cbm_store_free_lsp_surfaces(publish_rows, publish_row_count); cbm_pipeline_free_semantic_manifest(manifest, manifest_count); free(cov); free(purge_paths); diff --git a/src/pipeline/pipeline_internal.h b/src/pipeline/pipeline_internal.h index fdda368eb..ab0e57116 100644 --- a/src/pipeline/pipeline_internal.h +++ b/src/pipeline/pipeline_internal.h @@ -702,6 +702,10 @@ typedef struct { * belong to the same generation. */ const cbm_lsp_surface_row_t *surface_rows; int surface_row_count; + /* True when the caller already wrote this generation's surface rows + * into the staging store (delta patch); publish then skips the + * wholesale delete+rewrite. */ + bool surfaces_in_place; } cbm_pipeline_generation_t; /* Serialize and fully populate a sibling staging database, then atomically