fix(persist): interruptible fsync_worker sleep so persist_stop joins promptly

fsync_worker slept the full wal_sync_ms in one uninterruptible nanosleep()
before re-checking fsync_stop. persist_stop sets the flag and immediately
pthread_join()s, so the join blocked until the in-flight sleep returned —
up to a full wal_sync_ms per teardown. Tests set wal_sync_ms=5000 ("slow
fsync — no races"), so each persist_stop cost ~5s; across the ~7 suites
that use it the cumulative teardown time blew past the 300s controller-test
meson cap, killing the suite with SIGTERM.

Replace the nanosleep with pthread_cond_timedwait on a new fsync_mutex/
fsync_cond (added to persist_t, init in persist_init, destroyed in
persist_fini). persist_stop now sets fsync_stop and signals fsync_cond
under fsync_mutex before joining, so the worker wakes immediately. fdatasync
still runs under p->lock, released before re-acquiring fsync_mutex — no lock
nesting. The "commit: WAIT deadline expired" log lines were unrelated
intentional negative tests (50-300ms), not the cause.

Full controller suite now completes in ~10s (was >300s TIMEOUT); 0 failures.

Logged as bug-604.

src/controller/persist.c

@@ -174,19 +174,37 @@ static void* fsync_worker(void* arg)
 {
     persist_t* p = arg;
 
+    /* Sleep the configured interval via a timed condvar wait rather than an
+     * uninterruptible nanosleep, so persist_stop wakes us immediately instead
+     * of blocking its pthread_join for up to a full wal_sync_ms (5s in tests
+     * that set a slow interval — was the controller-suite timeout cause). */
+    pthread_mutex_lock(&p->fsync_mutex);
     while (!atomic_load_explicit(&p->fsync_stop, memory_order_acquire)) {
-        long ms = (long)p->config.wal_sync_ms;
-        struct timespec ts = {
-            .tv_sec  = ms / 1000,
-            .tv_nsec = (ms % 1000) * 1000000L,
-        };
-        nanosleep(&ts, NULL);
+        long            ms = (long)p->config.wal_sync_ms;
+        struct timespec dl;
 
+        clock_gettime(CLOCK_REALTIME, &dl);
+        dl.tv_sec += ms / 1000;
+        dl.tv_nsec += (ms % 1000) * 1000000L;
+        if (dl.tv_nsec >= 1000000000L) {
+            dl.tv_sec += 1;
+            dl.tv_nsec -= 1000000000L;
+        }
+        pthread_cond_timedwait(&p->fsync_cond, &p->fsync_mutex, &dl);
+
+        /* Stop requested during the wait: exit before the next fdatasync. */
+        if (atomic_load_explicit(&p->fsync_stop, memory_order_acquire))
+            break;
+
+        /* fdatasync under p->lock — never held together with fsync_mutex. */
+        pthread_mutex_unlock(&p->fsync_mutex);
         pthread_mutex_lock(&p->lock);
         if (p->wal_fd >= 0)
             fdatasync(p->wal_fd);
         pthread_mutex_unlock(&p->lock);
+        pthread_mutex_lock(&p->fsync_mutex);
     }
+    pthread_mutex_unlock(&p->fsync_mutex);
     return NULL;
 }
 
@@ -253,6 +271,8 @@ int persist_init(persist_t* p, persist_config_t* cfg, cluster_t* c)
     pthread_mutex_init(&p->lock, NULL);
     pthread_mutex_init(&p->checkpoint_mutex, NULL);
     pthread_cond_init(&p->checkpoint_done, NULL);
+    pthread_mutex_init(&p->fsync_mutex, NULL);
+    pthread_cond_init(&p->fsync_cond, NULL);
     repl_init(&p->repl);   /* P2-4: fan-out replicator lifecycle */
 
     WARN("WAL/snapshot stored unencrypted in %s — "
@@ -285,7 +305,12 @@ int persist_start(persist_t* p)
 void persist_stop(persist_t* p)
 {
     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. */
+        pthread_mutex_lock(&p->fsync_mutex);
         atomic_store_explicit(&p->fsync_stop, 1, memory_order_release);
+        pthread_cond_signal(&p->fsync_cond);
+        pthread_mutex_unlock(&p->fsync_mutex);
         pthread_join(p->fsync_thread, NULL);
         atomic_store_explicit(&p->fsync_thread_running, 0, memory_order_release);
     }
@@ -305,6 +330,8 @@ void persist_fini(persist_t* p)
     pthread_mutex_destroy(&p->lock);
     pthread_mutex_destroy(&p->checkpoint_mutex);
     pthread_cond_destroy(&p->checkpoint_done);
+    pthread_mutex_destroy(&p->fsync_mutex);
+    pthread_cond_destroy(&p->fsync_cond);
 }
 
 /* Forward declaration — defined after checkpoint internals */