fix(pipeline): quarantine the real destination and report why a publish stopped

main and this branch each grew their own atomic publication while the branch was
outstanding, and neither knew about the other:

  main    cbm_pipeline_run() copies the live database into an mkstemp staging
          file, points the whole run at that copy, and renames it over the
          destination at the end. That is what stops an INCREMENTAL run from
          mutating the live database in place.
  branch  cbm_pipeline_publish_generation() builds a generation in its own stage
          file, validates it (integrity check, FTS rebuild, seal), quarantines a
          corrupt destination to a fresh .corrupt name, then renames.

The rebase kept both, nested, so the branch's layer ran against main's staging
file rather than the real database. Two consequences, both silent:

1. The quarantine became a no-op. Its "existing destination" was a staging file
   we had just created, so a genuinely corrupt destination reached
   cbm_rename_replace and was overwritten -- the one copy of the bytes that
   would explain the corruption, destroyed by the recovery path.

2. Every publish failure collapsed to CBM_NOT_FOUND, because the outer wrapper
   returned a bare -1 for its own errors, a cancellation, and a failed persist
   alike. A caller could not tell "aborted, your data is intact" from "the
   persist failed", which is the only distinction that matters at that moment.

Rather than pick one implementation, put each concern at the layer that owns it.
The outer wrapper owns the real destination's lifecycle, so it now owns the
quarantine: prepare_publish_destination() calls the branch's existing
prepare_existing_generation_for_replace() when the destination could not be
copied, which moves it aside only when it is verifiably not a readable SQLite
database. A destination that is valid (the backup failed for some other reason)
is sealed and replaced as before, never renamed away, so a good database is
never mislabelled .corrupt. main's guard that refuses to drop sidecars holding
uncommitted pages is kept ahead of it. A failed rename rolls the quarantine
back, so a caller is never left with no database at all.

The wrapper also stops flattening. Everything it does happens before the
publishing rename, so an abort there is genuinely non-destructive and can say
so: cancellation reports CBM_PIPELINE_ABORT_PRESERVE_DB, a failed seal reports
CBM_PIPELINE_PERSIST_FAILED, and a status from the inner publish propagates
unchanged. Both MCP call sites test only `rc == 0` and are unaffected; the codes
stay in pipeline_internal.h beside the stages that raise them, and pipeline.h no
longer claims a -1 it does not return.

Quarantining is now also correctly refused at the inner layer, which was the
same confusion in the other direction. publish_generation's destination is the
staging file the wrapper created moments earlier, so when that file was not a
readable database it was being parked as
`<db>.stage.<random>.corrupt` -- debris named after a temp file, which nothing
collects and no one can interpret. It only surfaced once the status codes stopped
collapsing, because the assertions that count leftover stage artifacts sat behind
the return-code assertions that failed first. The inner caller now discards an
unreadable staging file instead, and only the wrapper, which owns the user's real
database, ever quarantines.

Two of main's own tests asserted the bare -1. Both are named
cancelled_*_reindex_preserves_committed_db and now assert
CBM_PIPELINE_ABORT_PRESERVE_DB -- the value whose name is their subject. That
tightens the assertion rather than relaxing it: -2 answers their question and
-1 did not.

Closes the last 10 pipeline failures from the rebase.
pipeline, incremental, store_nodes, mcp: 662 passed, 2 skipped.

Signed-off-by: Martin Vogel <martin.vogel.tech@gmail.com>

src/pipeline/pipeline.c

@@ -1541,8 +1541,17 @@ static int quarantine_existing_generation(const char *db_path,
     return CBM_PIPELINE_PERSIST_FAILED;
 }
 
