feat: add schedule_checkpoint() with WAL record-count trigger and idempotency CAS guard

src/controller/persist.c

@@ -208,6 +208,17 @@ int persist_init(persist_t* p, persist_config_t* cfg, cluster_t* c)
     p->last_replicated_crc = WAL_CRC_NONE;
     p->last_crc            = WAL_CRC_NONE;
 
+    /* Enforce minimum value for max_wal_records to prevent runaway checkpointing */
+    if (p->config.max_wal_records < 100) {
+        ERROR("persist: max_wal_records=%u too low (minimum 100) — setting to 100",
+              p->config.max_wal_records);
+        p->config.max_wal_records = 100;
+    }
+
+    /* Initialise atomic checkpoint-trigger counters */
+    atomic_store(&p->wal_record_count, 0);
+    atomic_store(&p->checkpoint_in_progress, 0);
+
     /* Create WAL and snapshot directories */
     if (mkdir_p(p->wal_path,  0700) != 0) return -errno;
     if (mkdir_p(p->snap_path, 0700) != 0) return -errno;
@@ -278,6 +289,9 @@ void persist_fini(persist_t* p)
     pthread_cond_destroy(&p->checkpoint_done);
 }
 
+/* Forward declaration — defined after checkpoint internals */
+static void schedule_checkpoint(persist_t* persist);
+
 int persist_wal_async(persist_t* p, uint32_t type,
                       const void* payload, uint32_t len)
 {
@@ -352,6 +366,16 @@ int persist_wal_async(persist_t* p, uint32_t type,
     }
 
     pthread_mutex_unlock(&p->lock);
+
+    /* WAL record-count checkpoint trigger: schedule an async checkpoint once
+     * the record count reaches the configured threshold.  The fetch_add is
+     * outside the lock — atomic_fetch_add is sufficient for the counter; the
+     * checkpoint itself is guarded by its own CAS inside schedule_checkpoint. */
+    if (atomic_fetch_add(&p->wal_record_count, 1) + 1 >=
+        p->config.max_wal_records) {
+        schedule_checkpoint(p);
+    }
+
     return 0;
 }
 
@@ -528,6 +552,57 @@ static void rotate_snapshots(persist_t* p)
         free(names[i]);
 }
 
+/* ------------------------------------------------------------------ */
+/* Async checkpoint scheduling (schedule_checkpoint + task fn)         */
+/* ------------------------------------------------------------------ */
+
+/* checkpoint_task_fn: executed by delaymgr worker thread.
+ *
+ * The CAS in schedule_checkpoint already holds the "in_progress" slot (=1).
+ * We must clear it before calling persist_checkpoint(), which has its own
+ * atomic_exchange guard.  After persist_checkpoint() returns (it resets the
+ * flag to 0), we also reset wal_record_count so the counter starts fresh. */
+static int checkpoint_task_fn(void* arg)
+{
+    persist_t* persist = (persist_t*)arg;
+
+    /* Release the "scheduled" slot so persist_checkpoint's guard can acquire */
+    atomic_store(&persist->checkpoint_in_progress, 0);
+
+    persist_checkpoint(persist);
+
+    /* Reset the record counter so the next checkpoint window starts clean.
+     * Done after persist_checkpoint returns — the flag is already 0 by now. */
+    atomic_store(&persist->wal_record_count, 0);
+    return 0;
+}
+
+/* schedule_checkpoint: idempotent async checkpoint request via delaymgr.
+ * Safe to call from any thread.  Uses atomic CAS to ensure at most one
+ * checkpoint is scheduled or running at a time.
+ *
+ * If cluster is NULL (unit tests without a live delaymgr) the function
+ * falls back to a synchronous checkpoint. */
+static void schedule_checkpoint(persist_t* persist)
+{
+    int expected = 0;
+    if (!atomic_compare_exchange_strong(&persist->checkpoint_in_progress,
+                                        &expected, 1))
+        return;   /* already scheduled or running — idempotency guard */
+
+    if (persist->cluster == NULL) {
+        /* Unit-test / no-cluster path: run synchronously and reset. */
+        atomic_store(&persist->checkpoint_in_progress, 0);
+        persist_checkpoint(persist);
+        atomic_store(&persist->wal_record_count, 0);
+        return;
+    }
+
+    /* Schedule via delaymgr with 1 s coalescing delay, single execution */
+    delay_task(&persist->cluster->delaymgr, 1, 1,
+               checkpoint_task_fn, persist);
+}
+
 /* ------------------------------------------------------------------ */
 /* Public: persist_checkpoint                                           */
 /* ------------------------------------------------------------------ */

src/controller/testsuite/test_persist_wal.c

@@ -126,6 +126,48 @@ static void test_wal_checkpoint_type_rejected(void)
     printf("test_wal_checkpoint_type_rejected: PASS\n");
 }
 
+/* ------------------------------------------------------------------ */
+/* Schedule-checkpoint and WAL record-count threshold tests            */
+/* ------------------------------------------------------------------ */
+
+/* test: schedule_checkpoint CAS guard is idempotent */
+static void test_schedule_checkpoint_cas_idempotent(void)
+{
+    /* Simulate what schedule_checkpoint does: CAS 0→1 twice.
+     * Second must fail (idempotency guard). */
+    atomic_int flag = 0;
+    int expected = 0;
+    int swapped = atomic_compare_exchange_strong(&flag, &expected, 1);
+    TEST_ASSERT_EQUAL_INT(1, swapped);   /* first CAS succeeds */
+    TEST_ASSERT_EQUAL_INT(1, atomic_load(&flag));
+
+    expected = 0;
+    swapped = atomic_compare_exchange_strong(&flag, &expected, 1);
+    TEST_ASSERT_EQUAL_INT(0, swapped);   /* second CAS fails — idempotent */
+    atomic_store(&flag, 0);              /* cleanup */
+}
+
+/* test: wal_record_count increments and triggers at threshold */
+static void test_wal_record_count_threshold(void)
+{
+    persist_config_t cfg = persist_config_defaults();
+    cfg.max_wal_records = 5;
+
+    atomic_uint count = 0;
+    for (int i = 0; i < 5; i++)
+        atomic_fetch_add(&count, 1);
+
+    TEST_ASSERT_TRUE(atomic_load(&count) >= cfg.max_wal_records);
+}
+
+/* test: persist_config_defaults has max_wal_records set */
+static void test_persist_config_defaults_max_wal_records(void)
+{
+    persist_config_t cfg = persist_config_defaults();
+    TEST_ASSERT_EQUAL_UINT(DEFAULT_MAX_WAL_RECORDS, cfg.max_wal_records);
+    TEST_ASSERT_TRUE(cfg.max_wal_records >= 100);
+}
+
 /* ------------------------------------------------------------------ */
 /* Suite entry point (called from controller_test.c)                   */
 /* ------------------------------------------------------------------ */
@@ -137,4 +179,7 @@ void test_persist_wal(void)
     RUN_TEST(test_persist_init_creates_directories);
     RUN_TEST(test_persist_wal_async_writes_entry);
     RUN_TEST(test_wal_checkpoint_type_rejected);
+    RUN_TEST(test_schedule_checkpoint_cas_idempotent);
+    RUN_TEST(test_wal_record_count_threshold);
+    RUN_TEST(test_persist_config_defaults_max_wal_records);
 }