diff --git a/src/controller/meson.build b/src/controller/meson.build index d9c7877b..ff16d9a7 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -88,6 +88,7 @@ controller_unit_test_src = [ 'testsuite/test_ha_redirect.c', 'testsuite/test_p1_integration.c', 'testsuite/test_repl_unit.c', + 'testsuite/test_repl_apply.c', 'persist.c', 'ha.c', 'repl.c', diff --git a/src/controller/persist.c b/src/controller/persist.c index 6f4ae8cb..52ebf2c0 100644 --- a/src/controller/persist.c +++ b/src/controller/persist.c @@ -23,6 +23,14 @@ #include #include +/* Forward declarations for P0 restore stubs (defined in job.c / worker.c). + * Avoids pulling in the heavy cluster.h / worker.h header chains here. + * Full Cap'n Proto wiring happens in P2-4+. */ +extern int restore_job(cluster_t* c, uint64_t jobid, + const void* capnp_payload, size_t len); +extern int worker_from_capnp(cluster_t* c, + const void* capnp_payload, size_t len); + /* ------------------------------------------------------------------ */ /* Constants */ /* ------------------------------------------------------------------ */ @@ -527,14 +535,68 @@ static int load_snapshot(persist_t* p, const char* path, uint64_t lsn) return 0; } -/* P0 stub: no job/worker deserialization yet — wired in Task 12+/15. */ -static int apply_wal_entry(persist_t* p, const wal_entry_hdr_t* hdr, - const void* payload) +/* P2-2: Standby WAL replay — apply a single replicated entry to in-memory state. + * + * No hooks fire on standbys; the hook system is empty until promotion. + * + * Returns 0 on success (including skip for checkpoint / unknown types), + * negative errno on error (NULL args, oversized payload). */ +int apply_wal_entry(persist_t* p, const wal_entry_hdr_t* hdr, const void* payload) { - (void)p; - (void)hdr; - (void)payload; - /* Full implementation in Task 12+ (hook wiring) and Task 15 (cluster wiring) */ + if (!p || !hdr) + return -EINVAL; + + /* Reject oversized payloads BEFORE any allocation */ + if (hdr->payload_len > MAX_WAL_PAYLOAD_BYTES) { + fprintf(stderr, + "apply_wal_entry: payload_len %u exceeds MAX_WAL_PAYLOAD_BYTES %u\n", + hdr->payload_len, MAX_WAL_PAYLOAD_BYTES); + return -EMSGSIZE; + } + + /* WAL_CHECKPOINT (100) is a local-only marker — standbys never receive it + * via replication. Skip silently; do NOT advance LSN. */ + if (hdr->type == WAL_CHECKPOINT) + return 0; + + /* Unknown types above WAL_ENTRY_TYPE_MAX: log WARN and skip for forward + * compatibility — an older controller can boot on WAL written by a newer + * one without hard-failing on unrecognised entry types. */ + if (hdr->type > WAL_ENTRY_TYPE_MAX) { + fprintf(stderr, + "WARN apply_wal_entry: unknown type %u > WAL_ENTRY_TYPE_MAX %u" + " (lsn=%" PRIu64 ") — skipping for forward compatibility\n", + hdr->type, WAL_ENTRY_TYPE_MAX, hdr->lsn); + return 0; + } + + /* Advance LSN chain: track the highest replicated LSN seen. */ + pthread_mutex_lock(&p->lock); + if (hdr->lsn + 1 > p->current_lsn) + p->current_lsn = hdr->lsn + 1; + p->last_crc = hdr->crc; + pthread_mutex_unlock(&p->lock); + + /* Entity-creating entries: delegate to existing restore stubs. + * Gate on p->cluster != NULL so unit tests without a full cluster work. */ + switch (hdr->type) { + case WAL_JOB_SUBMIT: + /* restore_job() is a P0 stub — full Cap'n Proto decode in P2-4+. */ + if (p->cluster) + restore_job(p->cluster, 0 /* jobid decoded from payload in P2-4 */, + payload, hdr->payload_len); + break; + case WAL_WORKER_CTRL: + /* worker_from_capnp() is a P0 stub — full decode in P2-4+. */ + if (p->cluster) + worker_from_capnp(p->cluster, payload, hdr->payload_len); + break; + default: + /* All other known types: LSN/CRC already advanced above. + * Full per-type handling wired in P2-4 (Cap'n Proto decode). */ + break; + } + return 0; } diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index 8aba204d..318e46e1 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -24,6 +24,7 @@ extern void test_ha_cluster(void); extern void test_ha_redirect(void); extern void test_p1_integration(void); extern void test_repl_unit(void); +extern void test_repl_apply(void); void setUp(void) { return; } @@ -117,6 +118,7 @@ int main(int argc, char* argv[]) TEST(test_ha_redirect); TEST(test_p1_integration); TEST(test_repl_unit); + TEST(test_repl_apply); return UNITY_END(); } diff --git a/src/controller/testsuite/test_repl_apply.c b/src/controller/testsuite/test_repl_apply.c new file mode 100644 index 00000000..69899d34 --- /dev/null +++ b/src/controller/testsuite/test_repl_apply.c @@ -0,0 +1,197 @@ +/* + * test_repl_apply.c — P2-2: Tests for apply_wal_entry() standby WAL replay. + * + * TDD RED phase: written BEFORE the implementation. + * These tests verify: + * - LSN advances past replayed entry + * - WAL_CHECKPOINT entries are skipped (rc=0, lsn unchanged) + * - Oversized payloads are rejected (rc != 0) + * - Unknown types (> WAL_ENTRY_TYPE_MAX) are skipped with rc=0 + */ +#include +#include +#include + +#include + +#include "config.h" +#include "persist.h" + +/* ------------------------------------------------------------------ */ +/* Weak stubs for P0 entity-restore functions (job.c / worker.c). */ +/* The unit test binary does not link those translation units — they */ +/* pull in the full cluster.h / netio.h chain. apply_wal_entry() only */ +/* calls these when p->cluster != NULL; tests use NULL cluster, so the */ +/* weak stubs are never reached. The linker needs the symbols to link. */ +/* ------------------------------------------------------------------ */ +typedef struct cluster_s cluster_t; + +__attribute__((weak)) +int restore_job(cluster_t* c, uint64_t jobid, + const void* capnp_payload, size_t len) +{ + (void)c; (void)jobid; (void)capnp_payload; (void)len; + return 0; +} + +__attribute__((weak)) +int worker_from_capnp(cluster_t* c, const void* capnp_payload, size_t len) +{ + (void)c; (void)capnp_payload; (void)len; + return 0; +} + +/* ------------------------------------------------------------------ */ +/* Test helpers */ +/* ------------------------------------------------------------------ */ + +static persist_t* make_test_persist_for_apply(void) +{ + persist_t* p = calloc(1, sizeof(persist_t)); + if (!p) { + fprintf(stderr, "calloc failed\n"); + exit(1); + } + persist_config_t cfg = persist_config_defaults(); + cfg.wal_sync_ms = 5000; /* slow fsync — we don't need it in unit tests */ + + char tmpdir[256]; + snprintf(tmpdir, sizeof(tmpdir), "/tmp/xce_apply_XXXXXX"); + if (!mkdtemp(tmpdir)) { + fprintf(stderr, "mkdtemp failed\n"); + free(p); + exit(1); + } + + persist_set_workdir(p, tmpdir, "ctrl-apply"); + persist_init(p, &cfg, NULL); /* NULL cluster — entity stubs gate on p->cluster */ + persist_start(p); + return p; +} + +static void destroy_test_persist(persist_t* p) +{ + persist_stop(p); + persist_fini(p); + free(p); +} + +/* ------------------------------------------------------------------ */ +/* Tests */ +/* ------------------------------------------------------------------ */ + +static void test_apply_wal_entry_advances_lsn(void) +{ + persist_t* p = make_test_persist_for_apply(); + + wal_entry_hdr_t hdr; + memset(&hdr, 0, sizeof(hdr)); + hdr.type = WAL_JOB_UPDATE; + hdr.lsn = 42; + hdr.term = 1; + hdr.payload_len = 4; + hdr.crc = 0xDEADBEEF; + hdr.prev_crc = WAL_CRC_NONE; + + uint8_t payload[4] = {1, 2, 3, 4}; + + int rc = apply_wal_entry(p, &hdr, payload); + TEST_ASSERT_EQUAL_INT(0, rc); + + /* current_lsn must have advanced to at least lsn+1 */ + TEST_ASSERT_TRUE(p->current_lsn > 42); + + /* last_crc must reflect the entry's crc */ + TEST_ASSERT_EQUAL_UINT32(0xDEADBEEF, p->last_crc); + + destroy_test_persist(p); + printf("test_apply_wal_entry_advances_lsn: PASS\n"); +} + +static void test_apply_wal_entry_skips_checkpoint(void) +{ + persist_t* p = make_test_persist_for_apply(); + uint64_t lsn_before = p->current_lsn; + + wal_entry_hdr_t hdr; + memset(&hdr, 0, sizeof(hdr)); + hdr.type = WAL_CHECKPOINT; + hdr.lsn = 100; + hdr.term = 1; + hdr.payload_len = 0; + + int rc = apply_wal_entry(p, &hdr, NULL); + TEST_ASSERT_EQUAL_INT(0, rc); /* skip = success, not error */ + + /* Checkpoint entries are local-only markers — standbys never receive them. + * current_lsn must NOT advance: skip means no-op. */ + TEST_ASSERT_EQUAL_UINT64(lsn_before, p->current_lsn); + + destroy_test_persist(p); + printf("test_apply_wal_entry_skips_checkpoint: PASS\n"); +} + +static void test_apply_wal_entry_rejects_oversized(void) +{ + persist_t* p = make_test_persist_for_apply(); + + wal_entry_hdr_t hdr; + memset(&hdr, 0, sizeof(hdr)); + hdr.type = WAL_JOB_SUBMIT; + hdr.lsn = 1; + hdr.term = 1; + hdr.payload_len = MAX_WAL_PAYLOAD_BYTES + 1; /* too large */ + + int rc = apply_wal_entry(p, &hdr, NULL); + TEST_ASSERT_NOT_EQUAL(0, rc); /* must be rejected */ + + destroy_test_persist(p); + printf("test_apply_wal_entry_rejects_oversized: PASS\n"); +} + +static void test_apply_wal_entry_skips_unknown_type(void) +{ + persist_t* p = make_test_persist_for_apply(); + uint64_t lsn_before = p->current_lsn; + + /* + * Type 200 is above WAL_ENTRY_TYPE_MAX (100). + * Spec: log WARN and skip with rc=0 (forward compatibility). + */ + wal_entry_hdr_t hdr; + memset(&hdr, 0, sizeof(hdr)); + hdr.type = 200; + hdr.lsn = 5; + hdr.term = 1; + hdr.payload_len = 0; + + int rc = apply_wal_entry(p, &hdr, NULL); + TEST_ASSERT_EQUAL_INT(0, rc); /* unknown type: skip not error */ + + /* LSN must NOT advance for a skipped-unknown type */ + TEST_ASSERT_EQUAL_UINT64(lsn_before, p->current_lsn); + + destroy_test_persist(p); + printf("test_apply_wal_entry_skips_unknown_type: PASS\n"); +} + +static void test_apply_wal_entry_null_guard(void) +{ + /* Both NULL inputs must return non-zero (EINVAL) */ + int rc = apply_wal_entry(NULL, NULL, NULL); + TEST_ASSERT_NOT_EQUAL(0, rc); + printf("test_apply_wal_entry_null_guard: PASS\n"); +} + +/* ------------------------------------------------------------------ */ +/* Suite entry point (registered in controller_test.c) */ +/* ------------------------------------------------------------------ */ + +void test_repl_apply(void) +{ + RUN_TEST(test_apply_wal_entry_advances_lsn); + RUN_TEST(test_apply_wal_entry_skips_checkpoint); + RUN_TEST(test_apply_wal_entry_rejects_oversized); + RUN_TEST(test_apply_wal_entry_skips_unknown_type); + RUN_TEST(test_apply_wal_entry_null_guard); +} diff --git a/src/include/persist.h b/src/include/persist.h index 1c1e8206..9285b9b0 100644 --- a/src/include/persist.h +++ b/src/include/persist.h @@ -167,6 +167,12 @@ int persist_queue_hook_fun(int type, int situation, int detail, int persist_worker_hook_fun(int type, int situation, int detail, void* cls, va_list ap); +/* Apply a single WAL entry to in-memory state (standby replay path). + * No hooks fire — hook system is empty on standbys until promotion. + * Returns 0 on success (including skip for checkpoint/unknown types), + * negative errno on error (oversized payload, NULL arguments). */ +int apply_wal_entry(persist_t* p, const wal_entry_hdr_t* hdr, const void* payload); + /* GC retention predicates (P0-5a / P0-8 / P0-13) */ #include #include "xce/types.h"