fix(controller): stop a failed test from hanging the unit suite (bug-1429); mkstemp for upgrade temp file (bug-1861)

A failing TEST_ASSERT longjmps out and skips the fixture teardown, leaving
the commit engine's sweep thread and persist's fsync_worker asleep on a
static struct. The next test re-inits it underneath them, corrupting the
condvar, and a later teardown blocks in pthread_cond_destroy/join until
the 300s meson timeout. The failure line itself was lost because stdout
was fully buffered.

- controller_test: line-buffer stdout; tearDown() reaps what a failed test
  leaked via new XCE_TESTING registries (persist_test_reap_unstopped, then
  commit_test_reap_unfinalized_engines -- persist_stop before engine fini).
  Clean run reaps 0; an injected failure now fails in 2s instead of hanging.
- persist: flusher_flush_once reports only its own durable advance, so
  fsync_worker no longer re-fires the durable callback for an advance a
  concurrent persist_flush made (flaky "must not re-fire" test). Regression
  test test_flusher_pass_does_not_claim_concurrent_flush_advance.
- tests: jap observer is atomic and waits (bounded) for a continuation that
  may resolve on the flusher thread; pump catch-up tests wait for trailing
  replies instead of a single scan.
- config_repl: upgrade decode temp file via mkstemp instead of a
  pthread_self()-derived /tmp name (same value across processes, opened
  without O_EXCL -- collisions and a symlink target).

Verified: controller_unit_test 1222/1222; 24 parallel runs (6x4), no
hangs or ASAN errors.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

src/controller/testsuite/controller_test.c

@@ -5,6 +5,9 @@
 
 #include <unity/unity.h>
 
+#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);

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

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

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