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