diff --git a/src/controller/persist.c b/src/controller/persist.c index 8909bb4e..7020d19a 100644 --- a/src/controller/persist.c +++ b/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 */ /* ------------------------------------------------------------------ */ diff --git a/src/controller/testsuite/test_persist_wal.c b/src/controller/testsuite/test_persist_wal.c index a235c44c..e07fc4e8 100644 --- a/src/controller/testsuite/test_persist_wal.c +++ b/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); }