feat(store): drain ready overlay compactions

Add a bounded store helper that repeatedly reuses compact_next for ready overlay generations. A named CBM_STORE_COMPACT_ALL_GENERATIONS constant documents the drain-all mode, while positive limits cap the number of generations compacted by one API call.

Cover no-ready drain behavior and a two-overlay canary that proves a one-generation limit leaves later ready overlays visible for a subsequent drain. This prepares the future background worker to reuse one store path instead of duplicating compaction loops.

Validation: make -f Makefile.cbm -j8 build/c/test-runner; CBM_ONLY_SUITE=store_nodes build/c/test-runner (105 passed); git diff --check; scripts/check-source-safety.sh src/store/store.c src/store/store.h tests/test_store_nodes.c; make -f Makefile.cbm -j8 cbm.
Signed-off-by: Andrew Hundt <ATHundt@gmail.com>

src/store/store.c

@@ -4191,6 +4191,41 @@ int cbm_store_compact_next_overlay_generation(cbm_store_t *s, const char *projec
     return CBM_STORE_OK;
 }
 
+int cbm_store_compact_ready_overlay_generations(cbm_store_t *s, const char *project,
+                                                int max_generations,
+                                                int *out_compacted) {
+    if (out_compacted) {
+        *out_compacted = 0;
+    }
+    if (!s || !s->db || !project || !project[0] || !out_compacted ||
+        max_generations < CBM_STORE_COMPACT_ALL_GENERATIONS) {
+        if (s) {
+            store_set_error(s, "compact_ready_overlay_generations: invalid argument");
+        }
+        return CBM_STORE_ERR;
+    }
+
+    int compacted = 0;
+    while (max_generations == CBM_STORE_COMPACT_ALL_GENERATIONS ||
+           compacted < max_generations) {
+        int64_t overlay_generation = 0;
+        int64_t index_generation = 0;
+        int rc = cbm_store_compact_next_overlay_generation(s, project, &overlay_generation,
+                                                           &index_generation);
+        if (rc == CBM_STORE_NOT_FOUND) {
+            break;
+        }
+        if (rc != CBM_STORE_OK) {
+            *out_compacted = compacted;
+            return rc;
+        }
+        compacted++;
+    }
+
+    *out_compacted = compacted;
+    return CBM_STORE_OK;
+}
+
 static int store_resolve_node_id(cbm_store_t *s, const char *project, const char *qn,
                                  int64_t *out_id) {
     const char *qns[1] = {qn};

src/store/store.h

@@ -45,6 +45,7 @@ typedef struct cbm_store cbm_store_t;
 #define CBM_STORE_OVERLAY_STATUS_COMPACTING "compacting"
 #define CBM_STORE_OVERLAY_STATUS_COMPACTED "compacted"
 #define CBM_STORE_OVERLAY_STATUS_FAILED "failed"
+#define CBM_STORE_COMPACT_ALL_GENERATIONS 0
 #define CBM_STORE_OVERLAY_TOMBSTONE_FILE "file"
 #define CBM_STORE_DERIVED_GENERATION_UNKNOWN 0
 #define CBM_STORE_DERIVED_KIND_DIRECT "direct"
@@ -703,6 +704,13 @@ int cbm_store_recover_overlay_compaction_claims(cbm_store_t *s, const char *proj
 int cbm_store_compact_next_overlay_generation(cbm_store_t *s, const char *project,
                                               int64_t *out_overlay_generation,
                                               int64_t *out_index_generation);
+/* Compact ready overlay generations by repeatedly calling compact_next.
+ * max_generations=CBM_STORE_COMPACT_ALL_GENERATIONS drains all currently ready
+ * work; positive values cap how many generations this API call compacts.
+ * Returns OK with out_compacted=0 when no work is ready. */
+int cbm_store_compact_ready_overlay_generations(cbm_store_t *s, const char *project,
+                                                int max_generations,
+                                                int *out_compacted);
 
 /* Publish one file's replacement facts into overlay storage. This does not
  * mutate canonical nodes/edges; active read paths decide later how to combine

tests/test_store_nodes.c

@@ -1448,6 +1448,12 @@ TEST(store_compact_next_overlay_generation_returns_not_found_without_ready_overl
     ASSERT_EQ(overlay_generation, 0);
     ASSERT_EQ(index_generation, 0);
 
+    int compacted = SENTINEL;
+    ASSERT_EQ(cbm_store_compact_ready_overlay_generations(
+                  s, "test", CBM_STORE_COMPACT_ALL_GENERATIONS, &compacted),
+              CBM_STORE_OK);
+    ASSERT_EQ(compacted, 0);
+
     cbm_store_close(s);
     PASS();
 }
@@ -2198,6 +2204,81 @@ TEST(store_compact_overlay_generation_promotes_delete_only_tombstone) {
     PASS();
 }
 
+TEST(store_compact_ready_overlay_generations_respects_batch_limit) {
+    enum { BASE_GENERATION = 1, COMPACT_ONE_GENERATION = 1 };
+    cbm_store_t *s = cbm_store_open_memory();
+    ASSERT_NOT_NULL(s);
+    ASSERT_EQ(cbm_store_upsert_project(s, "test", "/tmp/test"), CBM_STORE_OK);
+
+    int64_t generation = 0;
+    ASSERT_EQ(cbm_store_reserve_index_generation(s, "test", NULL, NULL, &generation),
+              CBM_STORE_OK);
+    ASSERT_EQ(generation, BASE_GENERATION);
+    ASSERT_EQ(store_publish_helper_file_delta(s, generation), CBM_STORE_OK);
+    ASSERT_EQ(store_publish_old_main_delta(s, generation), CBM_STORE_OK);
+    ASSERT_EQ(cbm_store_finish_index_generation(s, "test", generation,
+                                                CBM_STORE_INDEX_STATUS_COMPLETE),
+              CBM_STORE_OK);
+
+    int64_t first_overlay = 0;
+    ASSERT_EQ(cbm_store_reserve_overlay_generation(s, "test", BASE_GENERATION,
+                                                   &first_overlay),
+              CBM_STORE_OK);
+    cbm_store_file_delta_t delete_main = {.project = "test",
+                                          .rel_path = "main.go",
+                                          .generation = BASE_GENERATION,
+                                          .derived_view_name =
+                                              CBM_STORE_DERIVED_VIEW_NODES_FTS,
+                                          .derived_status =
+                                              CBM_STORE_DERIVED_STATUS_COMPLETE};
+    ASSERT_EQ(cbm_store_publish_overlay_file_delta(s, &delete_main, first_overlay),
+              CBM_STORE_OK);
+
+    int64_t second_overlay = 0;
+    ASSERT_EQ(cbm_store_reserve_overlay_generation(s, "test", BASE_GENERATION,
+                                                   &second_overlay),
+              CBM_STORE_OK);
+    cbm_store_file_delta_t delete_helper = {.project = "test",
+                                            .rel_path = "helper.go",
+                                            .generation = BASE_GENERATION,
+                                            .derived_view_name =
+                                                CBM_STORE_DERIVED_VIEW_NODES_FTS,
+                                            .derived_status =
+                                                CBM_STORE_DERIVED_STATUS_COMPLETE};
+    ASSERT_EQ(cbm_store_publish_overlay_file_delta(s, &delete_helper, second_overlay),
+              CBM_STORE_OK);
+
+    int compacted = -1;
+    ASSERT_EQ(cbm_store_compact_ready_overlay_generations(
+                  s, "test", COMPACT_ONE_GENERATION, &compacted),
+              CBM_STORE_OK);
+    ASSERT_EQ(compacted, 1);
+    ASSERT_EQ(store_count_overlay_generation_row(s, "test", first_overlay,
+                                                 BASE_GENERATION,
+                                                 CBM_STORE_OVERLAY_STATUS_READY),
+              0);
+    ASSERT_EQ(store_count_overlay_generation_row(s, "test", second_overlay,
+                                                 BASE_GENERATION,
+                                                 CBM_STORE_OVERLAY_STATUS_READY),
+              1);
+    ASSERT_EQ(store_node_qn_exists(s, "test", "test.main.Old"), 0);
+    ASSERT_EQ(store_node_qn_exists(s, "test", "test.helper.Helper"), 1);
+
+    compacted = -1;
+    ASSERT_EQ(cbm_store_compact_ready_overlay_generations(
+                  s, "test", CBM_STORE_COMPACT_ALL_GENERATIONS, &compacted),
+              CBM_STORE_OK);
+    ASSERT_EQ(compacted, 1);
+    ASSERT_EQ(store_count_overlay_generation_row(s, "test", second_overlay,
+                                                 BASE_GENERATION,
+                                                 CBM_STORE_OVERLAY_STATUS_READY),
+              0);
+    ASSERT_EQ(store_node_qn_exists(s, "test", "test.helper.Helper"), 0);
+
+    cbm_store_close(s);
+    PASS();
+}
+
 TEST(store_find_nodes_by_file_overlay_view_returns_latest_ready_rows) {
     enum { BASE_GENERATION = 1 };
     cbm_store_t *s = cbm_store_open_memory();
@@ -5461,6 +5542,7 @@ SUITE(store_nodes) {
     RUN_TEST(store_overlay_publish_prunes_superseded_file_rows_and_fts);
     RUN_TEST(store_compact_overlay_generation_promotes_metadata_and_cleans_overlay);
     RUN_TEST(store_compact_overlay_generation_promotes_delete_only_tombstone);
+    RUN_TEST(store_compact_ready_overlay_generations_respects_batch_limit);
     RUN_TEST(store_find_nodes_by_file_overlay_view_returns_latest_ready_rows);
     RUN_TEST(store_search_overlay_view_without_ready_overlay_matches_canonical_search);
     RUN_TEST(store_search_overlay_view_matches_full_rebuild_oracle);