+/* `quarantine_invalid` separates the two callers, which own very different
+ * destinations. The publishing wrapper passes true: its destination is the
+ * user's live database, and bytes that are not a readable database are the only
+ * evidence of what went wrong, so they are moved aside rather than overwritten.
+ * publish_generation passes false: its destination is a private staging file
+ * this process created moments ago, so an unreadable one is our own debris --
+ * parking that under a .corrupt name leaves a file in the database directory
+ * that nothing ever collects and that no one can interpret. */
 static int prepare_existing_generation_for_replace(const char *db_path,
-                                                   cbm_replacement_prepare_t *prepared) {
+                                                   cbm_replacement_prepare_t *prepared,
+                                                   bool quarantine_invalid) {
     if (!prepared) {
         return CBM_PIPELINE_PERSIST_FAILED;
     }
@@ -1554,6 +1563,10 @@ static int prepare_existing_generation_for_replace(const char *db_path,
         }
         int seal_rc = cbm_store_seal_existing_path_for_replace(db_path);
         if (seal_rc == CBM_STORE_NOT_FOUND) {
+            if (!quarantine_invalid) {
+                (void)cbm_unlink(db_path);
+                return cbm_remove_db_sidecars(db_path) == 0 ? 0 : CBM_PIPELINE_PERSIST_FAILED;
+            }
             return quarantine_existing_generation(db_path, prepared);
         }
         if (seal_rc != CBM_STORE_OK) {
@@ -1659,7 +1672,9 @@ int cbm_pipeline_publish_generation(const cbm_pipeline_generation_t *generation)
         return CBM_PIPELINE_ABORT_PRESERVE_DB;
     }
     cbm_replacement_prepare_t prepared = {0};
-    if (prepare_existing_generation_for_replace(generation->final_db_path, &prepared) != 0) {
+    /* The destination here is the caller's staging file, not the live database:
+     * the publishing wrapper owns quarantining that. */
+    if (prepare_existing_generation_for_replace(generation->final_db_path, &prepared, false) != 0) {
         discard_generation_stage(stage_path);
         free(stage_path);
         return CBM_PIPELINE_PERSIST_FAILED;
@@ -2169,27 +2184,47 @@ static bool db_sidecars_absent(const char *db_path) {
     return true;
 }
 
-static bool prepare_publish_destination(const char *final_path, bool final_existed,
-                                        bool backup_succeeded) {
+/* Ready the real destination to receive the staged generation. Returns 0, or a
+ * CBM_PIPELINE_* code the caller propagates.
+ *
+ * `prepared` records whether the previous destination was moved aside, so a
+ * failed rename can put it back. It is zeroed here and is only meaningful on a
+ * 0 return. */
+static int prepare_publish_destination(const char *final_path, bool final_existed,
+                                       bool backup_succeeded,
+                                       cbm_replacement_prepare_t *prepared) {
+    memset(prepared, 0, sizeof(*prepared));
     struct stat current_st;
     bool final_exists_now = stat(final_path, &current_st) == 0;
     if (final_exists_now != final_existed) {
-        return false;
+        /* The destination appeared or vanished while we were indexing. Someone
+         * else owns it now; leave whatever is there alone. */
+        return CBM_PIPELINE_ABORT_PRESERVE_DB;
     }
     if (!final_exists_now) {
         /* A crashed generation can leave sidecars without a main file. */
-        return cbm_remove_db_sidecars(final_path) == 0;
+        return cbm_remove_db_sidecars(final_path) == 0 ? 0 : CBM_PIPELINE_PERSIST_FAILED;
     }
     if (!backup_succeeded) {
-        bool safe_to_replace = db_sidecars_absent(final_path);
-        if (!safe_to_replace) {
+        /* Sidecars alongside an un-copyable destination may hold the only
+         * committed pages; refuse rather than drop them. */
+        if (!db_sidecars_absent(final_path)) {
             cbm_log_error("pipeline.err", "phase", "publish", "reason",
                           "backup_failed_sidecars_preserved", "path", final_path);
+            return CBM_PIPELINE_PERSIST_FAILED;
         }
-        return safe_to_replace;
+        /* The destination could not be copied. If it is not a readable SQLite
+         * database it is corrupt, and the publishing rename would destroy the
+         * only copy of those bytes -- so move it aside under a fresh .corrupt
+         * name first, never overwriting an earlier quarantine. A destination
+         * that IS valid (the backup failed for some other reason) is sealed and
+         * replaced as usual, never renamed away. */
+        return prepare_existing_generation_for_replace(final_path, prepared, true);
     }
     return cbm_store_prepare_path_for_replace(final_path) == CBM_STORE_OK &&
-           cbm_remove_db_sidecars(final_path) == 0;
+                   cbm_remove_db_sidecars(final_path) == 0
+               ? 0
+               : CBM_PIPELINE_PERSIST_FAILED;
 }
 
 static int seal_staging_db(const char *staging_path) {
@@ -2266,11 +2301,28 @@ int cbm_pipeline_run(cbm_pipeline_t *p) {
     free(p->db_path);
     p->db_path = configured_db_path;
 
-    if (rc != 0 || check_cancel(p) || seal_staging_db(staging_path) != 0) {
+    /* Report WHY the run stopped. Everything below happens before the final
+     * rename, so the live database is still the previous generation and an
+     * abort is genuinely non-destructive -- a caller that cannot tell an
+     * aborted run from a failed persist cannot tell whether its data survived.
+     * The staging file is discarded on every one of these paths. */
+    if (rc != 0) {
         cleanup_staging_db(staging_path);
         free(staging_path);
         free(final_path);
-        return CBM_NOT_FOUND;
+        return rc;
+    }
+    if (check_cancel(p)) {
+        cleanup_staging_db(staging_path);
+        free(staging_path);
+        free(final_path);
+        return CBM_PIPELINE_ABORT_PRESERVE_DB;
+    }
+    if (seal_staging_db(staging_path) != 0) {
+        cleanup_staging_db(staging_path);
+        free(staging_path);
+        free(final_path);
+        return CBM_PIPELINE_PERSIST_FAILED;
     }
 
     if (p->before_publish_hook) {
@@ -2280,7 +2332,7 @@ int cbm_pipeline_run(cbm_pipeline_t *p) {
         cleanup_staging_db(staging_path);
         free(staging_path);
         free(final_path);
-        return CBM_NOT_FOUND;
+        return CBM_PIPELINE_ABORT_PRESERVE_DB;
     }
 
     /* A test hook may inspect the DB through SQLite and re-enable WAL mode;
@@ -2289,17 +2341,30 @@ int cbm_pipeline_run(cbm_pipeline_t *p) {
         cleanup_staging_db(staging_path);
         free(staging_path);
         free(final_path);
-        return CBM_NOT_FOUND;
+        return CBM_PIPELINE_PERSIST_FAILED;
     }
 
-    if (!prepare_publish_destination(final_path, final_existed, backup_succeeded) ||
-        (p->rename_hook ? p->rename_hook(staging_path, final_path, p->rename_hook_ctx)
-                        : cbm_rename_replace(staging_path, final_path)) != 0) {
+    cbm_replacement_prepare_t prepared = {0};
+    int prepare_rc = prepare_publish_destination(final_path, final_existed, backup_succeeded,
+                                                 &prepared);
+    if (prepare_rc != 0) {
         cbm_log_error("pipeline.err", "phase", "publish", "path", final_path);
         cleanup_staging_db(staging_path);
         free(staging_path);
         free(final_path);
-        return CBM_NOT_FOUND;
+        return prepare_rc;
+    }
+    if ((p->rename_hook ? p->rename_hook(staging_path, final_path, p->rename_hook_ctx)
+                        : cbm_rename_replace(staging_path, final_path)) != 0) {
+        cbm_log_error("pipeline.err", "phase", "publish", "path", final_path);
+        /* Put a quarantined destination back: the publish did not happen, so
+         * leaving the previous generation parked under .corrupt would present
+         * the caller with no database at all. */
+        (void)rollback_quarantined_generation(final_path, &prepared);
+        cleanup_staging_db(staging_path);
+        free(staging_path);
+        free(final_path);
+        return CBM_PIPELINE_PERSIST_FAILED;
     }
 
     rc = export_after_publish(p, final_path, was_incremental);

src/pipeline/pipeline.h

@@ -56,8 +56,16 @@ void cbm_pipeline_set_persistence(cbm_pipeline_t *p, bool enabled);
 /* Free a pipeline and all its internal state. NULL-safe. */
 void cbm_pipeline_free(cbm_pipeline_t *p);
 
-/* Run the full indexing pipeline. Returns 0 on success, -1 on error.
- * Discovers files, extracts, resolves, and dumps to SQLite. */
+/* Run the full indexing pipeline. Discovers files, extracts, resolves, and
+ * dumps to SQLite. Returns 0 on success and non-zero on failure.
+ *
+ * Treating any non-zero as "the run failed" is always correct. Callers that
+ * need to know whether the PREVIOUS generation survived can distinguish the
+ * failures by value: the run publishes by renaming a fully validated staging
+ * database over the destination, so every abort before that rename leaves the
+ * existing database in place. Those codes (CBM_PIPELINE_ABORT_PRESERVE_DB and
+ * CBM_PIPELINE_PERSIST_FAILED) are defined in pipeline_internal.h alongside the
+ * stages that raise them. */
 int cbm_pipeline_run(cbm_pipeline_t *p);
 
 /* Request cancellation of a running pipeline (thread-safe). */

tests/test_pipeline.c

@@ -9731,7 +9731,10 @@ TEST(cancelled_full_reindex_preserves_committed_db) {
     ASSERT_TRUE(hook.staging_existed);
     ASSERT_TRUE(hook.staging_was_valid);
     ASSERT_GT(hook.staged_candidates, 0); /* anti-vacuous: new full output was staged */
-    ASSERT_EQ(rc, -1);
+    /* Cancelling at the publish boundary aborts before the rename, so the
+     * committed database is untouched -- which is exactly what this code says.
+     * It was a bare -1 while every publish failure collapsed to one value. */
+    ASSERT_EQ(rc, CBM_PIPELINE_ABORT_PRESERVE_DB);
     ASSERT_TRUE(live_valid);
     ASSERT_EQ(nodes_after, baseline_nodes);
     ASSERT_EQ(helper_after, baseline_helper);
@@ -9796,7 +9799,9 @@ TEST(cancelled_incremental_reindex_preserves_committed_db) {
     ASSERT_TRUE(hook.staging_existed);
     ASSERT_TRUE(hook.staging_was_valid);
     ASSERT_GT(hook.staged_candidates, 0); /* anti-vacuous: changed output was staged */
-    ASSERT_EQ(rc, -1);
+    /* As above: an incremental run cancelled at the publish boundary preserves
+     * the committed database, and now reports that specifically. */
+    ASSERT_EQ(rc, CBM_PIPELINE_ABORT_PRESERVE_DB);
     ASSERT_TRUE(live_valid);
     ASSERT_EQ(nodes_after, baseline_nodes);
     ASSERT_EQ(helper_after, baseline_helper);