diff --git a/src/store/store.c b/src/store/store.c index 612122757..2974e7644 100644 --- a/src/store/store.c +++ b/src/store/store.c @@ -4076,6 +4076,56 @@ int cbm_store_claim_ready_overlay_generation(cbm_store_t *s, const char *project return CBM_STORE_OK; } +int cbm_store_recover_compacting_overlay_generations(cbm_store_t *s, const char *project, + int *out_recovered) { + if (out_recovered) { + *out_recovered = 0; + } + if (!s || !s->db || !project || !project[0] || !out_recovered) { + if (s) { + store_set_error(s, "recover_compacting_overlay_generations: invalid argument"); + } + return CBM_STORE_ERR; + } + + int rc = cbm_store_begin(s); + if (rc != CBM_STORE_OK) { + return rc; + } + + sqlite3_stmt *stmt = NULL; + static const char sql[] = + "UPDATE overlay_generations SET status = ?3, updated_at = ?4 " + "WHERE project = ?1 AND status = ?2;"; + if (sqlite3_prepare_v2(s->db, sql, CBM_NOT_FOUND, &stmt, NULL) != SQLITE_OK) { + store_set_error_sqlite(s, "recover_compacting_overlay_generations prepare"); + (void)cbm_store_rollback(s); + return CBM_STORE_ERR; + } + char ts[CBM_SZ_64]; + iso_now(ts, sizeof(ts)); + bind_text(stmt, ST_COL_1, project); + bind_text(stmt, ST_COL_2, CBM_STORE_OVERLAY_STATUS_COMPACTING); + bind_text(stmt, ST_COL_3, CBM_STORE_OVERLAY_STATUS_READY); + bind_text(stmt, ST_COL_4, ts); + rc = sqlite3_step(stmt); + sqlite3_finalize(stmt); + if (rc != SQLITE_DONE) { + store_set_error_sqlite(s, "recover_compacting_overlay_generations"); + (void)cbm_store_rollback(s); + return CBM_STORE_ERR; + } + int recovered = sqlite3_changes(s->db); + + rc = cbm_store_commit(s); + if (rc != CBM_STORE_OK) { + (void)cbm_store_rollback(s); + return rc; + } + *out_recovered = recovered; + 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}; diff --git a/src/store/store.h b/src/store/store.h index 386d4d4a4..91058fedc 100644 --- a/src/store/store.h +++ b/src/store/store.h @@ -694,6 +694,10 @@ int cbm_store_count_overlay_generations(cbm_store_t *s, const char *project, int cbm_store_claim_ready_overlay_generation(cbm_store_t *s, const char *project, int64_t *out_overlay_generation, int64_t *out_base_generation); +/* Requeue compacting overlay generations after process restart or confirmed + * worker shutdown. Callers must not run this while a compactor is active. */ +int cbm_store_recover_compacting_overlay_generations(cbm_store_t *s, const char *project, + int *out_recovered); /* Publish one file's replacement facts into overlay storage. This does not * mutate canonical nodes/edges; active read paths decide later how to combine diff --git a/tests/test_store_nodes.c b/tests/test_store_nodes.c index a4a04af9c..2ebeb2724 100644 --- a/tests/test_store_nodes.c +++ b/tests/test_store_nodes.c @@ -1373,6 +1373,63 @@ TEST(store_claim_ready_overlay_generation_ignores_nonready_and_validates_outputs PASS(); } +TEST(store_recover_compacting_overlay_generations_requeues_abandoned_claims) { + enum { BASE_GENERATION = 5, SENTINEL = 42 }; + 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 compacting = 0; + int64_t failed = 0; + ASSERT_EQ(cbm_store_reserve_overlay_generation(s, "test", BASE_GENERATION, + &compacting), + CBM_STORE_OK); + ASSERT_EQ(cbm_store_reserve_overlay_generation(s, "test", BASE_GENERATION, &failed), + CBM_STORE_OK); + ASSERT_EQ(cbm_store_set_overlay_generation_status(s, "test", compacting, + CBM_STORE_OVERLAY_STATUS_READY), + CBM_STORE_OK); + ASSERT_EQ(cbm_store_set_overlay_generation_status(s, "test", failed, + CBM_STORE_OVERLAY_STATUS_FAILED), + CBM_STORE_OK); + + int64_t claimed = 0; + int64_t base = 0; + ASSERT_EQ(cbm_store_claim_ready_overlay_generation(s, "test", &claimed, &base), + CBM_STORE_OK); + ASSERT_EQ(claimed, compacting); + ASSERT_EQ(base, BASE_GENERATION); + + int recovered = SENTINEL; + ASSERT_EQ(cbm_store_recover_compacting_overlay_generations(s, "test", &recovered), + CBM_STORE_OK); + ASSERT_EQ(recovered, 1); + ASSERT_EQ(store_count_overlay_generation_row(s, "test", compacting, BASE_GENERATION, + CBM_STORE_OVERLAY_STATUS_READY), + 1); + ASSERT_EQ(store_count_overlay_generation_row(s, "test", failed, BASE_GENERATION, + CBM_STORE_OVERLAY_STATUS_FAILED), + 1); + + ASSERT_EQ(cbm_store_claim_ready_overlay_generation(s, "test", &claimed, &base), + CBM_STORE_OK); + ASSERT_EQ(claimed, compacting); + + recovered = SENTINEL; + ASSERT_EQ(cbm_store_recover_compacting_overlay_generations(s, "test", &recovered), + CBM_STORE_OK); + ASSERT_EQ(recovered, 1); + recovered = SENTINEL; + ASSERT_EQ(cbm_store_recover_compacting_overlay_generations(s, "missing", &recovered), + CBM_STORE_OK); + ASSERT_EQ(recovered, 0); + ASSERT_EQ(cbm_store_recover_compacting_overlay_generations(s, "test", NULL), + CBM_STORE_ERR); + + cbm_store_close(s); + PASS(); +} + TEST(store_overlay_file_delta_publish_rows_and_tombstone) { enum { BASE_GENERATION = 3 }; cbm_store_t *s = cbm_store_open_memory(); @@ -5345,6 +5402,7 @@ SUITE(store_nodes) { RUN_TEST(store_overlay_generation_rejects_invalid_inputs); RUN_TEST(store_claim_ready_overlay_generation_claims_oldest_once); RUN_TEST(store_claim_ready_overlay_generation_ignores_nonready_and_validates_outputs); + RUN_TEST(store_recover_compacting_overlay_generations_requeues_abandoned_claims); RUN_TEST(store_overlay_file_delta_publish_rows_and_tombstone); RUN_TEST(store_delete_project_clears_overlay_fts); RUN_TEST(store_overlay_file_delta_publish_rejects_invalid_delta_without_rows);