diff --git a/src/controller/commit.c b/src/controller/commit.c index 237ae59c..75bf781a 100644 --- a/src/controller/commit.c +++ b/src/controller/commit.c @@ -51,6 +51,9 @@ * * Requirements: 1.1, 1.2, 2.1, 2.2, 2.3, 4.3, 4.4, 5.1, 6.3, 7.1, 7.2, 7.3 */ +#ifndef _GNU_SOURCE +#define _GNU_SOURCE /* pthread_getattr_np (XCE_TESTING engine reaper) */ +#endif #include #include #include @@ -353,6 +356,93 @@ static void commit_on_durable_advance(void* ctx) commit_resolve_pending(ce); } +#ifdef XCE_TESTING +/* bug-1429: registry of initialized-but-not-finalized engines, see + * commit_test_reap_unfinalized_engines() in commit.h. */ +#define COMMIT_TEST_MAX_LIVE 64 +static pthread_mutex_t g_live_engines_lock = PTHREAD_MUTEX_INITIALIZER; +static commit_engine_t* g_live_engines[COMMIT_TEST_MAX_LIVE]; +static int g_n_live_engines; + +static void live_engine_add(commit_engine_t* ce) +{ + pthread_mutex_lock(&g_live_engines_lock); + if (g_n_live_engines < COMMIT_TEST_MAX_LIVE) { + g_live_engines[g_n_live_engines++] = ce; + } + pthread_mutex_unlock(&g_live_engines_lock); +} + +static void live_engine_del(commit_engine_t* ce) +{ + int i; + + pthread_mutex_lock(&g_live_engines_lock); + for (i = 0; i < g_n_live_engines; i++) { + if (g_live_engines[i] == ce) { + g_live_engines[i] = g_live_engines[--g_n_live_engines]; + break; + } + } + pthread_mutex_unlock(&g_live_engines_lock); +} + +static bool on_own_stack(const void* addr) +{ + pthread_attr_t attr; + void* base = NULL; + size_t size = 0; + bool in = false; + + if (pthread_getattr_np(pthread_self(), &attr) == 0) { + if (pthread_attr_getstack(&attr, &base, &size) == 0) { + in = (const char*)addr >= (const char*)base + && (const char*)addr < (const char*)base + size; + } + pthread_attr_destroy(&attr); + } + return in; +} + +int commit_test_reap_unfinalized_engines(void) +{ + commit_engine_t* live[COMMIT_TEST_MAX_LIVE]; + int n, i, reaped = 0; + + pthread_mutex_lock(&g_live_engines_lock); + n = g_n_live_engines; + memcpy(live, g_live_engines, sizeof(live[0]) * (size_t)n); + g_n_live_engines = 0; + pthread_mutex_unlock(&g_live_engines_lock); + + for (i = 0; i < n; i++) { + commit_engine_t* ce = live[i]; + + if (on_own_stack(ce)) { + WARN("commit: test left a stack-local engine %p unfinalized " + "-- its sweep thread may outlive the frame", (void*)ce); + continue; + } + WARN("commit: reaping engine %p the test left unfinalized", + (void*)ce); + if (ce->sweep_thread != 0) { + pthread_mutex_lock(&ce->lock); + ce->sweep_stop = 1; + pthread_cond_broadcast(&ce->cond); + pthread_mutex_unlock(&ce->lock); + pthread_join(ce->sweep_thread, NULL); + ce->sweep_thread = 0; + } + if (ce->p != NULL) { + persist_set_durable_cb(ce->p, NULL, NULL); + persist_set_truncate_cb(ce->p, NULL, NULL); + } + reaped++; + } + return reaped; +} +#endif + /* ============================================================ * Lifecycle * ============================================================ */ @@ -425,6 +515,9 @@ int commit_engine_init(commit_engine_t* ce, persist_t* p, int n_peers, int n_vcp return -prc; } } +#ifdef XCE_TESTING + live_engine_add(ce); +#endif return 0; } @@ -447,6 +540,9 @@ void commit_engine_fini(commit_engine_t* ce) if (ce == NULL) { return; } +#ifdef XCE_TESTING + live_engine_del(ce); +#endif /* raft-apply-decoupling task 1.3: stop + join the timeout-sweep thread * BEFORE tearing down lock/cond/continuations -- it reads all three. * Signal under the lock and broadcast to wake it from either diff --git a/src/controller/config_repl.c b/src/controller/config_repl.c index 0c7834d5..a9a646dd 100644 --- a/src/controller/config_repl.c +++ b/src/controller/config_repl.c @@ -1723,11 +1723,13 @@ static int do_upgrade_config_with_bytes(cluster_t *c, endpoint_t *ep, /* 2. Write to a temp file and decode via load_cluster_config_from_file(). * This reuses the same decode path as the normal config load and avoids * direct in-memory capnp buffer manipulation. */ - char tmp_path[4096]; - snprintf(tmp_path, sizeof(tmp_path), "/tmp/xce_upgrade_cfg.%lu.tmp", - (unsigned long)pthread_self()); + /* mkstemp, not a pthread_self()-derived name: that is the same value in + * every process (so two controllers/test runs on one host clobbered each + * other's file) and, opened without O_EXCL in world-writable /tmp, a + * symlink target. mkstemp creates it exclusively, mode 0600. */ + char tmp_path[] = "/tmp/xce_upgrade_cfg.XXXXXX"; - int fd = open(tmp_path, O_WRONLY | O_CREAT | O_TRUNC, 0600); + int fd = mkstemp(tmp_path); if (fd < 0) { send_upgrade_error(ep, req, c, -EINVAL, "Failed to create temp decode file"); return -errno; diff --git a/src/controller/include/commit.h b/src/controller/include/commit.h index 9eefbacc..c642af2b 100644 --- a/src/controller/include/commit.h +++ b/src/controller/include/commit.h @@ -118,6 +118,21 @@ typedef struct commit_engine { int commit_engine_init(commit_engine_t* ce, persist_t* p, int n_peers, int n_vcpus); void commit_engine_fini(commit_engine_t* ce); +#ifdef XCE_TESTING +/* bug-1429: a failed TEST_ASSERT longjmps out of the test and skips the + * fixture's commit_engine_fini(), leaving the engine's sweep thread asleep + * on ce->cond. The next test re-inits the same static engine underneath it + * (corrupting the condvar), and a later teardown hangs until the suite's + * 300s timeout -- turning one visible FAIL into an opaque TIMEOUT. The test + * runner's tearDown() calls this after every test: it stops + joins the + * sweep thread of every engine initialized but never finalized, and + * unhooks its persist callbacks. lock/cond are left valid (a leaked + * flusher may still be mid-callback). Engines on the calling thread's + * stack are only dropped from the registry: their frame is already gone. + * Returns the number of engines reaped. */ +int commit_test_reap_unfinalized_engines(void); +#endif + /* raft-drive-async task 1.6: fold commit-wait admission onto the vtaskmgr * pool's single LivenessAdmission reservation. After this, commit_submit(WAIT) * acquires a pool blocking-admission slot (cap = n_vcpus - VTASK_LIVENESS_RESERVED) diff --git a/src/controller/persist.c b/src/controller/persist.c index 4119f65d..8cc2d27b 100644 --- a/src/controller/persist.c +++ b/src/controller/persist.c @@ -745,8 +745,9 @@ static bool flusher_work_pending_locked(persist_t* p) * for the whole snapshot+sync+advance so no concurrent append can extend * the segment between the tip read and the fdatasync. Returns the new * durable_lsn (unchanged if there was nothing to sync). */ -static uint64_t flusher_flush_once(persist_t* p) +static uint64_t flusher_flush_once(persist_t* p, bool* advanced) { + *advanced = false; #ifdef XCE_TESTING atomic_fetch_add_explicit(&p->test_flusher_flush_attempt_count, 1, memory_order_relaxed); @@ -800,8 +801,13 @@ static uint64_t flusher_flush_once(persist_t* p) * durable_lsn <= current_lsn - 1 is preserved because tip == current_lsn-1 * was read under this same lock). One fdatasync advances durable_lsn past * however many LSNs were appended since the last sync (batching). */ - if (tip > dur) + /* bug-1429: report the advance only if THIS pass made it, decided under + * p->lock -- a concurrent persist_flush() may already have moved + * durable_lsn, and that advance is its to report, not ours */ + if (tip > dur) { atomic_store_explicit(&p->durable_lsn, tip, memory_order_release); + *advanced = true; + } #ifdef XCE_TESTING /* Record which thread ran this successful fdatasync (task 5.1 / Req 1.6): @@ -856,9 +862,10 @@ static void* fsync_worker(void* arg) * landed during the flush, the level is still pending and we flush * again without needing its signal (lost-wakeup-safe). */ pthread_mutex_unlock(&p->fsync_mutex); - uint64_t before = atomic_load_explicit(&p->durable_lsn, - memory_order_acquire); - uint64_t after = flusher_flush_once(p); + bool advanced = false; + uint64_t before = atomic_load_explicit(&p->durable_lsn, + memory_order_acquire); + uint64_t after = flusher_flush_once(p, &advanced); /* COMMIT-EVAL WAKE (task 2.2 consumer wired): durable_lsn is advanced * INSIDE flusher_flush_once (atomic store with release) BEFORE we @@ -873,9 +880,11 @@ static void* fsync_worker(void* arg) * the advanced watermark on its own re-check, or (b) is blocked in * pthread_cond_wait and is woken by this broadcast; it can never both * miss the watermark AND miss the wake. We only fire the callback when - * durable_lsn actually advanced (after > before): a no-op flush (or a - * failed fdatasync that held the watermark) needs no commit re-eval, - * avoiding spurious wakes. + * THIS pass advanced durable_lsn: a no-op flush (or a failed fdatasync + * that held the watermark) needs no commit re-eval, and an advance a + * concurrent persist_flush() made is fired by persist_flush itself + * (bug-1429: a before/after snapshot here credited that advance to + * the flusher too and fired the callback twice). * * LOCK ORDERING (cycle-free): the flusher releases p->lock inside * flusher_flush_once, THEN the callback takes ce->lock (commit side). @@ -883,7 +892,7 @@ static void* fsync_worker(void* arg) * it never takes p->lock while holding ce->lock. So the only ordering * is p->lock → ce->lock; there is no ce->lock → p->lock edge, hence no * deadlock cycle. The callback explicitly MUST NOT take p->lock. */ - if (after > before && p->on_durable_advance != NULL) { + if (advanced && p->on_durable_advance != NULL) { p->on_durable_advance(p->durable_cb_ctx); } @@ -1286,6 +1295,80 @@ int persist_init(persist_t* p, persist_config_t* cfg, cluster_t* c) return 0; } +#ifdef XCE_TESTING +/* bug-1429: registry of started-but-not-stopped persist instances, see + * persist_test_reap_unstopped() in persist.h. */ +#define PERSIST_TEST_MAX_LIVE 64 +static pthread_mutex_t g_live_persist_lock = PTHREAD_MUTEX_INITIALIZER; +static persist_t* g_live_persist[PERSIST_TEST_MAX_LIVE]; +static int g_n_live_persist; + +static void live_persist_add(persist_t* p) +{ + pthread_mutex_lock(&g_live_persist_lock); + if (g_n_live_persist < PERSIST_TEST_MAX_LIVE) { + g_live_persist[g_n_live_persist++] = p; + } + pthread_mutex_unlock(&g_live_persist_lock); +} + +static void live_persist_del(persist_t* p) +{ + int i; + + pthread_mutex_lock(&g_live_persist_lock); + for (i = 0; i < g_n_live_persist; i++) { + if (g_live_persist[i] == p) { + g_live_persist[i] = g_live_persist[--g_n_live_persist]; + break; + } + } + pthread_mutex_unlock(&g_live_persist_lock); +} + +static bool persist_on_own_stack(const void* addr) +{ + pthread_attr_t attr; + void* base = NULL; + size_t size = 0; + bool in = false; + + if (pthread_getattr_np(pthread_self(), &attr) == 0) { + if (pthread_attr_getstack(&attr, &base, &size) == 0) { + in = (const char*)addr >= (const char*)base + && (const char*)addr < (const char*)base + size; + } + pthread_attr_destroy(&attr); + } + return in; +} + +int persist_test_reap_unstopped(void) +{ + persist_t* live[PERSIST_TEST_MAX_LIVE]; + int n, i, reaped = 0; + + pthread_mutex_lock(&g_live_persist_lock); + n = g_n_live_persist; + memcpy(live, g_live_persist, sizeof(live[0]) * (size_t)n); + g_n_live_persist = 0; + pthread_mutex_unlock(&g_live_persist_lock); + + for (i = 0; i < n; i++) { + if (persist_on_own_stack(live[i])) { + WARN("persist: test left a stack-local persist %p running " + "-- its flusher may outlive the frame", (void*)live[i]); + continue; + } + WARN("persist: stopping persist %p the test left running", + (void*)live[i]); + persist_stop(live[i]); + reaped++; + } + return reaped; +} +#endif + int persist_start(persist_t* p) { ctrl_metrics_init(); @@ -1302,13 +1385,20 @@ int persist_start(persist_t* p) atomic_store_explicit(&p->fsync_stop, 0, memory_order_release); int rc = pthread_create(&p->fsync_thread, NULL, fsync_worker, p); - if (rc == 0) + if (rc == 0) { atomic_store_explicit(&p->fsync_thread_running, 1, memory_order_release); +#ifdef XCE_TESTING + live_persist_add(p); +#endif + } return rc; } void persist_stop(persist_t* p) { +#ifdef XCE_TESTING + live_persist_del(p); +#endif if (atomic_load_explicit(&p->fsync_thread_running, memory_order_acquire)) { /* Set the stop flag and wake the timer wait under fsync_mutex so the * worker cannot miss the signal between its flag check and the wait. */ @@ -1997,6 +2087,11 @@ void persist_set_truncate_cb(persist_t* p, void (*fn)(void* ctx, uint64_t cutoff } #ifdef XCE_TESTING +uint64_t persist_test_flusher_flush_once(persist_t* p, bool* advanced) +{ + return flusher_flush_once(p, advanced); +} + void persist_test_set_force_fsync_fail(persist_t* p, int fail) { if (p) diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index 3427d377..19f6d2c0 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -5,6 +5,9 @@ #include +#include "commit.h" +#include "persist.h" + extern void test_persist_types(void); extern void test_persist_wal(void); extern void test_persist_schema(void); @@ -110,6 +113,12 @@ void setUp(void) { return; } void tearDown(void) { + /* bug-1429: runs even after a failed TEST_ASSERT -- stop the sweep + * thread of any commit engine whose fixture teardown was skipped, so one + * failure can't corrupt the next test's reused engine and hang the + * suite until its timeout */ + persist_test_reap_unstopped(); + commit_test_reap_unfinalized_engines(); printf("\n-----------------------------------------------------\n"); return; } @@ -178,6 +187,11 @@ int main(int argc, char* argv[]) { argp_parse(&argp, argc, argv, 0, 0, &arguments); + /* bug-1429: stdout is fully buffered when meson pipes it, so a FAIL + * line was lost whenever the run later hung (SIGTERM) or aborted + * (ASAN) -- line-buffer it so the first failure is always visible. */ + setvbuf(stdout, NULL, _IOLBF, 0); + UNITY_BEGIN(); TEST(test_persist_types); TEST(test_persist_wal); diff --git a/src/controller/testsuite/test_apply_pump_integration.c b/src/controller/testsuite/test_apply_pump_integration.c index cd201ed5..64fbd0ca 100644 --- a/src/controller/testsuite/test_apply_pump_integration.c +++ b/src/controller/testsuite/test_apply_pump_integration.c @@ -632,7 +632,8 @@ static void test_pump_command_and_fact_interleave_through_same_commit(void) tpi_wait_applied(watermark, 2000); tpi_reply_obs_t final_obs = { 0 }; - tpi_scan_replies(&g_tpi_req_ep, &final_obs); + /* bug-1429: replies can trail the walk (see the 3-command test) */ + tpi_wait_n_replies(&final_obs, 1, 2000); uint64_t committed_final = atomic_load(&g_tpi_c.persist.committed_lsn); job_t *jx_final = tpi_find_job(jobid_x); @@ -981,7 +982,11 @@ static void test_pump_mixed_facts_and_commands_catchup_original_order(void) tpi_wait_applied(watermark, 2000); tpi_reply_obs_t final_obs = { 0 }; - tpi_scan_replies(&g_tpi_req_ep, &final_obs); + /* bug-1429: the walk finishing does not mean every command's + * continuation has replied yet -- one can resolve on the + * fsync_worker thread after it; a single scan here flaked + * "Expected 3 Was 2" */ + tpi_wait_n_replies(&final_obs, 3, 2000); uint64_t committed_final = atomic_load(&g_tpi_c.persist.committed_lsn); job_t *ja_final = tpi_find_job(jobid_a); diff --git a/src/controller/testsuite/test_commit_barrier_rollout.c b/src/controller/testsuite/test_commit_barrier_rollout.c index 81af772d..c3cd0a0f 100644 --- a/src/controller/testsuite/test_commit_barrier_rollout.c +++ b/src/controller/testsuite/test_commit_barrier_rollout.c @@ -2032,9 +2032,9 @@ static int jap_read_wal_entry_at(persist_t *p, off_t at, /* Notify-callback capture (persist_job_commit_notify_fn). */ typedef struct { - int call_count; - uint64_t last_lsn; - int last_rc; + atomic_int call_count; + uint64_t last_lsn; + int last_rc; } jap_notify_obs_t; static jap_notify_obs_t g_jap_obs; @@ -2044,12 +2044,34 @@ static void jap_notify_reset(void) memset(&g_jap_obs, 0, sizeof(g_jap_obs)); } +/* bug-1429: the continuation may resolve on the fsync_worker thread (its + * durable-advance callback re-evaluates commit and can claim the + * continuation between our ACK being recorded and our own resolve), so + * the count is published with release after last_lsn/last_rc. */ static void jap_notify_fn(void *ctx, uint64_t lsn, int rc) { jap_notify_obs_t *obs = (jap_notify_obs_t *)ctx; - obs->call_count++; obs->last_lsn = lsn; obs->last_rc = rc; + atomic_fetch_add_explicit(&obs->call_count, 1, memory_order_release); +} + +/* Wait (bounded) for the continuation count to reach `want`: after the + * covering ACK, resolution is guaranteed but may be running on the + * flusher thread. Returns the count seen. */ +static int jap_wait_calls(int want) +{ + int i, n = 0; + + for (i = 0; i < 2000; i++) { /* <= 2s */ + n = atomic_load_explicit(&g_jap_obs.call_count, memory_order_acquire); + if (n >= want) + break; + usleep(1000); + } + /* let a (wrongly) duplicate resolution surface before asserting == */ + usleep(2000); + return atomic_load_explicit(&g_jap_obs.call_count, memory_order_acquire); } /* ── One WAL append, one commit continuation, decodable back to the same @@ -2138,7 +2160,7 @@ static void test_job_array_submit_appends_exactly_one_wal_entry_and_one_continua * resolves the single registered continuation */ TEST_ASSERT_EQUAL_INT(0, persist_flush(&g_jap)); repl_on_ack(&g_jap.repl, &g_jaep_a, out_lsn); - TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_jap_obs.call_count, + TEST_ASSERT_EQUAL_INT_MESSAGE(1, jap_wait_calls(1), "the notify callback must fire EXACTLY ONCE for the whole array, " "not once per element"); TEST_ASSERT_EQUAL_UINT64(out_lsn, g_jap_obs.last_lsn); @@ -2288,7 +2310,7 @@ static void test_job_array_submit_one_entry_one_continuation_across_sizes(void) "array_size=%llu: exactly ONE commit continuation must " "resolve for the whole array, not one per element", (unsigned long long)array_size); - TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_jap_obs.call_count, msg); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, jap_wait_calls(1), msg); TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_jap_obs.last_rc, msg); } @@ -2450,7 +2472,7 @@ static void test_job_submit_single_payload_wal_type_matches_pre_feature_baseline "the continuation must not resolve before a covering ACK"); TEST_ASSERT_EQUAL_INT(0, persist_flush(&g_jap)); repl_on_ack(&g_jap.repl, &g_jaep_a, lsns[0]); - TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_jap_obs.call_count, + TEST_ASSERT_EQUAL_INT_MESSAGE(1, jap_wait_calls(1), "exactly ONE commit continuation must resolve for the single " "payload, same contract as before this feature"); TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_jap_obs.last_rc, 0); diff --git a/src/controller/testsuite/test_commit_durability.c b/src/controller/testsuite/test_commit_durability.c index 659ccada..9dc52f7d 100644 --- a/src/controller/testsuite/test_commit_durability.c +++ b/src/controller/testsuite/test_commit_durability.c @@ -1056,6 +1056,34 @@ static void test_persist_flush_fires_durable_advance_callback(void) printf("test_persist_flush_fires_durable_advance_callback: PASS\n"); } +/* bug-1429: fsync_worker used to snapshot durable_lsn, flush, and fire the + * durable-advance callback if the watermark moved -- crediting itself with + * an advance a concurrent persist_flush() made in between, so the callback + * fired twice for one advance (the flaky "must not re-fire" failure above + * under load). A flusher pass must report only its own advance. */ +static void test_flusher_pass_does_not_claim_concurrent_flush_advance(void) +{ + char tmp[256]; + strcpy(tmp, "/tmp/xce_flushown_XXXXXX"); + TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); + + persist_t* p = make_durability_persist(tmp); + TEST_ASSERT_NOT_NULL(p); + + bool advanced = true; + + /* persist_flush() advances the watermark first ... */ + TEST_ASSERT_EQUAL_INT(0, persist_wal_async(p, WAL_JOB_UPDATE, "a", 1)); + TEST_ASSERT_EQUAL_INT(0, persist_flush(p)); + /* ... so the flusher's pass right after has nothing of its own */ + persist_test_flusher_flush_once(p, &advanced); + TEST_ASSERT_FALSE_MESSAGE(advanced, + "a flusher pass must not report an advance persist_flush() made"); + + teardown_durability_persist(p); + printf("test_flusher_pass_does_not_claim_concurrent_flush_advance: PASS\n"); +} + /* ------------------------------------------------------------------ * * Test 5.1b (Req 1.5 / 2.5): losing the sole durable contributor * * leaves the acknowledged op present on a surviving member. * @@ -2599,6 +2627,7 @@ void test_commit_durability(void) /* task 5.1 — durable-majority commit wired end to end */ RUN_TEST(test_durable_majority_commit_end_to_end); RUN_TEST(test_persist_flush_fires_durable_advance_callback); + RUN_TEST(test_flusher_pass_does_not_claim_concurrent_flush_advance); RUN_TEST(test_sole_durable_contributor_loss_survives); RUN_TEST(test_durability_work_off_iomgr_preserves_liveness); RUN_TEST(test_recover_verdict_clean_intact_log); diff --git a/src/include/persist.h b/src/include/persist.h index a93bb236..98a3220f 100644 --- a/src/include/persist.h +++ b/src/include/persist.h @@ -1022,6 +1022,21 @@ uint64_t persist_test_durable_lsn_on_disk(persist_t* p); * Compiled out of production (xce_controller links without it). */ void persist_test_set_force_fsync_fail(persist_t* p, int fail); +/* bug-1429: run one flusher pass (the fsync_worker's flush step) and report + * whether THAT pass advanced durable_lsn -- the worker fires the + * durable-advance callback only on its own advance, never on one a + * concurrent persist_flush() made. Returns the durable watermark. */ +uint64_t persist_test_flusher_flush_once(persist_t* p, bool* advanced); + +/* bug-1429: persist_stop() every instance persist_start()ed but never + * stopped -- a failed TEST_ASSERT skips the fixture teardown, and the next + * test re-inits the same static persist under its live fsync_worker + * (corrupting fsync_cond; persist_fini then hangs). Called from the test + * runner's tearDown() before commit_test_reap_unfinalized_engines(), the + * documented persist_stop-before-commit_engine_fini order. Stack-local + * instances are skipped. Returns the number stopped. */ +int persist_test_reap_unstopped(void); + /* Read the standby-path fdatasync counter (task 2.3, persist-before-ACK * regression). Counts the inline fdatasync calls made by * persist_wal_write_replicated. A test snapshots this before the call and