/* * 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 #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, uint64_t entry_lsn) { (void)c; (void)jobid; (void)capnp_payload; (void)len; (void)entry_lsn; 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; } __attribute__((weak)) int restore_job_state(cluster_t* c, uint64_t jobid, uint32_t state) { (void)c; (void)jobid; (void)state; return 0; } __attribute__((weak)) int restore_queue_state(cluster_t* c, const char* name, uint32_t status) { (void)c; (void)name; (void)status; 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; } /* Like make_test_persist_for_apply but reports the chosen tmpdir so a test * can re-open the same WAL via a FRESH persist_t (restart/recover proof). */ static persist_t* make_test_persist_in(char* tmpdir_out, size_t cap) { 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; snprintf(tmpdir_out, cap, "/tmp/xce_replwr_XXXXXX"); if (!mkdtemp(tmpdir_out)) { fprintf(stderr, "mkdtemp failed\n"); free(p); exit(1); } persist_set_workdir(p, tmpdir_out, "ctrl-replwr"); persist_init(p, &cfg, NULL); persist_start(p); return p; } /* Re-open an existing workdir as a fresh persist_t (simulates a restart). */ static persist_t* reopen_test_persist(const char* tmpdir) { 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; persist_set_workdir(p, tmpdir, "ctrl-replwr"); persist_init(p, &cfg, NULL); persist_start(p); return p; } /* Count *.wal files under /ctrl-replwr/data/wal */ static int count_replwr_wal_files(const char* tmpdir) { char wal_path[512]; snprintf(wal_path, sizeof(wal_path), "%s/ctrl-replwr/data/wal", tmpdir); DIR* dir = opendir(wal_path); if (!dir) return 0; int count = 0; struct dirent* ent; while ((ent = readdir(dir)) != NULL) { size_t len = strlen(ent->d_name); if (len > 4 && strcmp(ent->d_name + len - 4, ".wal") == 0) count++; } closedir(dir); return count; } /* Total bytes across all *.wal segment files (proxy for "entry on disk"). */ static off_t total_replwr_wal_bytes(const char* tmpdir) { char wal_path[512]; snprintf(wal_path, sizeof(wal_path), "%s/ctrl-replwr/data/wal", tmpdir); DIR* dir = opendir(wal_path); if (!dir) return 0; off_t total = 0; struct dirent* ent; while ((ent = readdir(dir)) != NULL) { size_t len = strlen(ent->d_name); if (len <= 4 || strcmp(ent->d_name + len - 4, ".wal") != 0) continue; char full[1024]; snprintf(full, sizeof(full), "%s/%s", wal_path, ent->d_name); struct stat st; if (stat(full, &st) == 0) total += st.st_size; } closedir(dir); return total; } 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.type, hdr.lsn, payload, hdr.payload_len); TEST_ASSERT_EQUAL_INT(0, rc); /* current_lsn high-water must have advanced to at least lsn+1 */ TEST_ASSERT_TRUE(p->current_lsn > 42); /* bug-778: apply_wal_entry is LOGICAL restore only — it must NOT touch the * on-disk CRC chain link. last_crc stays at its initial value (the physical * writer / replay owns it); it must NOT adopt the header's crc. */ TEST_ASSERT_EQUAL_UINT32(WAL_CRC_NONE, 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.type, hdr.lsn, NULL, hdr.payload_len); 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.type, hdr.lsn, NULL, hdr.payload_len); 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.type, hdr.lsn, NULL, hdr.payload_len); 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) { /* NULL persist must return non-zero (EINVAL) */ int rc = apply_wal_entry(NULL, WAL_JOB_UPDATE, 1, NULL, 0); TEST_ASSERT_NOT_EQUAL(0, rc); printf("test_apply_wal_entry_null_guard: PASS\n"); } /* ================================================================== */ /* persist_wal_write_replicated — standby persists replicated entries */ /* (durability: persist-before-ACK). RED before the fix. */ /* ================================================================== */ /* A streamed entry written to the standby's WAL must survive a restart: * a FRESH persist_recover over the same WAL replays it (crc-valid) and * advances current_lsn. Before the fix the segment is empty / nothing * is recovered. */ static void test_repl_write_persists_and_recovers(void) { char tmpdir[256]; persist_t* p = make_test_persist_in(tmpdir, sizeof(tmpdir)); /* Fresh standby: first replicated entry is lsn 1 (1-based; reserve-lsn-0). */ const char payload[] = "replicated_job_update_payload"; int rc = persist_wal_write_replicated(p, /*lsn=*/1, WAL_JOB_UPDATE, payload, (uint32_t)strlen(payload)); TEST_ASSERT_EQUAL_INT(0, rc); /* current_lsn advanced to lsn+1 */ TEST_ASSERT_EQUAL_UINT64(2, p->current_lsn); /* A .wal segment must now exist on disk. */ TEST_ASSERT_TRUE(count_replwr_wal_files(tmpdir) >= 1); TEST_ASSERT_TRUE(total_replwr_wal_bytes(tmpdir) > 0); /* Simulate a restart: drop this persist_t, re-open the same workdir, * and run recovery. The replicated entry must replay (crc-chain valid) * and current_lsn must advance past it. */ destroy_test_persist(p); persist_t* p2 = reopen_test_persist(tmpdir); int rec = persist_recover(p2); TEST_ASSERT_TRUE(rec >= 0); /* Recovery must have advanced current_lsn to 2 (replayed lsn 1). */ TEST_ASSERT_EQUAL_UINT64(2, p2->current_lsn); destroy_test_persist(p2); printf("test_repl_write_persists_and_recovers: PASS\n"); } /* Multiple replicated entries in sequence form a self-consistent crc chain * that the standby's OWN recovery validates with no chain break (all replay). */ static void test_repl_write_crc_chain_recovers(void) { char tmpdir[256]; persist_t* p = make_test_persist_in(tmpdir, sizeof(tmpdir)); const char* payloads[] = { "entry-one", "entry-two", "entry-three" }; for (int i = 0; i < 3; i++) { uint64_t lsn = (uint64_t)(i + 1); /* 1,2,3 */ int rc = persist_wal_write_replicated(p, lsn, WAL_JOB_UPDATE, payloads[i], (uint32_t)strlen(payloads[i])); TEST_ASSERT_EQUAL_INT(0, rc); } TEST_ASSERT_EQUAL_UINT64(4, p->current_lsn); destroy_test_persist(p); /* Fresh recovery must replay ALL three (no crc-chain break) → current_lsn=4. */ persist_t* p2 = reopen_test_persist(tmpdir); int rec = persist_recover(p2); TEST_ASSERT_TRUE(rec >= 0); TEST_ASSERT_EQUAL_UINT64(4, p2->current_lsn); destroy_test_persist(p2); printf("test_repl_write_crc_chain_recovers: PASS\n"); } /* Re-streaming the same lsn must be idempotent: the dedup guard skips the * second write (no double-write, no chain corruption). After a dup write, * recovery still yields exactly the single entry chain. */ static void test_repl_write_dedup_idempotent(void) { char tmpdir[256]; persist_t* p = make_test_persist_in(tmpdir, sizeof(tmpdir)); const char payload[] = "dup-entry"; int rc1 = persist_wal_write_replicated(p, 1, WAL_JOB_UPDATE, payload, (uint32_t)strlen(payload)); TEST_ASSERT_EQUAL_INT(0, rc1); uint64_t lsn_after_first = p->current_lsn; /* expect 2 */ off_t bytes_after_first = total_replwr_wal_bytes(tmpdir); /* Re-stream lsn 1 (already persisted): must be a no-op success. */ int rc2 = persist_wal_write_replicated(p, 1, WAL_JOB_UPDATE, payload, (uint32_t)strlen(payload)); TEST_ASSERT_EQUAL_INT(0, rc2); /* current_lsn must NOT advance further; no extra bytes written. */ TEST_ASSERT_EQUAL_UINT64(lsn_after_first, p->current_lsn); TEST_ASSERT_EQUAL_INT64((long long)bytes_after_first, (long long)total_replwr_wal_bytes(tmpdir)); /* Now legitimately stream lsn 2 — should append fine after the dup. */ int rc3 = persist_wal_write_replicated(p, 2, WAL_JOB_UPDATE, payload, (uint32_t)strlen(payload)); TEST_ASSERT_EQUAL_INT(0, rc3); TEST_ASSERT_EQUAL_UINT64(3, p->current_lsn); destroy_test_persist(p); /* Recovery must replay lsn 1 and lsn 2 (no duplicate / no break). */ persist_t* p2 = reopen_test_persist(tmpdir); int rec = persist_recover(p2); TEST_ASSERT_TRUE(rec >= 0); TEST_ASSERT_EQUAL_UINT64(3, p2->current_lsn); destroy_test_persist(p2); printf("test_repl_write_dedup_idempotent: PASS\n"); } /* WAL_CHECKPOINT must never be persisted via the replicated-write path * (standbys never receive it; it must be rejected, not written). */ static void test_repl_write_rejects_checkpoint(void) { char tmpdir[256]; persist_t* p = make_test_persist_in(tmpdir, sizeof(tmpdir)); int rc = persist_wal_write_replicated(p, 1, WAL_CHECKPOINT, NULL, 0); TEST_ASSERT_NOT_EQUAL(0, rc); /* current_lsn must not advance for a rejected checkpoint. */ TEST_ASSERT_EQUAL_UINT64(1, p->current_lsn); destroy_test_persist(p); printf("test_repl_write_rejects_checkpoint: 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); /* persist-before-ACK durability (standby persists replicated entries) */ RUN_TEST(test_repl_write_persists_and_recovers); RUN_TEST(test_repl_write_crc_chain_recovers); RUN_TEST(test_repl_write_dedup_idempotent); RUN_TEST(test_repl_write_rejects_checkpoint); }