perf(pipeline): delta fixes from scale testing — global watermark, full preseed, in-place surfaces

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 <martin.vogel.tech@gmail.com>

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);

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++;

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);

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