/* test_config_repl.c — Unity tests for config replication Phase 1 */ #define _GNU_SOURCE #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include "commit.h" #include "config_repl.h" #include "controller.capnp.h" #include "ha.h" #include "myconf.h" #include "persist.h" #include "repl.h" #include "request.h" #include "typemap.h" #include "wal_types.h" #include "xce/cluster_def.h" #ifdef XCE_TESTING /* Declared in ha.h — test seam for ha_apply_wal_catchup_entry. * ep and c are passed as NULL in unit tests (ACK path is skipped when * either is NULL — Task 3.1, ha-wal-replication). */ int ha_apply_wal_catchup_entry(ha_state_t *ha, persist_t *p, wal_catchup_entry_t *entry, struct endpoint_t *ep, cluster_t *c); #endif /* Per-test directory helpers — not Unity setUp/tearDown (those are in * controller_test.c). Each test that needs a temp dir calls cr_setup() * at the top and cr_teardown() via the label before returning. */ static char g_tmpdir[256]; /* g_instdir = g_tmpdir/ctrl0 — the effective_dir used by conf_save_config() */ static char g_instdir[300]; static char g_conf_dir[356]; /* tmpdir string for load_primary_config tests (no conf/current) */ static char tmpdir[256]; static void cr_setup(void) { snprintf(g_tmpdir, sizeof(g_tmpdir), "/tmp/test_cr_%d_%ld", (int)getpid(), (long)time(NULL)); mkdir(g_tmpdir, 0700); /* conf_save_config() writes to workdir/name/conf — mirror that here */ snprintf(g_instdir, sizeof(g_instdir), "%s/ctrl0", g_tmpdir); mkdir(g_instdir, 0700); snprintf(g_conf_dir, sizeof(g_conf_dir), "%s/conf", g_instdir); mkdir(g_conf_dir, 0700); } static void cr_teardown(void) { char cmd[512]; snprintf(cmd, sizeof(cmd), "rm -rf %s", g_tmpdir); system(cmd); g_tmpdir[0] = '\0'; g_conf_dir[0] = '\0'; } /* ── conf_read_current_filename ─────────────────── */ static void test_conf_read_current_filename_missing(void) { cr_setup(); char out[256] = {0}; int rc = conf_read_current_filename(g_tmpdir, out, sizeof(out)); TEST_ASSERT_EQUAL_INT(-ENOENT, rc); cr_teardown(); printf("test_conf_read_current_filename_missing: PASS\n"); } static void test_conf_read_current_filename_empty_file(void) { cr_setup(); char ptr_path[400]; snprintf(ptr_path, sizeof(ptr_path), "%s/current", g_conf_dir); int fd = open(ptr_path, O_WRONLY | O_CREAT | O_TRUNC, 0600); TEST_ASSERT_NOT_EQUAL(-1, fd); close(fd); char out[256] = {0}; /* conf/current lives under g_instdir/conf — pass g_instdir as base */ int rc = conf_read_current_filename(g_instdir, out, sizeof(out)); TEST_ASSERT_EQUAL_INT(-ENODATA, rc); cr_teardown(); printf("test_conf_read_current_filename_empty_file: PASS\n"); } static void test_conf_read_current_filename_path_separator(void) { cr_setup(); char ptr_path[400]; snprintf(ptr_path, sizeof(ptr_path), "%s/current", g_conf_dir); FILE *f = fopen(ptr_path, "w"); fprintf(f, "cluster.bin/../evil\n"); fclose(f); char out[256] = {0}; /* conf/current lives under g_instdir/conf — pass g_instdir as base */ int rc = conf_read_current_filename(g_instdir, out, sizeof(out)); TEST_ASSERT_EQUAL_INT(-EINVAL, rc); cr_teardown(); printf("test_conf_read_current_filename_path_separator: PASS\n"); } static void test_conf_read_current_filename_bad_prefix(void) { cr_setup(); char ptr_path[400]; snprintf(ptr_path, sizeof(ptr_path), "%s/current", g_conf_dir); FILE *f = fopen(ptr_path, "w"); fprintf(f, "badprefix.aabbccddeeff\n"); fclose(f); char out[256] = {0}; /* conf/current lives under g_instdir/conf — pass g_instdir as base */ int rc = conf_read_current_filename(g_instdir, out, sizeof(out)); TEST_ASSERT_EQUAL_INT(-EINVAL, rc); cr_teardown(); printf("test_conf_read_current_filename_bad_prefix: PASS\n"); } static void test_conf_read_current_filename_wrong_hex_length(void) { cr_setup(); char ptr_path[400]; snprintf(ptr_path, sizeof(ptr_path), "%s/current", g_conf_dir); FILE *f = fopen(ptr_path, "w"); fprintf(f, "cluster.bin.aabb\n"); /* only 4 hex chars, need 12 */ fclose(f); char out[256] = {0}; /* conf/current lives under g_instdir/conf — pass g_instdir as base */ int rc = conf_read_current_filename(g_instdir, out, sizeof(out)); TEST_ASSERT_EQUAL_INT(-EINVAL, rc); cr_teardown(); printf("test_conf_read_current_filename_wrong_hex_length: PASS\n"); } /* ── conf_save_config + conf_read_current_filename round-trip ─ */ static void test_conf_save_creates_pointer_file(void) { cr_setup(); const uint8_t data[] = "fake capnproto config blob"; size_t len = sizeof(data) - 1; uint8_t sha256[32]; SHA256(data, len, sha256); myconf_t myconf; memset(&myconf, 0, sizeof(myconf)); myconf.workdir = g_tmpdir; myconf.name = "ctrl0"; myconf.conf_keep_n = 2; myconf.conffile = NULL; int rc = conf_save_config(&myconf, data, len, sha256); TEST_ASSERT_EQUAL_INT(0, rc); /* conf/current must exist */ char ptr_path[400]; snprintf(ptr_path, sizeof(ptr_path), "%s/current", g_conf_dir); TEST_ASSERT_EQUAL_INT(0, access(ptr_path, F_OK)); free(myconf.conffile); cr_teardown(); printf("test_conf_save_creates_pointer_file: PASS\n"); } static void test_conf_save_filename_has_12_hex_suffix(void) { cr_setup(); const uint8_t data[] = "another fake blob"; size_t len = sizeof(data) - 1; uint8_t sha256[32]; SHA256(data, len, sha256); myconf_t myconf; memset(&myconf, 0, sizeof(myconf)); myconf.workdir = g_tmpdir; myconf.name = "ctrl0"; myconf.conf_keep_n = 2; myconf.conffile = NULL; conf_save_config(&myconf, data, len, sha256); char fname[256] = {0}; /* conf_save_config() writes to workdir/name/conf — read from g_instdir */ int rc = conf_read_current_filename(g_instdir, fname, sizeof(fname)); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_EQUAL_INT(0, strncmp(fname, "cluster.bin.", 12)); const char *hex = fname + 12; TEST_ASSERT_EQUAL_INT(12, (int)strlen(hex)); TEST_ASSERT_EQUAL_INT(12, (int)strspn(hex, "0123456789abcdef")); free(myconf.conffile); cr_teardown(); printf("test_conf_save_filename_has_12_hex_suffix: PASS\n"); } /* ── conf_trim_old_versions ─────────────────────── */ static void test_conf_trim_keeps_only_n_versions(void) { cr_setup(); myconf_t myconf; memset(&myconf, 0, sizeof(myconf)); myconf.workdir = g_tmpdir; myconf.name = "ctrl0"; myconf.conf_keep_n = 2; myconf.conffile = NULL; uint8_t sha256[32]; for (int i = 0; i < 3; i++) { char data[32]; snprintf(data, sizeof(data), "config version %d", i); SHA256((uint8_t *)data, strlen(data), sha256); usleep(15000); /* ensure distinct mtime ordering */ conf_save_config(&myconf, (uint8_t *)data, strlen(data), sha256); free(myconf.conffile); myconf.conffile = NULL; } /* Count cluster.bin.* files (exclude .tmp) */ int count = 0; DIR *d = opendir(g_conf_dir); TEST_ASSERT_NOT_NULL(d); struct dirent *e; while ((e = readdir(d)) != NULL) { if (strncmp(e->d_name, "cluster.bin.", 12) == 0 && strstr(e->d_name, ".tmp") == NULL) count++; } closedir(d); TEST_ASSERT_EQUAL_INT(2, count); free(myconf.conffile); cr_teardown(); printf("test_conf_trim_keeps_only_n_versions: PASS\n"); } /* ── conf_get_saved_tag ─────────────────────────── */ static void test_conf_get_saved_tag_no_config(void) { cr_setup(); /* conf_get_saved_tag looks for conf/current under its argument; * after the path fix, conf lives under workdir/name = g_instdir */ uint64_t tag = conf_get_saved_tag(g_instdir); TEST_ASSERT_EQUAL_UINT64(0, tag); cr_teardown(); printf("test_conf_get_saved_tag_no_config: PASS\n"); } static void test_conf_get_saved_tag_after_save(void) { cr_setup(); const uint8_t data[] = "tag test blob"; size_t len = sizeof(data) - 1; uint8_t sha256[32]; SHA256(data, len, sha256); myconf_t myconf; memset(&myconf, 0, sizeof(myconf)); myconf.workdir = g_tmpdir; myconf.name = "ctrl0"; myconf.conf_keep_n = 2; myconf.conffile = NULL; conf_save_config(&myconf, data, len, sha256); free(myconf.conffile); uint64_t tag = conf_get_saved_tag(g_instdir); TEST_ASSERT_NOT_EQUAL(0, tag); /* Tag must equal first 8 bytes of sha256 in host order */ uint64_t expected; memcpy(&expected, sha256, 8); /* be64toh conversion */ #if __BYTE_ORDER__ == __ORDER_LITTLE_ENDIAN__ expected = __builtin_bswap64(expected); #endif TEST_ASSERT_EQUAL_UINT64(expected, tag); cr_teardown(); printf("test_conf_get_saved_tag_after_save: PASS\n"); } /* ── conf_check_stale ───────────────────────────── */ static void test_conf_check_stale_disabled(void) { cr_setup(); /* timeout 0 = disabled, never stale */ TEST_ASSERT_EQUAL_INT(0, conf_check_stale(g_instdir, 0)); cr_teardown(); printf("test_conf_check_stale_disabled: PASS\n"); } static void test_conf_check_stale_fresh(void) { cr_setup(); const uint8_t data[] = "stale test blob"; uint8_t sha256[32]; SHA256(data, sizeof(data) - 1, sha256); myconf_t myconf; memset(&myconf, 0, sizeof(myconf)); myconf.workdir = g_tmpdir; myconf.name = "ctrl0"; myconf.conf_keep_n = 2; myconf.conffile = NULL; conf_save_config(&myconf, data, sizeof(data) - 1, sha256); free(myconf.conffile); /* Just saved — not stale with 24h timeout */ TEST_ASSERT_EQUAL_INT(0, conf_check_stale(g_instdir, 86400)); cr_teardown(); printf("test_conf_check_stale_fresh: PASS\n"); } static void test_conf_check_stale_aged(void) { cr_setup(); const uint8_t data[] = "old config blob"; uint8_t sha256[32]; SHA256(data, sizeof(data) - 1, sha256); myconf_t myconf; memset(&myconf, 0, sizeof(myconf)); myconf.workdir = g_tmpdir; myconf.name = "ctrl0"; myconf.conf_keep_n = 2; myconf.conffile = NULL; conf_save_config(&myconf, data, sizeof(data) - 1, sha256); free(myconf.conffile); /* Age the versioned file by 20 seconds via utime */ char fname[256] = {0}; /* conf_save_config() writes to workdir/name/conf — read from g_instdir */ conf_read_current_filename(g_instdir, fname, sizeof(fname)); char ver_path[512]; snprintf(ver_path, sizeof(ver_path), "%s/%s", g_conf_dir, fname); struct utimbuf old_time; old_time.actime = time(NULL) - 20; old_time.modtime = time(NULL) - 20; utime(ver_path, &old_time); /* 5-second threshold → stale */ TEST_ASSERT_EQUAL_INT(1, conf_check_stale(g_instdir, 5)); /* 30-second threshold → not stale */ TEST_ASSERT_EQUAL_INT(0, conf_check_stale(g_instdir, 30)); cr_teardown(); printf("test_conf_check_stale_aged: PASS\n"); } /* ── config_contains_reserved_name ─────────────── */ static void test_config_contains_reserved_name_absent(void) { /* NULL config — no reserved name */ TEST_ASSERT_EQUAL_INT(0, config_contains_reserved_name(NULL)); printf("test_config_contains_reserved_name_absent: PASS\n"); } /* ── sanitise_cn ────────────────────────────────── */ static void test_sanitise_cn_printable(void) { char *out = sanitise_cn("ctrl01.example.com", 64); TEST_ASSERT_EQUAL_STRING("ctrl01.example.com", out); printf("test_sanitise_cn_printable: PASS\n"); } static void test_sanitise_cn_control_chars(void) { char *out = sanitise_cn("evil\ninjection", 64); /* newline replaced with '?' */ TEST_ASSERT_NOT_NULL(strchr(out, '?')); TEST_ASSERT_NULL(strchr(out, '\n')); printf("test_sanitise_cn_control_chars: PASS\n"); } static void test_sanitise_cn_truncated(void) { char *out = sanitise_cn("abcdefghij", 5); TEST_ASSERT_EQUAL_INT(5, (int)strlen(out)); printf("test_sanitise_cn_truncated: PASS\n"); } /* ── load_primary_config (via test-only helper) ── */ #ifdef XCE_TESTING static void test_load_primary_config_missing_returns_error(void) { /* Create a minimal cluster_t with no conf/current file */ cluster_t c; memset(&c, 0, sizeof(c)); myconf_t myconf; memset(&myconf, 0, sizeof(myconf)); /* Use a fresh tmpdir with no conf/current */ snprintf(tmpdir, sizeof(tmpdir), "/tmp/test_lpc_%d_%ld", (int)getpid(), (long)time(NULL)); mkdir(tmpdir, 0700); myconf.workdir = tmpdir; myconf.name = "ctrl0"; myconf.conf_keep_n = 2; myconf.conf_stale_timeout = 86400; myconf.conffile = NULL; c.myconf = &myconf; /* No conf/current exists — must fail (not FATAL in test context) */ int rc = load_primary_config_for_test(&c); TEST_ASSERT_TRUE(rc < 0); /* conffile must still be NULL (no heap allocation on failure) */ TEST_ASSERT_NULL(myconf.conffile); /* Cleanup */ char cmd[512]; snprintf(cmd, sizeof(cmd), "rm -rf %s", tmpdir); system(cmd); printf("test_load_primary_config_missing_returns_error: PASS\n"); } #endif /* ── test_conf_save_config_path_consistency ─────── */ static void test_conf_save_config_path_consistency(void) { /* Verify conf_save_config() writes to workdir/name/conf, not workdir/conf. * This mirrors the path that load_primary_config() reads from. */ char workdir[256]; snprintf(workdir, sizeof(workdir), "/tmp/test_pathbug_%d", (int)getpid()); char instdir[300]; snprintf(instdir, sizeof(instdir), "%s/ctrl01", workdir); char conf_dir[356]; snprintf(conf_dir, sizeof(conf_dir), "%s/conf", instdir); mkdir(workdir, 0700); /* conf_save_config() builds effective_dir = workdir/name internally, * then mkdirs effective_dir/conf — but effective_dir itself must exist. */ mkdir(instdir, 0700); /* Fake 32-byte sha256 and dummy payload */ uint8_t sha256[32]; memset(sha256, 0xab, sizeof(sha256)); const uint8_t payload[] = "dummy config bytes for path test"; size_t plen = sizeof(payload) - 1; myconf_t myconf; memset(&myconf, 0, sizeof(myconf)); myconf.workdir = workdir; myconf.name = "ctrl01"; myconf.conf_keep_n = 2; myconf.conffile = NULL; int rc = conf_save_config(&myconf, payload, plen, sha256); TEST_ASSERT_EQUAL_INT(0, rc); /* Versioned file must be under workdir/name/conf/, not workdir/conf/ */ TEST_ASSERT_NOT_NULL(myconf.conffile); TEST_ASSERT_EQUAL_INT(0, strncmp(myconf.conffile, conf_dir, strlen(conf_dir))); /* conf/current pointer file must exist at workdir/name/conf/current */ char ptr_path[400]; snprintf(ptr_path, sizeof(ptr_path), "%s/current", conf_dir); TEST_ASSERT_EQUAL_INT(0, access(ptr_path, F_OK)); /* workdir/conf/ must NOT exist (wrong location) */ char wrong_conf[300]; snprintf(wrong_conf, sizeof(wrong_conf), "%s/conf", workdir); struct stat st; TEST_ASSERT_NOT_EQUAL(0, stat(wrong_conf, &st)); free(myconf.conffile); /* Cleanup */ char cmd[512]; snprintf(cmd, sizeof(cmd), "rm -rf %s", workdir); system(cmd); printf("test_conf_save_config_path_consistency: PASS\n"); } /* ── WAL_CONFIG_MARKER codec (ha-term-safety task 4.1) ─────────────── * The config-change marker payload carries ONLY {version_tag, sha256} — * a reference, never the configuration bytes (R7.2/R7.4). These tests * exercise the fixed-size encode/decode codec that rides inside the * raft_entry_blob_t envelope with type == WAL_CONFIG_MARKER. */ /* Round-trip: encode {version_tag, sha256} then decode → identical, and the * encoded form is exactly the fixed marker size (tag + 32-byte hash) with NO * configuration bytes appended. */ static void test_config_marker_encode_decode_roundtrip(void) { const uint64_t version_tag = 0x0123456789abcdefULL; uint8_t sha256[32]; for (int i = 0; i < 32; i++) sha256[i] = (uint8_t)(0xA0 + i); uint8_t buf[64]; size_t enc_len = 0; int rc = config_marker_encode(version_tag, sha256, buf, sizeof(buf), &enc_len); TEST_ASSERT_EQUAL_INT(0, rc); /* Encoded form carries NO config bytes: exactly 8 (tag) + 32 (sha256). */ TEST_ASSERT_EQUAL_UINT(WAL_CONFIG_MARKER_PAYLOAD_LEN, enc_len); TEST_ASSERT_EQUAL_UINT(8u + 32u, enc_len); uint64_t got_tag = 0; uint8_t got_sha[32]; memset(got_sha, 0, sizeof(got_sha)); rc = config_marker_decode(buf, enc_len, &got_tag, got_sha); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_EQUAL_HEX64(version_tag, got_tag); TEST_ASSERT_EQUAL_HEX8_ARRAY(sha256, got_sha, 32); printf("test_config_marker_encode_decode_roundtrip: PASS\n"); } /* The WAL entry type value must be exactly 50 and must not collide with the * neighbouring WAL_NOOP (40) or WAL_CHECKPOINT (100) types (design.md). */ static void test_config_marker_enum_value_is_50(void) { TEST_ASSERT_EQUAL_INT(50, WAL_CONFIG_MARKER); TEST_ASSERT_NOT_EQUAL(WAL_NOOP, WAL_CONFIG_MARKER); TEST_ASSERT_NOT_EQUAL(WAL_CHECKPOINT, WAL_CONFIG_MARKER); /* Strictly between WAL_NOOP and WAL_CHECKPOINT. */ TEST_ASSERT_TRUE(WAL_CONFIG_MARKER > WAL_NOOP); TEST_ASSERT_TRUE(WAL_CONFIG_MARKER < WAL_CHECKPOINT); printf("test_config_marker_enum_value_is_50: PASS\n"); } /* A truncated (short) buffer must fail cleanly with -EINVAL, never read past * the end of the supplied buffer. */ static void test_config_marker_decode_truncated_fails(void) { uint8_t sha256[32]; memset(sha256, 0x5c, sizeof(sha256)); uint8_t buf[WAL_CONFIG_MARKER_PAYLOAD_LEN]; size_t enc_len = 0; TEST_ASSERT_EQUAL_INT(0, config_marker_encode(0xdeadbeefcafef00dULL, sha256, buf, sizeof(buf), &enc_len)); uint64_t tag = 0; uint8_t sha_out[32]; /* One byte short of a full marker payload → reject, no OOB read. */ int rc = config_marker_decode(buf, enc_len - 1, &tag, sha_out); TEST_ASSERT_EQUAL_INT(-EINVAL, rc); /* Zero-length buffer → reject. */ rc = config_marker_decode(buf, 0, &tag, sha_out); TEST_ASSERT_EQUAL_INT(-EINVAL, rc); printf("test_config_marker_decode_truncated_fails: PASS\n"); } /* A too-small output buffer must fail with -ENOSPC and not overflow. */ static void test_config_marker_encode_short_buffer_fails(void) { uint8_t sha256[32]; memset(sha256, 0x11, sizeof(sha256)); uint8_t small[WAL_CONFIG_MARKER_PAYLOAD_LEN - 1]; size_t enc_len = 123; /* sentinel — must be left untouched on failure */ int rc = config_marker_encode(7, sha256, small, sizeof(small), &enc_len); TEST_ASSERT_EQUAL_INT(-ENOSPC, rc); printf("test_config_marker_encode_short_buffer_fails: PASS\n"); } /* ── Config marker append + committed-config derive (task 4.2) ──────── * * These bring a node up as the raft LEADER so config_append_marker routes the * WAL_CONFIG_MARKER through ha_primary_log_append → raft_log_append (the same * unified-log path job state uses), and assert the committed config is DERIVED * from the latest committed marker within the committed prefix (R7.1/7.3/7.6/ * 7.7/7.11). Mirrors make_cluster/make_leader from test_term_safety_adapter.c. */ static cluster_t *t42_make_cluster(const char *tmpdir) { cluster_t *c = calloc(1, sizeof(cluster_t)); if (!c) return NULL; persist_config_t cfg = persist_config_defaults(); cfg.wal_sync_ms = 5000; cfg.snap_keep_n = 2; persist_set_workdir(&c->persist, tmpdir, "ctrl-test"); if (persist_init(&c->persist, &cfg, c) != 0) { free(c); return NULL; } if (persist_start(&c->persist) != 0) { persist_fini(&c->persist); free(c); return NULL; } c->ha.raft_c = c; return c; } static void t42_teardown(cluster_t *c) { if (!c) return; persist_stop(&c->persist); persist_fini(&c->persist); free(c); } /* Bring the node up as the raft LEADER at `term` (so raft_log_append is enabled * — it returns -EPERM off the leader; see make_leader in the adapter suite). */ static void t42_make_leader(cluster_t *c, uint64_t term, int n_peers) { atomic_store(&c->ha.role, CTRL_PRIMARY); atomic_store(&c->ha.term, term); ha_init_raft(&c->ha, c, n_peers); c->ha.raft.term = term; c->ha.raft.role = RAFT_LEADER; } /* Fill an sha256 with a deterministic pattern; tag = be64 of first 8 bytes, * matching the config_repl version_tag convention (config_version_tag). */ static uint64_t t42_make_ref(uint8_t sha256[32], uint8_t seed) { for (int i = 0; i < 32; i++) sha256[i] = (uint8_t)(seed + i); uint64_t tag; memcpy(&tag, sha256, 8); return be64toh(tag); } /* (a) Appending a config change writes a WAL_CONFIG_MARKER through the unified * log: the entry lands at an LSN with the live election term stamped, and its * payload decodes to {version_tag, sha256} (R7.1/7.3). */ static void test_config_marker_append_via_unified_log(void) { char tmp[256]; strcpy(tmp, "/tmp/xce_cr42_append_XXXXXX"); TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); const uint64_t T = 7; cluster_t *c = t42_make_cluster(tmp); TEST_ASSERT_NOT_NULL(c); t42_make_leader(c, T, /*n_peers*/1); uint8_t sha[32]; uint64_t tag = t42_make_ref(sha, 0x10); uint64_t lsn = 0; TEST_ASSERT_EQUAL_INT(0, config_append_marker(c, tag, sha, &lsn)); /* Library assigned LSN 1 and stamped the live term for that entry. */ TEST_ASSERT_EQUAL_UINT64(1, lsn); TEST_ASSERT_EQUAL_UINT64(1, c->ha.raft.log_len); TEST_ASSERT_EQUAL_UINT64(T, persist_entry_term(&c->persist, 1)); t42_teardown(c); printf("test_config_marker_append_via_unified_log: PASS\n"); } /* (b) After the marker COMMITS (committed_lsn covers its LSN), the derived * committed-config reference == that marker's {version_tag, sha256} (R7.7). */ static void test_committed_marker_derives_after_commit(void) { char tmp[256]; strcpy(tmp, "/tmp/xce_cr42_derive_XXXXXX"); TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); cluster_t *c = t42_make_cluster(tmp); TEST_ASSERT_NOT_NULL(c); t42_make_leader(c, /*term*/4, /*n_peers*/1); /* A job entry then a config marker, so the marker is NOT at LSN 1. */ wal_jobstate_t js = { .jobid = 1, .state = 1 }; uint64_t jlsn = 0; TEST_ASSERT_EQUAL_INT(0, ha_primary_log_append(c, WAL_JOB_UPDATE, &js, sizeof(js), &jlsn)); uint8_t sha[32]; uint64_t tag = t42_make_ref(sha, 0x40); uint64_t mlsn = 0; TEST_ASSERT_EQUAL_INT(0, config_append_marker(c, tag, sha, &mlsn)); TEST_ASSERT_EQUAL_UINT64(2, mlsn); TEST_ASSERT_EQUAL_INT(0, persist_flush(&c->persist)); /* Drive committed_lsn to the marker's LSN (the commit advance source). */ atomic_store_explicit(&c->persist.committed_lsn, mlsn, memory_order_release); uint64_t got_tag = 0; uint8_t got_sha[32]; memset(got_sha, 0, sizeof(got_sha)); TEST_ASSERT_EQUAL_INT(0, config_committed_marker(c, &got_tag, got_sha)); TEST_ASSERT_EQUAL_HEX64(tag, got_tag); TEST_ASSERT_EQUAL_HEX8_ARRAY(sha, got_sha, 32); t42_teardown(c); printf("test_committed_marker_derives_after_commit: PASS\n"); } /* (c) An UNCOMMITTED marker (LSN > committed_lsn) does NOT change the derived * committed config: with committed_lsn at the first marker, a later second * marker is invisible until committed (R7.7/7.11). */ static void test_uncommitted_marker_not_yet_committed_config(void) { char tmp[256]; strcpy(tmp, "/tmp/xce_cr42_uncommit_XXXXXX"); TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); cluster_t *c = t42_make_cluster(tmp); TEST_ASSERT_NOT_NULL(c); t42_make_leader(c, /*term*/5, /*n_peers*/1); uint8_t sha1[32], sha2[32]; uint64_t tag1 = t42_make_ref(sha1, 0x10); uint64_t tag2 = t42_make_ref(sha2, 0x80); uint64_t l1 = 0, l2 = 0; TEST_ASSERT_EQUAL_INT(0, config_append_marker(c, tag1, sha1, &l1)); TEST_ASSERT_EQUAL_INT(0, config_append_marker(c, tag2, sha2, &l2)); TEST_ASSERT_EQUAL_UINT64(1, l1); TEST_ASSERT_EQUAL_UINT64(2, l2); TEST_ASSERT_EQUAL_INT(0, persist_flush(&c->persist)); /* Only the FIRST marker is committed. The second is uncommitted. */ atomic_store_explicit(&c->persist.committed_lsn, l1, memory_order_release); uint64_t got_tag = 0; uint8_t got_sha[32]; TEST_ASSERT_EQUAL_INT(0, config_committed_marker(c, &got_tag, got_sha)); /* Derived config is the FIRST (committed) marker, NOT the uncommitted one. */ TEST_ASSERT_EQUAL_HEX64(tag1, got_tag); TEST_ASSERT_EQUAL_HEX8_ARRAY(sha1, got_sha, 32); /* With NOTHING committed, there is no committed config marker at all. */ atomic_store_explicit(&c->persist.committed_lsn, 0ULL, memory_order_release); TEST_ASSERT_EQUAL_INT(-ENOENT, config_committed_marker(c, &got_tag, got_sha)); t42_teardown(c); printf("test_uncommitted_marker_not_yet_committed_config: PASS\n"); } /* (d) A divergent config marker at a conflicting (term,lsn) is truncated by the * reconcile path exactly like a job entry: on_truncate_from(index) removes the * marker (and everything above), leaving the committed prefix intact. This * proves the marker is an ORDINARY log entry under (term,lsn) matching (R7.6). * * We append a committed job at LSN 1, then a divergent marker at LSN 2, drive * committed_lsn to 1, then truncate from LSN 2 (the conflict point) and confirm * the marker is gone while the committed job prefix survives. */ static void test_divergent_marker_truncated_like_job_entry(void) { char tmp[256]; strcpy(tmp, "/tmp/xce_cr42_trunc_XXXXXX"); TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); cluster_t *c = t42_make_cluster(tmp); TEST_ASSERT_NOT_NULL(c); t42_make_leader(c, /*term*/6, /*n_peers*/1); /* LSN 1: committed job entry. LSN 2: a (divergent) config marker. */ wal_jobstate_t js = { .jobid = 9, .state = 2 }; uint64_t jlsn = 0; TEST_ASSERT_EQUAL_INT(0, ha_primary_log_append(c, WAL_JOB_UPDATE, &js, sizeof(js), &jlsn)); TEST_ASSERT_EQUAL_UINT64(1, jlsn); uint8_t sha[32]; uint64_t tag = t42_make_ref(sha, 0x55); uint64_t mlsn = 0; TEST_ASSERT_EQUAL_INT(0, config_append_marker(c, tag, sha, &mlsn)); TEST_ASSERT_EQUAL_UINT64(2, mlsn); TEST_ASSERT_EQUAL_INT(0, persist_flush(&c->persist)); /* The committed prefix is LSN 1 (the job). The marker at LSN 2 is the * divergent suffix. The derived committed config is therefore NONE yet. */ atomic_store_explicit(&c->persist.committed_lsn, 1ULL, memory_order_release); uint64_t got_tag = 0; uint8_t got_sha[32]; TEST_ASSERT_EQUAL_INT(-ENOENT, config_committed_marker(c, &got_tag, got_sha)); /* Reconcile truncates from the conflict point (LSN 2) exactly as it would a * job entry — the SAME on_truncate_from seam used by (term,lsn) matching. */ TEST_ASSERT_EQUAL_INT(0, ha_cb_truncate_from_for_test(&c->ha, /*index*/2)); TEST_ASSERT_EQUAL_INT(0, persist_flush(&c->persist)); /* The marker is gone (term index shrank); the committed job prefix is intact. */ TEST_ASSERT_EQUAL_UINT64(0, persist_entry_term(&c->persist, 2)); TEST_ASSERT_EQUAL_UINT64(6, persist_entry_term(&c->persist, 1)); /* Still no committed config marker — the divergent one never committed. */ TEST_ASSERT_EQUAL_INT(-ENOENT, config_committed_marker(c, &got_tag, got_sha)); t42_teardown(c); printf("test_divergent_marker_truncated_like_job_entry: PASS\n"); } /* (e) INTEGRATION (Finding 1 remediation): drive a config change through the * LIVE commit path (config_commit_change — the helper do_upgrade_config_with_bytes * now calls) and assert (i) a WAL_CONFIG_MARKER lands at the LIVE term, and * (ii) once the marker COMMITS through the real WAL commit barrier, the DERIVED * committed config (config_committed_marker) equals it. This proves the live * config-change path appends + commits + derives the marker — not the retired * term-blind committed_config_tag push/ack/advance. * * A single-member commit engine is wired so config_commit_change's WAL barrier * (repl_wait_majority_ack/OPCode_walAck → commit_submit) self-quorums and * advances committed_lsn for real, exactly as the production single-node path. */ static void test_config_commit_change_live_path_commits_and_derives(void) { char tmp[256]; strcpy(tmp, "/tmp/xce_cr42_live_XXXXXX"); TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); const uint64_t T = 9; cluster_t *c = t42_make_cluster(tmp); TEST_ASSERT_NOT_NULL(c); t42_make_leader(c, T, /*n_peers*/1); /* Single-member commit engine: the marker's WAL LSN self-quorums, so * driving commit_submit on it advances committed_lsn through the REAL commit * machinery — the same path that commits the marker in production once a * majority acks the WAL stream (no manual committed_lsn poke). */ TEST_ASSERT_EQUAL_INT(0, commit_engine_init(&c->commit, &c->persist, /*n_peers*/1, /*n_vcpus*/8)); repl_set_commit_engine(&c->persist.repl, &c->commit); uint8_t sha[32]; uint64_t tag = t42_make_ref(sha, 0x30); /* The LIVE entry point do_upgrade_config_with_bytes now calls: append the * WAL_CONFIG_MARKER consensus record + out-of-band byte push + the existing * majority barrier. With no standbys (single member) the byte barrier * self-quorums and the call returns 0. */ int acked = -1; TEST_ASSERT_EQUAL_INT(0, config_commit_change(c, (const uint8_t *)"cfg-bytes", 9, sha, tag, &acked)); /* (i) The marker landed at LSN 1 with the LIVE election term stamped. */ TEST_ASSERT_EQUAL_UINT64(1, c->ha.raft.log_len); TEST_ASSERT_EQUAL_UINT64(T, persist_entry_term(&c->persist, 1)); /* No standbys registered → zero out-of-band pushes (barrier owns the verdict). */ TEST_ASSERT_EQUAL_INT(0, acked); /* Make the marker durable (advances durable_lsn) then commit its WAL LSN * through the real commit engine (self-quorum), exactly as a majority WAL * ack does in production — this advances committed_lsn so the derive sees * the marker in the committed prefix. */ TEST_ASSERT_EQUAL_INT(0, persist_flush(&c->persist)); TEST_ASSERT_EQUAL_INT(0, commit_submit(&c->commit, /*marker LSN*/1, COMMIT_MODE_WAIT, 1000)); TEST_ASSERT_TRUE( atomic_load_explicit(&c->persist.committed_lsn, memory_order_acquire) >= 1); /* (ii) The DERIVED committed config equals the change just committed. */ uint64_t got_tag = 0; uint8_t got_sha[32]; TEST_ASSERT_EQUAL_INT(0, config_committed_marker(c, &got_tag, got_sha)); TEST_ASSERT_EQUAL_HEX64(tag, got_tag); TEST_ASSERT_EQUAL_HEX8_ARRAY(sha, got_sha, 32); /* committed_config_tag is now a CACHE mirroring the derived marker. */ TEST_ASSERT_EQUAL_HEX64(tag, atomic_load_explicit(&c->ha.committed_config_tag, memory_order_acquire)); commit_engine_fini(&c->commit); t42_teardown(c); printf("test_config_commit_change_live_path_commits_and_derives: PASS\n"); } /* (f) COMPACTION SAFETY (Finding 2 remediation): a checkpoint that TRIMS the WAL * segment holding a committed marker must NOT lose the derived committed config. * Without the sentinel-carried reference + fallback, the live-WAL scan regresses * to -ENOENT after the trim. We commit a marker, force a NEW segment + checkpoint * (which trims the old segment containing the marker), and assert the derive * STILL returns the committed reference (R7.7/R7.10). */ static void test_committed_marker_survives_checkpoint_trim(void) { char tmp[256]; strcpy(tmp, "/tmp/xce_cr42_trim_XXXXXX"); TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); const uint64_t T = 11; cluster_t *c = t42_make_cluster(tmp); TEST_ASSERT_NOT_NULL(c); t42_make_leader(c, T, /*n_peers*/1); TEST_ASSERT_EQUAL_INT(0, commit_engine_init(&c->commit, &c->persist, /*n_peers*/1, /*n_vcpus*/8)); repl_set_commit_engine(&c->persist.repl, &c->commit); uint8_t sha[32]; uint64_t tag = t42_make_ref(sha, 0x70); /* Append the marker (segment 0, the lowest-LSN segment) and commit its WAL * LSN through the real engine (self-quorum) so it enters the committed * prefix — the production marker-commit path. */ int acked = -1; TEST_ASSERT_EQUAL_INT(0, config_commit_change(c, (const uint8_t *)"cfg", 3, sha, tag, &acked)); TEST_ASSERT_EQUAL_INT(0, persist_flush(&c->persist)); TEST_ASSERT_EQUAL_INT(0, commit_submit(&c->commit, /*marker LSN*/1, COMMIT_MODE_WAIT, 1000)); uint64_t committed = atomic_load_explicit(&c->persist.committed_lsn, memory_order_acquire); TEST_ASSERT_TRUE(committed >= 1); /* Sanity: before any trim, the live-WAL scan finds the marker. */ uint64_t got_tag = 0; uint8_t got_sha[32]; TEST_ASSERT_EQUAL_INT(0, config_committed_marker(c, &got_tag, got_sha)); TEST_ASSERT_EQUAL_HEX64(tag, got_tag); /* Roll to a NEW segment so the checkpoint's trim has an older segment (the * one holding the marker) to unlink — the compaction hazard. */ TEST_ASSERT_EQUAL_INT(0, persist_test_force_new_segment(&c->persist)); /* Checkpoint: stamps the committed-marker reference into the sentinel, then * trims the older segment (which physically held the marker). */ TEST_ASSERT_EQUAL_INT(0, persist_checkpoint(&c->persist)); /* The marker's segment is gone, so the live-WAL scan alone would regress to * -ENOENT; the derive MUST fall back to the durable sentinel reference. */ got_tag = 0; memset(got_sha, 0, sizeof(got_sha)); TEST_ASSERT_EQUAL_INT(0, config_committed_marker(c, &got_tag, got_sha)); TEST_ASSERT_EQUAL_HEX64(tag, got_tag); TEST_ASSERT_EQUAL_HEX8_ARRAY(sha, got_sha, 32); commit_engine_fini(&c->commit); t42_teardown(c); printf("test_committed_marker_survives_checkpoint_trim: PASS\n"); } /* ── Config byte delivery: hash-verify-before-activate + serve-fence (task 4.3) ─ * * config_activate_committed is the CONSUMER of config_committed_marker (4.2's * derive). It keys the out-of-band config-push byte delivery to the committed * marker's {version_tag, sha256}: it VERIFIES the locally-held bytes' sha256 * against the committed marker BEFORE activating (R7.8), and on a mismatch (or * bytes absent locally) keeps the member from serving under the new configuration * and KEEPS RETRYING rather than activating mismatched config (R7.9) — the loud * non-serving fence that NEVER self-terminates. On recovery it activates the * configuration referenced by the LATEST committed marker (R7.10). * * These tests reuse the 4.2 leader fixture (t42_*) but additionally wire c->myconf * so the locally-held config bytes path (conf_save_config / conf_read_raw / * load_primary_config) is exercised. */ /* Build a VALID Cap'n Proto ClusterConfig blob (decodable by load_primary_config) * so the match path can actually ACTIVATE. Mirrors cb5_make_config_blob. */ static uint8_t *t43_make_config_blob(size_t *len_out) { cluster_config_t cfg; struct capn cap; struct capn_segment *cs; ClusterConfig_ptr p; uint8_t *buf; size_t sz; memset(&cfg, 0, sizeof(cfg)); cfg.clustername = (char *)"t43-upgrade"; cfg.port = 9991; cfg.config_generation = 1; capn_init_malloc(&cap); cs = capn_root(&cap).seg; encode_ClusterConfig_ptr(cs, &p, &cfg); capn_setp(capn_root(&cap), 0, p.p); sz = (size_t)capn_size(&cap); buf = (uint8_t *)malloc(sz); TEST_ASSERT_NOT_NULL(buf); capn_write_mem(&cap, buf, sz, 0); capn_free(&cap); *len_out = sz; return buf; } /* Wire a myconf_t into the cluster so conf_save_config / load_primary_config * resolve under //conf (the production layout). */ static myconf_t *t43_attach_myconf(cluster_t *c, const char *tmpdir) { myconf_t *mc = calloc(1, sizeof(myconf_t)); TEST_ASSERT_NOT_NULL(mc); mc->workdir = strdup(tmpdir); mc->name = strdup("ctrl-test"); mc->conf_keep_n = 2; mc->conffile = NULL; c->myconf = mc; return mc; } static void t43_free_myconf(cluster_t *c) { myconf_t *mc = c->myconf; if (!mc) return; if (mc->cluster_config) { free_ClusterConfig(mc->cluster_config); free(mc->cluster_config); } free(mc->conffile); free(mc->workdir); free(mc->name); free(mc); c->myconf = NULL; } /* Commit a WAL_CONFIG_MARKER for `sha`/`tag` through the real commit engine so * config_committed_marker returns it (the committed-prefix authority 4.3 reads). */ static void t43_commit_marker(cluster_t *c, uint64_t tag, const uint8_t sha[32]) { uint64_t mlsn = 0; TEST_ASSERT_EQUAL_INT(0, config_append_marker(c, tag, sha, &mlsn)); TEST_ASSERT_EQUAL_INT(0, persist_flush(&c->persist)); TEST_ASSERT_EQUAL_INT(0, commit_submit(&c->commit, mlsn, COMMIT_MODE_WAIT, 1000)); TEST_ASSERT_TRUE( atomic_load_explicit(&c->persist.committed_lsn, memory_order_acquire) >= mlsn); } /* (4.3-a) MATCH activates: a committed marker whose locally-held bytes' sha256 * MATCHES the marker activates the configuration and leaves the member UN-fenced * (R7.8). */ static void test_config_activate_match_activates_and_unfences(void) { char tmp[256]; strcpy(tmp, "/tmp/xce_cr43_match_XXXXXX"); TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); cluster_t *c = t42_make_cluster(tmp); TEST_ASSERT_NOT_NULL(c); t42_make_leader(c, /*term*/3, /*n_peers*/1); TEST_ASSERT_EQUAL_INT(0, commit_engine_init(&c->commit, &c->persist, 1, 8)); repl_set_commit_engine(&c->persist.repl, &c->commit); myconf_t *mc = t43_attach_myconf(c, tmp); /* The locally-held bytes (saved exactly as the config-push handler would). */ size_t blen = 0; uint8_t *bytes = t43_make_config_blob(&blen); uint8_t sha[32]; SHA256(bytes, blen, sha); uint64_t tag; { uint64_t t; memcpy(&t, sha, 8); tag = be64toh(t); } TEST_ASSERT_EQUAL_INT(0, conf_save_config(mc, bytes, blen, sha)); free(bytes); /* Commit a marker that references EXACTLY those bytes. */ t43_commit_marker(c, tag, sha); /* HASH-VERIFIED → activate, fence stays clear. */ TEST_ASSERT_EQUAL_INT(0, config_activate_committed(c)); TEST_ASSERT_EQUAL_INT(0, config_serve_fenced(c)); /* Activation loaded the config into myconf. */ TEST_ASSERT_NOT_NULL(mc->cluster_config); TEST_ASSERT_EQUAL_HEX64(tag, mc->config_version_tag); t43_free_myconf(c); commit_engine_fini(&c->commit); t42_teardown(c); printf("test_config_activate_match_activates_and_unfences: PASS\n"); } /* (4.3-b) MISMATCH fences + retries, NEVER activates: an injected hash mismatch * (locally-held bytes whose sha256 != the committed marker) keeps the member * FENCED (non-serving), ALIVE, and retrying, and does NOT activate the mismatched * config (R7.9). */ static void test_config_activate_mismatch_fences_and_does_not_activate(void) { char tmp[256]; strcpy(tmp, "/tmp/xce_cr43_mismatch_XXXXXX"); TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); cluster_t *c = t42_make_cluster(tmp); TEST_ASSERT_NOT_NULL(c); t42_make_leader(c, /*term*/4, /*n_peers*/1); TEST_ASSERT_EQUAL_INT(0, commit_engine_init(&c->commit, &c->persist, 1, 8)); repl_set_commit_engine(&c->persist.repl, &c->commit); myconf_t *mc = t43_attach_myconf(c, tmp); /* Locally-held bytes are blob A (sha A). */ size_t blenA = 0; uint8_t *bytesA = t43_make_config_blob(&blenA); uint8_t shaA[32]; SHA256(bytesA, blenA, shaA); TEST_ASSERT_EQUAL_INT(0, conf_save_config(mc, bytesA, blenA, shaA)); free(bytesA); /* But the committed marker references a DIFFERENT hash (injected mismatch: * the bytes for this marker have NOT arrived out-of-band yet). */ uint8_t shaB[32]; uint64_t tagB = t42_make_ref(shaB, 0x99); /* deterministic != shaA */ TEST_ASSERT_FALSE(memcmp(shaA, shaB, 32) == 0); t43_commit_marker(c, tagB, shaB); /* MISMATCH → fenced, NOT activated, member stays alive (returns, no abort). */ TEST_ASSERT_EQUAL_INT(-EAGAIN, config_activate_committed(c)); TEST_ASSERT_EQUAL_INT(1, config_serve_fenced(c)); /* The mismatched marker's config was NOT activated: the active config tag is * still the locally-held blob A's tag, NEVER the marker's tagB. */ TEST_ASSERT_NOT_EQUAL(tagB, mc->config_version_tag); /* RETRYING: a second call while still mismatched stays fenced (idempotent, * never activates mismatched) — the member keeps retrying, alive. */ TEST_ASSERT_EQUAL_INT(-EAGAIN, config_activate_committed(c)); TEST_ASSERT_EQUAL_INT(1, config_serve_fenced(c)); TEST_ASSERT_NOT_EQUAL(tagB, mc->config_version_tag); t43_free_myconf(c); commit_engine_fini(&c->commit); t42_teardown(c); printf("test_config_activate_mismatch_fences_and_does_not_activate: PASS\n"); } /* (4.3-c) OBTAINED-OUT-OF-BAND bytes are verified before activate: a member is * fenced while the marker's bytes are absent locally; once the matching bytes * arrive out-of-band (saved exactly as handle_repl_config_push would), the next * activate hash-verifies them against the marker and activates, clearing the * fence (R7.5/R7.8). */ static void test_config_activate_obtained_bytes_verified_before_activate(void) { char tmp[256]; strcpy(tmp, "/tmp/xce_cr43_obtain_XXXXXX"); TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); cluster_t *c = t42_make_cluster(tmp); TEST_ASSERT_NOT_NULL(c); t42_make_leader(c, /*term*/5, /*n_peers*/1); TEST_ASSERT_EQUAL_INT(0, commit_engine_init(&c->commit, &c->persist, 1, 8)); repl_set_commit_engine(&c->persist.repl, &c->commit); myconf_t *mc = t43_attach_myconf(c, tmp); /* The marker's referenced bytes. Compute its hash; commit the marker BEFORE * the bytes are present locally (no conf_save_config yet). */ size_t blen = 0; uint8_t *bytes = t43_make_config_blob(&blen); uint8_t sha[32]; SHA256(bytes, blen, sha); uint64_t tag; { uint64_t t; memcpy(&t, sha, 8); tag = be64toh(t); } t43_commit_marker(c, tag, sha); /* Bytes NOT present locally yet → fenced, not activated (R7.9). */ TEST_ASSERT_EQUAL_INT(-EAGAIN, config_activate_committed(c)); TEST_ASSERT_EQUAL_INT(1, config_serve_fenced(c)); /* Bytes arrive out-of-band: the config-push transport saves them locally * exactly as handle_repl_config_push does (conf_save_config). */ TEST_ASSERT_EQUAL_INT(0, conf_save_config(mc, bytes, blen, sha)); free(bytes); /* Now the obtained bytes hash-verify against the committed marker → activate, * fence clears. */ TEST_ASSERT_EQUAL_INT(0, config_activate_committed(c)); TEST_ASSERT_EQUAL_INT(0, config_serve_fenced(c)); TEST_ASSERT_NOT_NULL(mc->cluster_config); TEST_ASSERT_EQUAL_HEX64(tag, mc->config_version_tag); t43_free_myconf(c); commit_engine_fini(&c->commit); t42_teardown(c); printf("test_config_activate_obtained_bytes_verified_before_activate: PASS\n"); } /* (4.3-d) RECOVERY activates the LATEST committed marker: with two committed * markers, config_activate_committed activates the config referenced by the * LATEST committed marker (the one config_committed_marker derives), obtaining * its bytes from local out-of-band storage (R7.10). */ static void test_config_activate_recovery_uses_latest_committed_marker(void) { char tmp[256]; strcpy(tmp, "/tmp/xce_cr43_recover_XXXXXX"); TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); cluster_t *c = t42_make_cluster(tmp); TEST_ASSERT_NOT_NULL(c); t42_make_leader(c, /*term*/6, /*n_peers*/1); TEST_ASSERT_EQUAL_INT(0, commit_engine_init(&c->commit, &c->persist, 1, 8)); repl_set_commit_engine(&c->persist.repl, &c->commit); myconf_t *mc = t43_attach_myconf(c, tmp); /* Two distinct configs, each committed as a marker in turn; the LATEST wins. */ size_t blen1 = 0, blen2 = 0; uint8_t *b1 = t43_make_config_blob(&blen1); /* Make blob 2 differ so its hash differs from blob 1. */ uint8_t *b2 = t43_make_config_blob(&blen2); TEST_ASSERT_TRUE(blen2 >= 1); b2[blen2 - 1] ^= 0xFF; /* perturb → distinct sha */ uint8_t sha1[32], sha2[32]; SHA256(b1, blen1, sha1); SHA256(b2, blen2, sha2); TEST_ASSERT_FALSE(memcmp(sha1, sha2, 32) == 0); uint64_t tag2; { uint64_t t; memcpy(&t, sha2, 8); tag2 = be64toh(t); } uint64_t tag1; { uint64_t t; memcpy(&t, sha1, 8); tag1 = be64toh(t); } t43_commit_marker(c, tag1, sha1); t43_commit_marker(c, tag2, sha2); /* On recovery, the LOCALLY-held bytes are the LATEST config (blob 2 — what the * out-of-band store holds after the latest push). */ TEST_ASSERT_EQUAL_INT(0, conf_save_config(mc, b2, blen2, sha2)); free(b1); free(b2); /* Activation targets the LATEST committed marker (tag2), hash-verifies the * local bytes, and activates it. */ TEST_ASSERT_EQUAL_INT(0, config_activate_committed(c)); TEST_ASSERT_EQUAL_INT(0, config_serve_fenced(c)); TEST_ASSERT_EQUAL_HEX64(tag2, mc->config_version_tag); t43_free_myconf(c); commit_engine_fini(&c->commit); t42_teardown(c); printf("test_config_activate_recovery_uses_latest_committed_marker: PASS\n"); } /* ── task 4.3 REMEDIATION r1: production integration of the consumer ─────────── * * Round-1 review REJECTED 4.3 for three integration gaps, fixed below: * A — verify-before-activate was BYPASSED on the push path (handle_repl_config_ * push pre-activated via load_primary_config before the gate ran). * B — recovery-activate had NO production call site (boot loaded blindly). * C — the serve-fence had NO production reader (ha_dispatch_gate ignored it). */ /* (4.3-A) PUSH-PATH SOLE ACTIVATION: config_activate_committed is the SOLE * activation point. With a committed marker that references DIFFERENT bytes than * the ones just delivered out-of-band, the push-path sequence (conf_save_config * then config_activate_committed — NO pre-emptive load_primary_config) must NOT * make the pushed bytes live (verify-before-activate is a real GATE, not an * after-the-fact overlay). Regression for Finding A. */ static void test_push_path_mismatch_does_not_preactivate(void) { char tmp[256]; strcpy(tmp, "/tmp/xce_cr43_pushgate_XXXXXX"); TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); cluster_t *c = t42_make_cluster(tmp); TEST_ASSERT_NOT_NULL(c); t42_make_leader(c, /*term*/7, /*n_peers*/1); TEST_ASSERT_EQUAL_INT(0, commit_engine_init(&c->commit, &c->persist, 1, 8)); repl_set_commit_engine(&c->persist.repl, &c->commit); myconf_t *mc = t43_attach_myconf(c, tmp); /* A committed marker references blob B (sha B) — the config the cluster * agreed to, whose bytes have NOT yet arrived here. */ size_t blenB = 0; uint8_t *bytesB = t43_make_config_blob(&blenB); bytesB[blenB - 1] ^= 0x5A; /* perturb → distinct sha */ uint8_t shaB[32]; SHA256(bytesB, blenB, shaB); uint64_t tagB; { uint64_t t; memcpy(&t, shaB, 8); tagB = be64toh(t); } t43_commit_marker(c, tagB, shaB); free(bytesB); /* The push delivers blob A (a DIFFERENT/older config). The push handler now * SAVES the bytes but does NOT pre-activate; activation is gated. */ size_t blenA = 0; uint8_t *bytesA = t43_make_config_blob(&blenA); uint8_t shaA[32]; SHA256(bytesA, blenA, shaA); TEST_ASSERT_FALSE(memcmp(shaA, shaB, 32) == 0); TEST_ASSERT_EQUAL_INT(0, conf_save_config(mc, bytesA, blenA, shaA)); free(bytesA); /* The push-path SOLE activation: gate fences (marker references blob B, not * the just-saved blob A) and does NOT load blob A. */ TEST_ASSERT_EQUAL_INT(-EAGAIN, config_activate_committed(c)); TEST_ASSERT_EQUAL_INT(1, config_serve_fenced(c)); /* Blob A was NOT made live: no decoded config, no version tag for A. */ TEST_ASSERT_NULL(mc->cluster_config); t43_free_myconf(c); commit_engine_fini(&c->commit); t42_teardown(c); printf("test_push_path_mismatch_does_not_preactivate: PASS\n"); } /* (4.3-A/B) NO-MARKER FALLBACK: when there is NO committed marker yet (legacy / * first boot, or the marker has not committed), config_activate_committed FALLS * BACK to the existing load_primary_config behavior — it activates the * locally-held bytes (so config_version_tag is set for the freshness gate and the * push ACK) and does NOT fence. Absent-marker is NOT a mismatch. */ static void test_config_activate_no_marker_falls_back_and_loads(void) { char tmp[256]; strcpy(tmp, "/tmp/xce_cr43_fallback_XXXXXX"); TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); cluster_t *c = t42_make_cluster(tmp); TEST_ASSERT_NOT_NULL(c); t42_make_leader(c, /*term*/8, /*n_peers*/1); TEST_ASSERT_EQUAL_INT(0, commit_engine_init(&c->commit, &c->persist, 1, 8)); repl_set_commit_engine(&c->persist.repl, &c->commit); myconf_t *mc = t43_attach_myconf(c, tmp); /* Local bytes present, but NO committed marker exists. */ size_t blen = 0; uint8_t *bytes = t43_make_config_blob(&blen); uint8_t sha[32]; SHA256(bytes, blen, sha); uint64_t tag; { uint64_t t; memcpy(&t, sha, 8); tag = be64toh(t); } TEST_ASSERT_EQUAL_INT(0, conf_save_config(mc, bytes, blen, sha)); free(bytes); /* No marker → fall back to load_primary_config (blind load), NOT a fence. */ TEST_ASSERT_EQUAL_INT(0, config_activate_committed(c)); TEST_ASSERT_EQUAL_INT(0, config_serve_fenced(c)); TEST_ASSERT_NOT_NULL(mc->cluster_config); TEST_ASSERT_EQUAL_HEX64(tag, mc->config_version_tag); t43_free_myconf(c); commit_engine_fini(&c->commit); t42_teardown(c); printf("test_config_activate_no_marker_falls_back_and_loads: PASS\n"); } /* (4.3-C) FENCE IS ENFORCED by a production reader: while the config serve-fence * is set, ha_dispatch_gate refuses config-dependent state-changing operations on * the PRIMARY (loud non-serving, member stays alive) and a fenced member is * non-promotable (ha_config_promotion_fenced). Read-only/replication ops still * pass so the bytes can arrive out-of-band. Once correct bytes arrive and * re-verify clears the fence, the member serves again. Regression for Finding C. */ static void test_serve_fence_enforced_by_dispatch_gate_and_promotion(void) { char tmp[256]; strcpy(tmp, "/tmp/xce_cr43_fence_XXXXXX"); TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); cluster_t *c = t42_make_cluster(tmp); TEST_ASSERT_NOT_NULL(c); t42_make_leader(c, /*term*/9, /*n_peers*/1); TEST_ASSERT_EQUAL_INT(0, commit_engine_init(&c->commit, &c->persist, 1, 8)); repl_set_commit_engine(&c->persist.repl, &c->commit); myconf_t *mc = t43_attach_myconf(c, tmp); /* The committed config (blob TARGET) whose bytes have not yet arrived. */ size_t tlen = 0; uint8_t *tbytes = t43_make_config_blob(&tlen); uint8_t tsha[32]; SHA256(tbytes, tlen, tsha); uint64_t ttag; { uint64_t t; memcpy(&t, tsha, 8); ttag = be64toh(t); } t43_commit_marker(c, ttag, tsha); /* Stale local bytes (blob A) that do NOT match the committed marker. */ size_t alen = 0; uint8_t *abytes = t43_make_config_blob(&alen); abytes[alen - 1] ^= 0x33; uint8_t asha[32]; SHA256(abytes, alen, asha); TEST_ASSERT_FALSE(memcmp(asha, tsha, 32) == 0); TEST_ASSERT_EQUAL_INT(0, conf_save_config(mc, abytes, alen, asha)); free(abytes); /* Mismatch → fenced. */ TEST_ASSERT_EQUAL_INT(-EAGAIN, config_activate_committed(c)); TEST_ASSERT_EQUAL_INT(1, config_serve_fenced(c)); /* PRODUCTION READER: a fenced PRIMARY refuses config-dependent state-changing * ops (loud non-serving) but stays alive — the gate returns an error, not a * crash/exit. */ TEST_ASSERT_TRUE(ha_dispatch_gate(&c->ha, OPCode_jobsub) < 0); /* Replication / read-only ops still pass so the correct bytes can arrive. */ TEST_ASSERT_EQUAL_INT(0, ha_dispatch_gate(&c->ha, OPCode_replConfigPush)); TEST_ASSERT_EQUAL_INT(0, ha_dispatch_gate(&c->ha, OPCode_get)); /* Non-promotable while fenced. */ TEST_ASSERT_EQUAL_INT(1, ha_config_promotion_fenced(&c->ha)); /* Correct bytes arrive out-of-band (saved as the push handler would). */ TEST_ASSERT_EQUAL_INT(0, conf_save_config(mc, tbytes, tlen, tsha)); free(tbytes); /* Re-verify clears the fence → member serves and is promotable again. */ TEST_ASSERT_EQUAL_INT(0, config_activate_committed(c)); TEST_ASSERT_EQUAL_INT(0, config_serve_fenced(c)); TEST_ASSERT_EQUAL_HEX64(ttag, mc->config_version_tag); TEST_ASSERT_EQUAL_INT(0, ha_dispatch_gate(&c->ha, OPCode_jobsub)); TEST_ASSERT_EQUAL_INT(0, ha_config_promotion_fenced(&c->ha)); t43_free_myconf(c); commit_engine_fini(&c->commit); t42_teardown(c); printf("test_serve_fence_enforced_by_dispatch_gate_and_promotion: PASS\n"); } /* Append a WAL_CONFIG_MARKER for `sha`/`tag` and flush it to the WAL, but do NOT * advance the committed prefix past it — returns its LSN. Models a STANDBY whose * local WAL tip runs ahead of its committed_lsn: on a standby committed_lsn is * driven by the primary's commit-advance (commit_set_raft_commit_index / the * library), NOT by local durability, so a flushed-but-not-yet-committed marker is * the real push-before-commit state. The caller pins committed_lsn back to the * prior value (see the test) so config_committed_marker still derives the OLDER * marker until the NEW one is explicitly committed. */ static uint64_t t43_append_marker_uncommitted(cluster_t *c, uint64_t tag, const uint8_t sha[32]) { uint64_t mlsn = 0; TEST_ASSERT_EQUAL_INT(0, config_append_marker(c, tag, sha, &mlsn)); TEST_ASSERT_EQUAL_INT(0, persist_flush(&c->persist)); return mlsn; } /* (4.3-D / R7.9) COMMIT-ADVANCE RETRY clears a push-before-commit fence: the * config MARKER (WAL) and config BYTES (out-of-band push) arrive on INDEPENDENT * transports. On a 2nd upgrade the NEW bytes are saved BEFORE the NEW marker * enters the committed prefix, so config_committed_marker still returns the OLDER * marker → the freshly-saved NEW bytes hash-MISMATCH it → the member is serve- * FENCED. The NEW marker commits later via the WAL path, but on_commit_advanced * cannot re-activate (no disk I/O on the iomgr ACK thread) and the primary's * config-behind re-push net does not fire (the push already advanced the advertised * tag) — WITHOUT the commit-advance retry trigger the standby would stay * PERMANENTLY fenced. This test reproduces that ordering and asserts the retry * trigger (config_activate_retry_fn — the SAME off-thread entry the production * scheduler enqueues) CLEARS the fence and activates the NEW config once the NEW * marker is committed. Regression for Finding D. * * The single-member commit engine advances committed_lsn to the durable WAL tip, * so to hold the NEW marker out of the committed prefix this test PINS * committed_lsn at the OLD marker's LSN (modeling a standby whose committed prefix * lags its local WAL tip), then advances it past the NEW marker to model the NEW * marker committing via the WAL path. */ static void test_commit_advance_retry_clears_push_before_commit_fence(void) { char tmp[256]; strcpy(tmp, "/tmp/xce_cr43_retry_XXXXXX"); TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); cluster_t *c = t42_make_cluster(tmp); TEST_ASSERT_NOT_NULL(c); t42_make_leader(c, /*term*/10, /*n_peers*/1); TEST_ASSERT_EQUAL_INT(0, commit_engine_init(&c->commit, &c->persist, 1, 8)); repl_set_commit_engine(&c->persist.repl, &c->commit); myconf_t *mc = t43_attach_myconf(c, tmp); /* ── Upgrade 1 (OLD): commit OLD marker + save OLD bytes + activate. The * member is UN-fenced and serving the OLD config. ───────────────────────── */ size_t blenO = 0; uint8_t *bytesO = t43_make_config_blob(&blenO); uint8_t shaO[32]; SHA256(bytesO, blenO, shaO); uint64_t tagO; { uint64_t t; memcpy(&t, shaO, 8); tagO = be64toh(t); } TEST_ASSERT_EQUAL_INT(0, conf_save_config(mc, bytesO, blenO, shaO)); free(bytesO); t43_commit_marker(c, tagO, shaO); TEST_ASSERT_EQUAL_INT(0, config_activate_committed(c)); TEST_ASSERT_EQUAL_INT(0, config_serve_fenced(c)); TEST_ASSERT_EQUAL_HEX64(tagO, mc->config_version_tag); /* The OLD marker is the committed prefix boundary the standby currently sees. */ uint64_t committed_at_old = atomic_load_explicit(&c->persist.committed_lsn, memory_order_acquire); /* ── Upgrade 2 (NEW): the BYTES arrive (and are saved) BEFORE the NEW marker * enters the committed prefix — the push-before-commit ordering. ────────── */ size_t blenN = 0; uint8_t *bytesN = t43_make_config_blob(&blenN); bytesN[blenN - 1] ^= 0xA5; /* perturb → distinct sha */ uint8_t shaN[32]; SHA256(bytesN, blenN, shaN); uint64_t tagN; { uint64_t t; memcpy(&t, shaN, 8); tagN = be64toh(t); } TEST_ASSERT_FALSE(memcmp(shaO, shaN, 32) == 0); /* The NEW marker is APPENDED + flushed to the WAL (local tip advances). */ uint64_t new_mlsn = t43_append_marker_uncommitted(c, tagN, shaN); TEST_ASSERT_TRUE(new_mlsn > committed_at_old); /* The OLD config is currently LIVE (activated in upgrade 1). Capture the live * decoded-config pointer as the activation witness: load_primary_config frees * and reallocates mc->cluster_config, so a CHANGED pointer proves a (re)load * happened. conf_save_config (below) only writes bytes + bumps the advertised * config_version_tag — it does NOT decode/activate — so it must NOT change this * pointer; only a successful config_activate_committed does. */ cluster_config_t *live_before = mc->cluster_config; TEST_ASSERT_NOT_NULL(live_before); /* The NEW bytes are saved out-of-band (exactly as handle_repl_config_push * does). Note conf_save_config in this single-member fixture re-evaluates the * commit barrier, which would advance committed_lsn to the durable WAL tip, and * it bumps the advertised config_version_tag to tagN (this is the very reason * the primary's config-behind re-push net does not re-fire — Finding D); we * save FIRST, then PIN committed_lsn back below the NEW marker to model a * standby whose committed prefix (driven by the primary's commit-advance, not * local durability) has NOT yet advanced past the NEW marker. */ TEST_ASSERT_EQUAL_INT(0, conf_save_config(mc, bytesN, blenN, shaN)); atomic_store_explicit(&c->persist.committed_lsn, committed_at_old, memory_order_release); /* Activate now: config_committed_marker still returns the OLDER marker (tagO), * so the freshly-saved NEW bytes MISMATCH it → serve-FENCED, NEW config NOT * ACTIVATED. The live decoded config is UNCHANGED (still the OLD one) — the * mismatch gate never loaded the NEW bytes. This is the permanent-fence state * the retry trigger must rescue. */ TEST_ASSERT_EQUAL_INT(-EAGAIN, config_activate_committed(c)); TEST_ASSERT_EQUAL_INT(1, config_serve_fenced(c)); TEST_ASSERT_TRUE(mc->cluster_config == live_before); /* NEW not activated */ /* ── The NEW marker COMMITS later via the WAL path (the standby's committed * prefix advances past it, as the primary's commit-advance heartbeat would * drive). config_committed_marker now derives the NEW marker. ───────────── */ atomic_store_explicit(&c->persist.committed_lsn, new_mlsn, memory_order_release); /* DRIVE THE RETRY TRIGGER: the SAME off-thread activation entry the production * commit-advance path schedules (ha_cb_commit_advanced_stub → * config_schedule_activate_retry → delaymgr → config_activate_retry_fn). In * production the delaymgr worker runs this OFF the iomgr ACK thread; here we * invoke the entry directly (the fixture's delaymgr is not started). Now * config_committed_marker returns the NEW marker and the saved NEW bytes * hash-verify → the fence CLEARS and the NEW config ACTIVATES (cluster_config * is reloaded — a fresh pointer). */ TEST_ASSERT_EQUAL_INT(0, config_activate_retry_fn(c)); TEST_ASSERT_EQUAL_INT(0, config_serve_fenced(c)); /* fence CLEARED */ TEST_ASSERT_EQUAL_HEX64(tagN, mc->config_version_tag); /* NEW now live */ TEST_ASSERT_NOT_NULL(mc->cluster_config); TEST_ASSERT_TRUE(mc->cluster_config != live_before); /* (re)loaded NEW */ free(bytesN); t43_free_myconf(c); commit_engine_fini(&c->commit); t42_teardown(c); printf("test_commit_advance_retry_clears_push_before_commit_fence: PASS\n"); } /* ── wait_all_with_timeout (XCE_TESTING shim) ───── */ #ifdef XCE_TESTING /* Thread args for timeout tests */ typedef struct { atomic_int *stop; /* poll this; exit when set to allow pthread_join to return */ push_ctx_t *ctx; } timeout_test_arg_t; /* Polls stop flag at 5ms intervals; exits when stop is set WITHOUT setting acked. * This simulates a synchronous push that respects the done signal. */ static void *slow_push_thread(void *arg) { timeout_test_arg_t *a = (timeout_test_arg_t *)arg; while (!atomic_load(a->stop)) usleep(5000); /* Thread exits without setting acked — deadline fired before completion */ a->ctx->done = 1; return NULL; } static void *fast_push_thread(void *arg) { push_ctx_t *ctx = (push_ctx_t *)arg; ctx->acked = 1; ctx->done = 1; return NULL; } static void test_wait_all_with_timeout_slow_threads(void) { /* 2 threads that poll the done flag and exit without acking. * Timeout is 100ms. Expect: acked==0, done==1, returns within ~200ms. */ const int N = 2; pthread_t threads[2]; push_ctx_t ctxs[2]; timeout_test_arg_t args[2]; atomic_int done; memset(ctxs, 0, sizeof(ctxs)); atomic_init(&done, 0); for (int i = 0; i < N; i++) { ctxs[i].started = 1; args[i].stop = &done; args[i].ctx = &ctxs[i]; pthread_create(&threads[i], NULL, slow_push_thread, &args[i]); } struct timespec t0, t1; clock_gettime(CLOCK_MONOTONIC, &t0); int acked = wait_all_with_timeout_for_test(threads, ctxs, N, 100, &done); clock_gettime(CLOCK_MONOTONIC, &t1); long elapsed_ms = (t1.tv_sec - t0.tv_sec) * 1000 + (t1.tv_nsec - t0.tv_nsec) / 1000000; /* Threads did not ack within the 100ms timeout */ TEST_ASSERT_EQUAL_INT(0, acked); /* done flag must be set by wait_all_with_timeout */ TEST_ASSERT_EQUAL_INT(1, atomic_load(&done)); /* Total elapsed should be close to the 100ms timeout + small join overhead */ TEST_ASSERT_LESS_THAN(400, (int)elapsed_ms); printf("test_wait_all_with_timeout_slow_threads: PASS (elapsed=%ldms acked=%d)\n", elapsed_ms, acked); } static void test_wait_all_with_timeout_fast_threads(void) { /* 3 threads that complete immediately; timeout is 200ms. * Expect: all 3 acked, returns quickly (well under 200ms). */ const int N = 3; pthread_t threads[3]; push_ctx_t ctxs[3]; atomic_int done; memset(ctxs, 0, sizeof(ctxs)); atomic_init(&done, 0); for (int i = 0; i < N; i++) { ctxs[i].started = 1; pthread_create(&threads[i], NULL, fast_push_thread, &ctxs[i]); } struct timespec t0, t1; clock_gettime(CLOCK_MONOTONIC, &t0); int acked = wait_all_with_timeout_for_test(threads, ctxs, N, 200, &done); clock_gettime(CLOCK_MONOTONIC, &t1); long elapsed_ms = (t1.tv_sec - t0.tv_sec) * 1000 + (t1.tv_nsec - t0.tv_nsec) / 1000000; TEST_ASSERT_EQUAL_INT(N, acked); TEST_ASSERT_EQUAL_INT(1, atomic_load(&done)); /* Should have returned long before the 200ms timeout */ TEST_ASSERT_LESS_THAN(200, (int)elapsed_ms); printf("test_wait_all_with_timeout_fast_threads: PASS (elapsed=%ldms acked=%d)\n", elapsed_ms, acked); } #endif /* XCE_TESTING */ /* ── raw capnp shims (Task A) ───────────────────── */ /* Test 1: xce_replconfig_req_ssl encodes a request that round-trips * through the capnp decoder and preserves the version_tag. */ static void test_replconfig_req_ssl_encode(void) { int sv[2]; TEST_ASSERT_EQUAL_INT(0, socketpair(AF_UNIX, SOCK_STREAM, 0, sv)); /* Wrap the writer end of the socketpair in a bare BIO/SSL pair so we * can call SSL_write without doing a TLS handshake. Instead, we use * a memory BIO and let xce_replconfig_req_ssl write into it. */ SSL_library_init(); SSL_CTX *ctx = SSL_CTX_new(TLS_client_method()); TEST_ASSERT_NOT_NULL(ctx); SSL *ssl = SSL_new(ctx); TEST_ASSERT_NOT_NULL(ssl); /* Attach the writer fd as the underlying BIO. We don't perform a * handshake; SSL_write will just SYSCALL through to the fd. This * works because BIO_set_nbio is off and OpenSSL is willing to * write plaintext until handshake completes — except newer OpenSSL * refuses. Use a simpler approach: do the encode + raw fd write. */ /* Cleaner approach: bypass SSL wrapper and just exercise the encode * by re-using the public helper but with our own raw fd write. * We test the encoding path by invoking xce_replconfig_req_ssl * with a properly-constructed SSL only if available, otherwise * fall back to capnp encode verification via the public function * called against a duplicated socket pair. */ SSL_free(ssl); SSL_CTX_free(ctx); /* Use a pipe-based "raw" verification: capnp encode the request * ourselves the same way the function does, then decode and check * version_tag. This validates the wire layout without requiring * a full SSL handshake in unit tests. */ { struct capn c; struct capn_segment *cs; struct Request rq; struct ReplConfigRequest req_struct; ReplConfigRequest_ptr req_ptr; Request_ptr rp; capn_init_malloc(&c); cs = capn_root(&c).seg; memset(&rq, 0, sizeof(rq)); memset(&req_struct, 0, sizeof(req_struct)); req_struct.versionTag = 0xdeadbeefcafef00dULL; req_ptr = new_ReplConfigRequest(cs); write_ReplConfigRequest(&req_struct, req_ptr); rq.opcode = OPCode_replConfig; rq.body_which = Request_body_replconfigrequestmsg; rq.body.replconfigrequestmsg = req_ptr; encode_PeerInfo_ptr(cs, &rq.from, NULL); rp = new_Request(cs); write_Request(&rq, rp); capn_setp(capn_root(&c), 0, rp.p); int64_t sz = capn_size(&c); TEST_ASSERT_TRUE(sz > 0); uint8_t *buf = (uint8_t *)malloc((size_t)sz); TEST_ASSERT_NOT_NULL(buf); capn_write_mem(&c, buf, (size_t)sz, 0); capn_free(&c); /* Decode and verify */ struct capn d; TEST_ASSERT_EQUAL_INT(0, capn_init_mem(&d, buf, (size_t)sz, 0)); Request_ptr dp; dp.p = capn_getp(capn_root(&d), 0, 1); struct Request drq; read_Request(&drq, dp); TEST_ASSERT_EQUAL_INT(OPCode_replConfig, drq.opcode); TEST_ASSERT_EQUAL_INT(Request_body_replconfigrequestmsg, drq.body_which); ReplConfigRequest_ptr drp = drq.body.replconfigrequestmsg; struct ReplConfigRequest drs; read_ReplConfigRequest(&drs, drp); TEST_ASSERT_EQUAL_UINT64(0xdeadbeefcafef00dULL, drs.versionTag); capn_free(&d); free(buf); } close(sv[0]); close(sv[1]); printf("test_replconfig_req_ssl_encode: PASS\n"); } /* Test 2: xce_replconfig_reply_raw — verify NULL endpoint returns * -EINVAL and that the encode path with real bytes preserves the * sha256 and bytes payloads through a round-trip decoder. */ static void test_replconfig_reply_raw_encode(void) { /* NULL endpoint guard */ { uint8_t sha[32]; uint8_t cfg[64]; memset(sha, 0xaa, sizeof(sha)); memset(cfg, 0xbb, sizeof(cfg)); TEST_ASSERT_EQUAL_INT(-EINVAL, xce_replconfig_reply_raw(NULL, NULL, 1, 0x42, sha, sizeof(sha), cfg, sizeof(cfg))); } /* Encode-path test: directly build the same capnp structure that * xce_replconfig_reply_raw builds, then walk down Request → * RequestReply → ReplConfigReply and read sha256/bytes via raw * capn_list8 access. */ { struct capn c; struct capn_segment *cs; struct Request rq; struct RequestReply rr; struct ReplConfigReply reply_struct; ReplConfigReply_ptr reply_ptr; RequestReply_ptr rr_ptr; Request_ptr rp; const uint64_t tag = 0xfeedfacecafebabeULL; const uint64_t seq = 99; uint8_t sha256[32]; uint8_t config[128]; for (size_t i = 0; i < sizeof(sha256); i++) sha256[i] = (uint8_t)(0x10 + i); for (size_t i = 0; i < sizeof(config); i++) config[i] = (uint8_t)(i * 3); capn_init_malloc(&c); cs = capn_root(&c).seg; memset(&rq, 0, sizeof(rq)); memset(&rr, 0, sizeof(rr)); memset(&reply_struct, 0, sizeof(reply_struct)); reply_struct.upToDate = 0; reply_struct.versionTag = tag; { capn_list8 sl = capn_new_list8(cs, (int)sizeof(sha256)); capn_setv8(sl, 0, sha256, (int)sizeof(sha256)); reply_struct.sha256.p = sl.p; } { capn_list8 bl = capn_new_list8(cs, (int)sizeof(config)); capn_setv8(bl, 0, config, (int)sizeof(config)); reply_struct.bytes.p = bl.p; } reply_ptr = new_ReplConfigReply(cs); write_ReplConfigReply(&reply_struct, reply_ptr); rr.seq = seq; rr.opcode = OPCode_replConfig; rr.status = 0; rr.data_which = RequestReply_data_replconfigreplymsg; rr.data.replconfigreplymsg = reply_ptr; rr_ptr = new_RequestReply(cs); write_RequestReply(&rr, rr_ptr); rq.opcode = OPCode_reply; rq.seq = seq; rq.body_which = Request_body_replymsg; rq.body.replymsg = rr_ptr; encode_PeerInfo_ptr(cs, &rq.from, NULL); rp = new_Request(cs); write_Request(&rq, rp); capn_setp(capn_root(&c), 0, rp.p); int64_t sz = capn_size(&c); uint8_t *buf = (uint8_t *)malloc((size_t)sz); capn_write_mem(&c, buf, (size_t)sz, 0); capn_free(&c); /* Decode */ struct capn d; capn_init_mem(&d, buf, (size_t)sz, 0); Request_ptr dp; dp.p = capn_getp(capn_root(&d), 0, 1); struct Request drq; read_Request(&drq, dp); TEST_ASSERT_EQUAL_INT(Request_body_replymsg, drq.body_which); struct RequestReply drr; read_RequestReply(&drr, drq.body.replymsg); TEST_ASSERT_EQUAL_INT(RequestReply_data_replconfigreplymsg, drr.data_which); struct ReplConfigReply drs; read_ReplConfigReply(&drs, drr.data.replconfigreplymsg); TEST_ASSERT_EQUAL_INT(0, drs.upToDate); TEST_ASSERT_EQUAL_UINT64(tag, drs.versionTag); capn_list8 ds; ds.p = drs.sha256.p; capn_list8 db; db.p = drs.bytes.p; TEST_ASSERT_EQUAL_INT(32, capn_len(ds)); TEST_ASSERT_EQUAL_INT(128, capn_len(db)); uint8_t got_sha[32]; uint8_t got_cfg[128]; TEST_ASSERT_EQUAL_INT(32, capn_getv8(ds, 0, got_sha, 32)); TEST_ASSERT_EQUAL_INT(128, capn_getv8(db, 0, got_cfg, 128)); TEST_ASSERT_EQUAL_INT(0, memcmp(got_sha, sha256, 32)); TEST_ASSERT_EQUAL_INT(0, memcmp(got_cfg, config, 128)); capn_free(&d); free(buf); } printf("test_replconfig_reply_raw_encode: PASS\n"); } /* Test 3: xce_upgrade_config_req_extract_raw — build a raw * OP_upgradeConfig packet using the same list8 bypass pattern, then * parse it back and verify sha256 + bytes are recovered intact. */ static void test_upgrade_config_req_extract_raw(void) { const uint8_t sha256[32] = { 0x01,0x02,0x03,0x04,0x05,0x06,0x07,0x08, 0x09,0x0a,0x0b,0x0c,0x0d,0x0e,0x0f,0x10, 0x11,0x12,0x13,0x14,0x15,0x16,0x17,0x18, 0x19,0x1a,0x1b,0x1c,0x1d,0x1e,0x1f,0x20 }; const uint8_t cfg[] = "fake-upgrade-config-blob-payload-data"; const size_t cfglen = sizeof(cfg) - 1; /* Build a raw OP_upgradeConfig request packet */ struct capn c; struct capn_segment *cs; struct Request rq; struct UpgradeConfigRequest u_struct; UpgradeConfigRequest_ptr u_ptr; Request_ptr rp; capn_init_malloc(&c); cs = capn_root(&c).seg; memset(&rq, 0, sizeof(rq)); memset(&u_struct, 0, sizeof(u_struct)); { capn_list8 sl = capn_new_list8(cs, (int)sizeof(sha256)); capn_setv8(sl, 0, sha256, (int)sizeof(sha256)); u_struct.sha256.p = sl.p; } { capn_list8 bl = capn_new_list8(cs, (int)cfglen); capn_setv8(bl, 0, cfg, (int)cfglen); u_struct.bytes.p = bl.p; } u_ptr = new_UpgradeConfigRequest(cs); write_UpgradeConfigRequest(&u_struct, u_ptr); rq.opcode = OPCode_upgradeConfig; rq.body_which = Request_body_upgradeconfigrequestmsg; rq.body.upgradeconfigrequestmsg = u_ptr; encode_PeerInfo_ptr(cs, &rq.from, NULL); rp = new_Request(cs); write_Request(&rq, rp); capn_setp(capn_root(&c), 0, rp.p); int64_t sz = capn_size(&c); uint8_t *buf = (uint8_t *)malloc((size_t)sz); capn_write_mem(&c, buf, (size_t)sz, 0); capn_free(&c); /* Extract via the function under test */ uint8_t *out_bytes = NULL; uint8_t *out_sha256 = NULL; size_t out_len = 0; int rc = xce_upgrade_config_req_extract_raw(buf, (size_t)sz, &out_bytes, &out_len, &out_sha256); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_NOT_NULL(out_sha256); TEST_ASSERT_NOT_NULL(out_bytes); TEST_ASSERT_EQUAL_INT((int)cfglen, (int)out_len); TEST_ASSERT_EQUAL_INT(0, memcmp(out_sha256, sha256, 32)); TEST_ASSERT_EQUAL_INT(0, memcmp(out_bytes, cfg, cfglen)); free(out_bytes); free(out_sha256); free(buf); /* xce_upgrade_config_req_extract: returns -EINVAL on a zeroed request * (body_which is 0, not Request_body_upgradeconfigrequestmsg). */ { xce_request_t dummy_req; memset(&dummy_req, 0, sizeof(dummy_req)); uint8_t *b = NULL, *s = NULL; size_t l = 0; TEST_ASSERT_EQUAL_INT(-EINVAL, xce_upgrade_config_req_extract(&dummy_req, &b, &l, &s)); } printf("test_upgrade_config_req_extract_raw: PASS\n"); } /* ── Task 2.1: OPCode_walCatchupEntry handler ────────────────────────────── */ #ifdef XCE_TESTING /* Helpers shared by Task 2.1 tests: create/destroy a minimal persist_t. */ static persist_t *make_test_persist_catchup(char *tmpdir_out, size_t sz) { 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; /* no real fsync in unit tests */ snprintf(tmpdir_out, sz, "/tmp/xce_catchup_XXXXXX"); if (!mkdtemp(tmpdir_out)) { fprintf(stderr, "mkdtemp failed\n"); free(p); exit(1); } persist_set_workdir(p, tmpdir_out, "ctrl-catchup"); persist_init(p, &cfg, NULL); persist_start(p); return p; } static void destroy_test_persist_catchup(persist_t *p) { persist_stop(p); persist_fini(p); free(p); } /* bug-850/851 contiguity guard (persist.c:839): persist_wal_write_replicated * REFUSES a forward gap (entry.lsn > current_lsn → -EAGAIN) so a standby never * creates a WAL hole. A fresh persist has current_lsn == 1 (LSN 0 reserved, * persist.c:542). The catch-up tests below choose their own non-1 starting LSNs * (42, 10, 20, …); seed current_lsn to the first applied entry's LSN so the * first entry is contiguous (entry.lsn == current_lsn) and the guard accepts * the in-order stream — mirroring the real catch-up streamer, which streams * contiguously from current_lsn. current_lsn is a plain lock-protected field; * persist.c sets it directly under p->lock (e.g. persist.c:542,715), so do the * same here. */ static void seed_persist_current_lsn(persist_t *p, uint64_t next_lsn) { pthread_mutex_lock(&p->lock); p->current_lsn = next_lsn; pthread_mutex_unlock(&p->lock); } /* Test 2.1-A: successful single WAL entry advances applied_lsn */ static void test_wal_catchup_entry_advances_applied_lsn(void) { char tmpdir[256]; persist_t *p = make_test_persist_catchup(tmpdir, sizeof(tmpdir)); ha_state_t ha; memset(&ha, 0, sizeof(ha)); ha_config_t cfg; memset(&cfg, 0, sizeof(cfg)); cfg.heartbeat_timeout_ms = 10000; cfg.election_timeout_ms = 1500; ha_init(&ha, &cfg, 0); /* applied_lsn starts at 0 */ TEST_ASSERT_EQUAL_UINT64(0, (uint64_t)atomic_load_explicit(&ha.applied_lsn, memory_order_acquire)); /* Seed current_lsn so the entry at LSN 42 is contiguous (bug-850 guard). */ seed_persist_current_lsn(p, 42); /* Build a single-entry WAL catch-up with LSN=42, no payload */ wal_catchup_entry_t entry; memset(&entry, 0, sizeof(entry)); entry.lsn = 42; entry.entry_type = (uint8_t)WAL_JOB_UPDATE; entry.payload = NULL; /* payload_len=0; apply_wal_entry handles NULL */ entry.n_entries = 0; entry.entries = NULL; /* GREEN will implement ha_apply_wal_catchup_entry() */ int rc = ha_apply_wal_catchup_entry(&ha, p, &entry, NULL, NULL); TEST_ASSERT_EQUAL_INT(0, rc); /* bug-991: applied_lsn is the LAST CONTIGUOUSLY-HELD LSN, so a single applied * entry at LSN 42 advances applied_lsn to 42 (NOT 43). The +1 over-advertised * the standby's watermark and made the leader's catch-up streamer skip the * real next-needed LSN. */ TEST_ASSERT_EQUAL_UINT64(42, (uint64_t)atomic_load_explicit(&ha.applied_lsn, memory_order_acquire)); ha_fini(&ha); destroy_test_persist_catchup(p); char cmd[512]; snprintf(cmd, sizeof(cmd), "rm -rf %s", tmpdir); system(cmd); printf("test_wal_catchup_entry_advances_applied_lsn: PASS\n"); } /* Test 2.1-B: apply_wal_entry failure (NULL persist = -EINVAL) is logged and * skipped — applied_lsn must NOT advance, handler must not crash. * Req 3.3: corrupt/error entries are logged and skipped without crashing. */ static void test_wal_catchup_entry_error_skipped_no_crash(void) { ha_state_t ha; memset(&ha, 0, sizeof(ha)); ha_config_t cfg; memset(&cfg, 0, sizeof(cfg)); cfg.heartbeat_timeout_ms = 10000; cfg.election_timeout_ms = 1500; ha_init(&ha, &cfg, 0); atomic_store_explicit(&ha.applied_lsn, 10, memory_order_release); /* Pass NULL persist — apply_wal_entry returns -EINVAL for NULL p */ wal_catchup_entry_t entry; memset(&entry, 0, sizeof(entry)); entry.lsn = 99; entry.entry_type = (uint8_t)WAL_JOB_UPDATE; entry.payload = NULL; entry.n_entries = 0; entry.entries = NULL; /* Handler must NOT crash; must log ERROR; applied_lsn must remain at 10 */ int rc = ha_apply_wal_catchup_entry(&ha, /*persist=*/NULL, &entry, NULL, NULL); /* rc is non-zero (error propagated from apply_wal_entry(-EINVAL)) */ TEST_ASSERT_TRUE(rc != 0); /* applied_lsn must not advance on failure */ TEST_ASSERT_EQUAL_UINT64(10, (uint64_t)atomic_load_explicit(&ha.applied_lsn, memory_order_acquire)); ha_fini(&ha); printf("test_wal_catchup_entry_error_skipped_no_crash: PASS\n"); } /* Test 2.1-C: NULL entry is handled gracefully (null guard) */ static void test_wal_catchup_entry_null_entry_no_crash(void) { ha_state_t ha; memset(&ha, 0, sizeof(ha)); ha_config_t cfg; memset(&cfg, 0, sizeof(cfg)); cfg.heartbeat_timeout_ms = 10000; cfg.election_timeout_ms = 1500; ha_init(&ha, &cfg, 0); /* NULL entry — must not crash, must return error */ int rc = ha_apply_wal_catchup_entry(&ha, NULL, NULL, NULL, NULL); TEST_ASSERT_TRUE(rc != 0); ha_fini(&ha); printf("test_wal_catchup_entry_null_entry_no_crash: PASS\n"); } /* ── Task 2.2: batch WAL catch-up entries on standby ─────────────────────── */ /* Test 2.2-A: batch of 3 entries — applied_lsn advances to last+1. * Requirements 5.3, 5.5: apply each in order, advance after full batch. */ static void test_wal_catchup_batch_advances_applied_lsn(void) { char tmpdir[256]; persist_t *p = make_test_persist_catchup(tmpdir, sizeof(tmpdir)); ha_state_t ha; memset(&ha, 0, sizeof(ha)); ha_config_t cfg; memset(&cfg, 0, sizeof(cfg)); cfg.heartbeat_timeout_ms = 10000; cfg.election_timeout_ms = 1500; ha_init(&ha, &cfg, 0); /* applied_lsn starts at 0 */ TEST_ASSERT_EQUAL_UINT64(0, (uint64_t)atomic_load_explicit(&ha.applied_lsn, memory_order_acquire)); /* Seed current_lsn so the batch [10,11,12] is contiguous (bug-850 guard). */ seed_persist_current_lsn(p, 10); /* Build 3 sub-entries at LSNs 10, 11, 12 */ wal_catchup_entry_t e0, e1, e2; memset(&e0, 0, sizeof(e0)); memset(&e1, 0, sizeof(e1)); memset(&e2, 0, sizeof(e2)); e0.lsn = 10; e0.entry_type = (uint8_t)WAL_JOB_UPDATE; e1.lsn = 11; e1.entry_type = (uint8_t)WAL_JOB_UPDATE; e2.lsn = 12; e2.entry_type = (uint8_t)WAL_JOB_UPDATE; wal_catchup_entry_t *batch_entries[3] = { &e0, &e1, &e2 }; /* Outer entry carries the batch; lsn/entry_type ignored when n_entries > 0 */ wal_catchup_entry_t outer; memset(&outer, 0, sizeof(outer)); outer.n_entries = 3; outer.entries = batch_entries; int rc = ha_apply_wal_catchup_entry(&ha, p, &outer, NULL, NULL); TEST_ASSERT_EQUAL_INT(0, rc); /* bug-991: applied_lsn = last contiguously-held LSN = last_entry->lsn = 12 * (NOT 13). */ TEST_ASSERT_EQUAL_UINT64(12, (uint64_t)atomic_load_explicit(&ha.applied_lsn, memory_order_acquire)); ha_fini(&ha); destroy_test_persist_catchup(p); char cmd[512]; snprintf(cmd, sizeof(cmd), "rm -rf %s", tmpdir); system(cmd); printf("test_wal_catchup_batch_advances_applied_lsn: PASS\n"); } /* Test 2.2-B: partial batch (2 of max 3) — applied_lsn advances correctly. * Requirement 5.4: partial batch (fewer than max) sent and applied without wait. */ static void test_wal_catchup_partial_batch(void) { char tmpdir[256]; persist_t *p = make_test_persist_catchup(tmpdir, sizeof(tmpdir)); ha_state_t ha; memset(&ha, 0, sizeof(ha)); ha_config_t cfg; memset(&cfg, 0, sizeof(cfg)); cfg.heartbeat_timeout_ms = 10000; cfg.election_timeout_ms = 1500; ha_init(&ha, &cfg, 0); /* Pre-set applied_lsn to 5 to verify monotonic advance */ atomic_store_explicit(&ha.applied_lsn, 5, memory_order_release); /* Seed current_lsn so the batch [20,21] is contiguous (bug-850 guard). */ seed_persist_current_lsn(p, 20); wal_catchup_entry_t e0, e1; memset(&e0, 0, sizeof(e0)); memset(&e1, 0, sizeof(e1)); e0.lsn = 20; e0.entry_type = (uint8_t)WAL_JOB_UPDATE; e1.lsn = 21; e1.entry_type = (uint8_t)WAL_JOB_UPDATE; wal_catchup_entry_t *batch_entries[2] = { &e0, &e1 }; wal_catchup_entry_t outer; memset(&outer, 0, sizeof(outer)); outer.n_entries = 2; outer.entries = batch_entries; int rc = ha_apply_wal_catchup_entry(&ha, p, &outer, NULL, NULL); TEST_ASSERT_EQUAL_INT(0, rc); /* bug-991: applied_lsn = last contiguously-held LSN = 21 (NOT 22). Monotonic: * advances from the pre-set 5 to 21. */ TEST_ASSERT_EQUAL_UINT64(21, (uint64_t)atomic_load_explicit(&ha.applied_lsn, memory_order_acquire)); ha_fini(&ha); destroy_test_persist_catchup(p); char cmd[512]; snprintf(cmd, sizeof(cmd), "rm -rf %s", tmpdir); system(cmd); printf("test_wal_catchup_partial_batch: PASS\n"); } /* Test 2.2-C: batch with one error entry — error is logged and skipped, * remaining entries are still applied, applied_lsn advances for the good ones. * Requirement 3.3/5.5: error in batch → skip that entry, continue with rest. */ static void test_wal_catchup_batch_error_entry_skipped(void) { char tmpdir[256]; persist_t *p = make_test_persist_catchup(tmpdir, sizeof(tmpdir)); ha_state_t ha; memset(&ha, 0, sizeof(ha)); ha_config_t cfg; memset(&cfg, 0, sizeof(cfg)); cfg.heartbeat_timeout_ms = 10000; cfg.election_timeout_ms = 1500; ha_init(&ha, &cfg, 0); /* Seed current_lsn so the first good entry at LSN 30 is contiguous (bug-850 * guard). The NULL sub-entry is skipped (no write, current_lsn stays at 31); * the test only requires applied_lsn >= 30 (the first good entry held). */ seed_persist_current_lsn(p, 30); /* Good entry at LSN 30, then a NULL pointer (simulates corrupt entry), * then good entry at LSN 32. */ wal_catchup_entry_t e0, e2; memset(&e0, 0, sizeof(e0)); memset(&e2, 0, sizeof(e2)); e0.lsn = 30; e0.entry_type = (uint8_t)WAL_JOB_UPDATE; e2.lsn = 32; e2.entry_type = (uint8_t)WAL_JOB_UPDATE; /* NULL in the middle represents a missing/corrupt sub-entry */ wal_catchup_entry_t *batch_entries[3] = { &e0, NULL, &e2 }; wal_catchup_entry_t outer; memset(&outer, 0, sizeof(outer)); outer.n_entries = 3; outer.entries = batch_entries; /* Handler must not crash; applied_lsn advances for successful entries */ int rc = ha_apply_wal_catchup_entry(&ha, p, &outer, NULL, NULL); /* Overall rc == 0: outer batch call succeeds even with inner skips */ TEST_ASSERT_EQUAL_INT(0, rc); /* bug-991: applied_lsn = last contiguously-held LSN. e0 (LSN 30) applies, * the NULL sub-entry is skipped, e2 (LSN 32) applies → last-held is 32 (or at * least 30 if e2 were dropped). Must reach at least 30 (e0 held). */ uint64_t final_lsn = (uint64_t)atomic_load_explicit(&ha.applied_lsn, memory_order_acquire); TEST_ASSERT_TRUE(final_lsn >= 30); ha_fini(&ha); destroy_test_persist_catchup(p); char cmd[512]; snprintf(cmd, sizeof(cmd), "rm -rf %s", tmpdir); system(cmd); printf("test_wal_catchup_batch_error_entry_skipped: PASS\n"); } /* Test 2.2-D: n_entries==0 falls through to single-entry path (backward compat) */ static void test_wal_catchup_batch_zero_entries_fallback(void) { char tmpdir[256]; persist_t *p = make_test_persist_catchup(tmpdir, sizeof(tmpdir)); ha_state_t ha; memset(&ha, 0, sizeof(ha)); ha_config_t cfg; memset(&cfg, 0, sizeof(cfg)); cfg.heartbeat_timeout_ms = 10000; cfg.election_timeout_ms = 1500; ha_init(&ha, &cfg, 0); /* Seed current_lsn so the fallback single entry at LSN 50 is contiguous * (bug-850 guard). */ seed_persist_current_lsn(p, 50); /* n_entries == 0 → single-entry path from task 2.1 */ wal_catchup_entry_t entry; memset(&entry, 0, sizeof(entry)); entry.lsn = 50; entry.entry_type = (uint8_t)WAL_JOB_UPDATE; entry.n_entries = 0; entry.entries = NULL; int rc = ha_apply_wal_catchup_entry(&ha, p, &entry, NULL, NULL); TEST_ASSERT_EQUAL_INT(0, rc); /* bug-991: single-entry path advances applied_lsn to the last-held LSN = 50 * (NOT 51). */ TEST_ASSERT_EQUAL_UINT64(50, (uint64_t)atomic_load_explicit(&ha.applied_lsn, memory_order_acquire)); ha_fini(&ha); destroy_test_persist_catchup(p); char cmd[512]; snprintf(cmd, sizeof(cmd), "rm -rf %s", tmpdir); system(cmd); printf("test_wal_catchup_batch_zero_entries_fallback: PASS\n"); } /* ── Task 6.2: additional WAL catch-up unit tests ───────────────────────── */ /* Test 6.2-A: batch of exactly 256 entries (the maximum batch size from the * sender side, Requirement 5.2) — all applied, applied_lsn advances by 256. * Tests: "apply batch of 256" from task 6.2 spec. * Requirements 3.1, 3.5, 5.3, 5.5. */ static void test_wal_catchup_batch_256_entries(void) { char tmpdir[256]; persist_t *p = make_test_persist_catchup(tmpdir, sizeof(tmpdir)); ha_state_t ha; memset(&ha, 0, sizeof(ha)); ha_config_t cfg; memset(&cfg, 0, sizeof(cfg)); cfg.heartbeat_timeout_ms = 10000; cfg.election_timeout_ms = 1500; ha_init(&ha, &cfg, 0); /* Seed current_lsn so the consecutive batch starting at LSN 1000 is * contiguous (bug-850 guard). */ seed_persist_current_lsn(p, 1000); /* Allocate 256 sub-entries at consecutive LSNs starting at 1000 */ const int N = 256; wal_catchup_entry_t entries[N]; wal_catchup_entry_t *ptrs[N]; for (int i = 0; i < N; i++) { memset(&entries[i], 0, sizeof(entries[i])); entries[i].lsn = (uint64_t)(1000 + i); entries[i].entry_type = (uint8_t)WAL_JOB_UPDATE; entries[i].n_entries = 0; entries[i].entries = NULL; ptrs[i] = &entries[i]; } wal_catchup_entry_t outer; memset(&outer, 0, sizeof(outer)); outer.n_entries = N; outer.entries = ptrs; /* applied_lsn starts at 0 */ TEST_ASSERT_EQUAL_UINT64(0, (uint64_t)atomic_load_explicit(&ha.applied_lsn, memory_order_acquire)); int rc = ha_apply_wal_catchup_entry(&ha, p, &outer, NULL, NULL); TEST_ASSERT_EQUAL_INT(0, rc); /* bug-991: applied_lsn = last contiguously-held LSN = last_entry->lsn = * 1000+255 = 1255 (NOT 1256). */ TEST_ASSERT_EQUAL_UINT64(1255, (uint64_t)atomic_load_explicit(&ha.applied_lsn, memory_order_acquire)); ha_fini(&ha); destroy_test_persist_catchup(p); char cmd[512]; snprintf(cmd, sizeof(cmd), "rm -rf %s", tmpdir); system(cmd); printf("test_wal_catchup_batch_256_entries: PASS\n"); } /* Test 6.2-B: applied_lsn advances only AFTER the full batch is written, not * during mid-batch entry application. This verifies Req 5.5: the standby's * next heartbeat reply reflects the post-batch state, not a partial state. * * Strategy: start with a pre-set applied_lsn, apply a batch of 4 entries, * verify that applied_lsn does NOT advance to an intermediate value and IS * exactly the last contiguously-held LSN at the end (bug-991: last_lsn, not * last_lsn+1). We confirm the "only-after-full-batch" invariant by checking the * exact final value equals last_lsn (not some intermediate held LSN from a * mid-batch advance). * Requirements 5.3, 5.5. */ static void test_wal_catchup_batch_applied_lsn_only_after_full_batch(void) { char tmpdir[256]; persist_t *p = make_test_persist_catchup(tmpdir, sizeof(tmpdir)); ha_state_t ha; memset(&ha, 0, sizeof(ha)); ha_config_t cfg; memset(&cfg, 0, sizeof(cfg)); cfg.heartbeat_timeout_ms = 10000; cfg.election_timeout_ms = 1500; ha_init(&ha, &cfg, 0); /* Pre-set applied_lsn to a known value so we can distinguish partial * from full advance. */ atomic_store_explicit(&ha.applied_lsn, 100, memory_order_release); /* Seed current_lsn so the batch [200..203] is contiguous (bug-850 guard). */ seed_persist_current_lsn(p, 200); /* Build a batch of 4 entries at LSNs 200, 201, 202, 203 */ wal_catchup_entry_t e0, e1, e2, e3; memset(&e0, 0, sizeof(e0)); memset(&e1, 0, sizeof(e1)); memset(&e2, 0, sizeof(e2)); memset(&e3, 0, sizeof(e3)); e0.lsn = 200; e0.entry_type = (uint8_t)WAL_JOB_UPDATE; e1.lsn = 201; e1.entry_type = (uint8_t)WAL_JOB_UPDATE; e2.lsn = 202; e2.entry_type = (uint8_t)WAL_JOB_UPDATE; e3.lsn = 203; e3.entry_type = (uint8_t)WAL_JOB_UPDATE; wal_catchup_entry_t *ptrs[4] = { &e0, &e1, &e2, &e3 }; wal_catchup_entry_t outer; memset(&outer, 0, sizeof(outer)); outer.n_entries = 4; outer.entries = ptrs; int rc = ha_apply_wal_catchup_entry(&ha, p, &outer, NULL, NULL); TEST_ASSERT_EQUAL_INT(0, rc); /* bug-991: applied_lsn must be exactly 203 (the last contiguously-held LSN) — * the post-batch value. The test for correctness is that the final value * equals the LAST entry's lsn, not any intermediate held LSN. * Specifically, it must NOT remain at 200 or 201 (partial advance). */ uint64_t final_lsn = (uint64_t)atomic_load_explicit(&ha.applied_lsn, memory_order_acquire); /* Must have advanced past starting value (100) */ TEST_ASSERT_GREATER_THAN_UINT64(100, final_lsn); /* Must equal the last-held LSN (203), not any intermediate value */ TEST_ASSERT_EQUAL_UINT64(203, final_lsn); ha_fini(&ha); destroy_test_persist_catchup(p); char cmd[512]; snprintf(cmd, sizeof(cmd), "rm -rf %s", tmpdir); system(cmd); printf("test_wal_catchup_batch_applied_lsn_only_after_full_batch: PASS\n"); } /* ── Task 2.3: OPCode_nop heartbeat reply carries stateReport ──────────── */ #ifdef XCE_TESTING /* Test seam declared in cluster.c under #ifdef XCE_TESTING — * fills *sr from ha->applied_lsn / myconf->config_version_tag, * sets reply_out->data_which = stateReportMsg and reply_out->data.state_report = sr. * Req 1.1, 1.3, 1.4. */ void ha_nop_reply_state_report(ha_state_t *ha, myconf_t *myconf, state_report_t *sr, xce_reply_t *reply_out); /* Test 2.3-A: nop reply carries stateReport with applied_lsn and config_tag. * Requirements 1.1, 1.3, 1.4: standby populates stateReport in every * OPCode_nop heartbeat reply with current applied_lsn and config_version_tag. */ static void test_nop_reply_carries_state_report(void) { ha_state_t ha; memset(&ha, 0, sizeof(ha)); ha_config_t cfg; memset(&cfg, 0, sizeof(cfg)); cfg.heartbeat_timeout_ms = 10000; cfg.election_timeout_ms = 1500; ha_init(&ha, &cfg, 0); /* Set a known applied_lsn */ atomic_store_explicit(&ha.applied_lsn, 12345ULL, memory_order_release); /* Build a minimal myconf with a known config_version_tag */ myconf_t myconf; memset(&myconf, 0, sizeof(myconf)); myconf.config_version_tag = 0xABCD1234ULL; state_report_t sr; memset(&sr, 0, sizeof(sr)); xce_reply_t reply; memset(&reply, 0, sizeof(reply)); ha_nop_reply_state_report(&ha, &myconf, &sr, &reply); /* Reply must use stateReportMsg discriminant */ TEST_ASSERT_EQUAL_INT(RequestReply_data_stateReportMsg, reply.data_which); /* state_report pointer must point at our sr */ TEST_ASSERT_EQUAL_PTR(&sr, reply.data.state_report); /* applied_lsn must match ha->applied_lsn */ TEST_ASSERT_EQUAL_UINT64(12345ULL, sr.applied_lsn); /* config_tag must match myconf->config_version_tag */ TEST_ASSERT_EQUAL_UINT64(0xABCD1234ULL, sr.config_tag); ha_fini(&ha); printf("test_nop_reply_carries_state_report: PASS\n"); } /* Test 2.3-B: nop reply when applied_lsn==0 (fresh standby, no config). * Requirement 1.4: config_tag reports zero when no config received yet. */ static void test_nop_reply_zero_state_fresh_standby(void) { ha_state_t ha; memset(&ha, 0, sizeof(ha)); ha_config_t cfg; memset(&cfg, 0, sizeof(cfg)); cfg.heartbeat_timeout_ms = 10000; cfg.election_timeout_ms = 1500; ha_init(&ha, &cfg, 0); /* applied_lsn stays at 0; config_version_tag stays at 0 → no config yet */ myconf_t myconf; memset(&myconf, 0, sizeof(myconf)); state_report_t sr; memset(&sr, 0, sizeof(sr)); xce_reply_t reply; memset(&reply, 0, sizeof(reply)); ha_nop_reply_state_report(&ha, &myconf, &sr, &reply); TEST_ASSERT_EQUAL_INT(RequestReply_data_stateReportMsg, reply.data_which); TEST_ASSERT_EQUAL_PTR(&sr, reply.data.state_report); TEST_ASSERT_EQUAL_UINT64(0ULL, sr.applied_lsn); TEST_ASSERT_EQUAL_UINT64(0ULL, sr.config_tag); ha_fini(&ha); printf("test_nop_reply_zero_state_fresh_standby: PASS\n"); } #endif /* XCE_TESTING */ #endif /* XCE_TESTING */ /* ── Task 3.4: ha_stream_wal_to_standby batch WAL streaming ──────────────── */ #ifdef XCE_TESTING /* * xce_wal_stream_request_hook: test-only seam declared in config_repl.c. * When non-NULL, ha_stream_wal_to_standby() calls this INSTEAD OF xce_request() * for each batch message, allowing tests to inspect and count batches without * a real endpoint. * * Signature: int hook(endpoint_t *ep, xce_request_t *req) * ep — outbound endpoint (may be NULL in tests) * req — the batch OPCode_walCatchupEntry request being sent * Returns 0 on success (the hook owns no resources in the req). */ extern int (*xce_wal_stream_request_hook)(endpoint_t *ep, xce_request_t *req); /* Helper: write N minimal WAL entries into a single segment file in wal_dir. * Entries have consecutive LSNs starting at start_lsn. * Returns 0 on success, -1 on error. */ static int write_test_wal_segment(const char *wal_dir, uint64_t start_lsn, int n_entries) { char seg_path[4096]; snprintf(seg_path, sizeof(seg_path), "%s/seg-%016" PRIx64 ".wal", wal_dir, start_lsn); int fd = open(seg_path, O_WRONLY | O_CREAT | O_TRUNC, 0600); if (fd < 0) return -1; for (int i = 0; i < n_entries; i++) { wal_entry_hdr_t hdr; memset(&hdr, 0, sizeof(hdr)); hdr.lsn = start_lsn + (uint64_t)i; hdr.type = WAL_JOB_UPDATE; hdr.payload_len = 0; hdr.term = 1; ssize_t nw = write(fd, &hdr, sizeof(hdr)); if (nw != (ssize_t)sizeof(hdr)) { close(fd); return -1; } } close(fd); return 0; } /* Batch-counting state for the hook */ static int g_wal_stream_batch_count = 0; static int g_wal_stream_total_entries = 0; static int g_wal_stream_last_n_entries = 0; static int counting_wal_stream_hook(endpoint_t *ep, xce_request_t *req) { (void)ep; if (!req || !req->body.walcatchupentrymsg) return 0; wal_catchup_entry_t *batch = (wal_catchup_entry_t *)req->body.walcatchupentrymsg; g_wal_stream_batch_count++; g_wal_stream_last_n_entries = batch->n_entries; g_wal_stream_total_entries += batch->n_entries; return 0; } /* * Test 3.4-A: streaming 500 entries produces exactly 2 batches * (batch 1: 256 entries, batch 2: 244 entries). * Requirement 5.2: accumulate up to 256 entries per batch. * Requirement 5.4: partial final batch is sent immediately. */ static void test_wal_stream_500_entries_two_batches(void) { /* Set up WAL directory with 500 consecutive entries starting at LSN 1 */ char tmpdir[256]; snprintf(tmpdir, sizeof(tmpdir), "/tmp/xce_wal_stream_%d_%ld", (int)getpid(), (long)time(NULL)); TEST_ASSERT_EQUAL_INT(0, mkdir(tmpdir, 0700)); /* Write all 500 entries in a single segment starting at LSN 1 */ int rc = write_test_wal_segment(tmpdir, /*start_lsn=*/1, /*n_entries=*/500); TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, "Failed to write test WAL segment"); /* Install counting hook */ xce_wal_stream_request_hook = counting_wal_stream_hook; g_wal_stream_batch_count = 0; g_wal_stream_total_entries = 0; g_wal_stream_last_n_entries = 0; /* Stream entries [1..500]: standby_lsn=0, committed_lsn=500 */ rc = ha_stream_wal_to_standby(/*ep=*/NULL, tmpdir, /*p=*/NULL, /*leader_term=*/7, /*standby_lsn=*/0, /*committed_lsn=*/500); TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, "ha_stream_wal_to_standby failed"); TEST_ASSERT_EQUAL_INT_MESSAGE(2, g_wal_stream_batch_count, "500 entries must produce exactly 2 batches (256+244)"); TEST_ASSERT_EQUAL_INT_MESSAGE(500, g_wal_stream_total_entries, "All 500 entries must be sent"); TEST_ASSERT_EQUAL_INT_MESSAGE(244, g_wal_stream_last_n_entries, "Final batch must contain 244 entries (500 - 256)"); xce_wal_stream_request_hook = NULL; char cmd[512]; snprintf(cmd, sizeof(cmd), "rm -rf %s", tmpdir); system(cmd); printf("test_wal_stream_500_entries_two_batches: PASS\n"); } /* * Test 3.4-B: streaming fewer than 256 entries produces exactly 1 partial batch. * Requirement 5.4: partial final batch sent immediately without waiting. */ static void test_wal_stream_partial_batch(void) { char tmpdir[256]; snprintf(tmpdir, sizeof(tmpdir), "/tmp/xce_wal_stream_partial_%d_%ld", (int)getpid(), (long)time(NULL)); TEST_ASSERT_EQUAL_INT(0, mkdir(tmpdir, 0700)); /* Write 10 entries starting at LSN 5 */ int rc = write_test_wal_segment(tmpdir, /*start_lsn=*/5, /*n_entries=*/10); TEST_ASSERT_EQUAL_INT(0, rc); xce_wal_stream_request_hook = counting_wal_stream_hook; g_wal_stream_batch_count = 0; g_wal_stream_total_entries = 0; g_wal_stream_last_n_entries = 0; /* Stream entries [6..14]: standby_lsn=5, committed_lsn=14 */ rc = ha_stream_wal_to_standby(/*ep=*/NULL, tmpdir, /*p=*/NULL, /*leader_term=*/7, /*standby_lsn=*/5, /*committed_lsn=*/14); TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, "ha_stream_wal_to_standby failed"); TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_wal_stream_batch_count, "< 256 entries must produce exactly 1 partial batch"); TEST_ASSERT_EQUAL_INT_MESSAGE(9, g_wal_stream_total_entries, "9 entries in range [6..14] must be sent"); xce_wal_stream_request_hook = NULL; char cmd[512]; snprintf(cmd, sizeof(cmd), "rm -rf %s", tmpdir); system(cmd); printf("test_wal_stream_partial_batch: PASS\n"); } /* * Test 3.4-C: empty range (standby_lsn == committed_lsn) sends zero batches. * Requirement 5.4: no batch is sent when there are no entries to stream. */ static void test_wal_stream_empty_range_no_batches(void) { char tmpdir[256]; snprintf(tmpdir, sizeof(tmpdir), "/tmp/xce_wal_stream_empty_%d_%ld", (int)getpid(), (long)time(NULL)); TEST_ASSERT_EQUAL_INT(0, mkdir(tmpdir, 0700)); xce_wal_stream_request_hook = counting_wal_stream_hook; g_wal_stream_batch_count = 0; g_wal_stream_total_entries = 0; /* standby_lsn == committed_lsn: nothing to stream */ int rc = ha_stream_wal_to_standby(/*ep=*/NULL, tmpdir, /*p=*/NULL, /*leader_term=*/7, /*standby_lsn=*/100, /*committed_lsn=*/100); TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, "Empty range must return 0"); TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_wal_stream_batch_count, "Empty range must produce zero batches"); xce_wal_stream_request_hook = NULL; char cmd[512]; snprintf(cmd, sizeof(cmd), "rm -rf %s", tmpdir); system(cmd); printf("test_wal_stream_empty_range_no_batches: PASS\n"); } /* * bug-1028 — the catch-up streamer must source each streamed entry's term and * the AppendEntries prev_log_term from the AUTHORITATIVE per-LSN term index * (persist_entry_term), NOT the raw on-disk hdr.term. A legacy-0 entry has * hdr.term==0 on disk but persist_entry_term maps it to WAL_LEGACY_TERM(1) — * which is exactly what the standby's logOk and the library send path compute. * If the streamer sends the raw 0, prev_log_term(0) != standby's term(1) → logOk * rejects the new-term entry forever (the "stream WAL batches" wedge). * * Fixture: LSN 1 appended at term 0 (legacy), LSN 2 at term 3. Streaming * (standby_lsn=1, committed=2] must emit LSN 2 with term=3 and prev_log_term=1 * (NOT 0). RED before the fix (streamer sent hdr.term=0 for the predecessor). */ static uint64_t g_cap_term_for_lsn2 = 0; static uint64_t g_cap_prevterm_for_lsn2 = 0; static int g_cap_saw_lsn2 = 0; static int capture_terms_wal_stream_hook(endpoint_t *ep, xce_request_t *req) { (void)ep; if (!req || !req->body.walcatchupentrymsg) return 0; wal_catchup_entry_t *batch = (wal_catchup_entry_t *)req->body.walcatchupentrymsg; for (int i = 0; i < batch->n_entries; i++) { wal_catchup_entry_t *e = batch->entries[i]; if (e && e->lsn == 2) { g_cap_saw_lsn2 = 1; g_cap_term_for_lsn2 = e->term; g_cap_prevterm_for_lsn2 = e->prev_log_term; } } return 0; } static void test_wal_stream_term_from_index_not_raw_hdr(void) { char tmp[256]; strcpy(tmp, "/tmp/xce_cr_streamterm_XXXXXX"); TEST_ASSERT_NOT_NULL(mkdtemp(tmp)); cluster_t *c = t42_make_cluster(tmp); TEST_ASSERT_NOT_NULL(c); /* LSN 1: legacy entry (term 0 on disk → persist_entry_term maps to 1). */ persist_set_term(&c->persist, 0); TEST_ASSERT_EQUAL_INT(0, persist_wal_async(&c->persist, WAL_JOB_UPDATE, NULL, 0)); /* LSN 2: real new-term entry (term 3). */ persist_set_term(&c->persist, 3); TEST_ASSERT_EQUAL_INT(0, persist_wal_async(&c->persist, WAL_JOB_UPDATE, NULL, 0)); /* Sanity: the authoritative index normalises legacy-0 → 1, keeps 3. */ TEST_ASSERT_EQUAL_UINT64(WAL_LEGACY_TERM, persist_entry_term(&c->persist, 1)); TEST_ASSERT_EQUAL_UINT64(3, persist_entry_term(&c->persist, 2)); xce_wal_stream_request_hook = capture_terms_wal_stream_hook; g_cap_saw_lsn2 = 0; g_cap_term_for_lsn2 = 0; g_cap_prevterm_for_lsn2 = 0; /* Standby holds LSN 1, catching up to committed 2 — streams just LSN 2, * whose prev tuple references LSN 1 (the legacy entry). */ int rc = ha_stream_wal_to_standby(/*ep=*/NULL, c->persist.wal_path, &c->persist, /*leader_term=*/atomic_load_explicit( &c->ha.term, memory_order_acquire), /*standby_lsn=*/1, /*committed_lsn=*/2); TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, "ha_stream_wal_to_standby failed"); xce_wal_stream_request_hook = NULL; TEST_ASSERT_TRUE_MESSAGE(g_cap_saw_lsn2, "LSN 2 must be streamed"); TEST_ASSERT_EQUAL_UINT64_MESSAGE(3, g_cap_term_for_lsn2, "streamed entry term must come from persist_entry_term (3)"); TEST_ASSERT_EQUAL_UINT64_MESSAGE(WAL_LEGACY_TERM, g_cap_prevterm_for_lsn2, "bug-1028: prev_log_term must be the index-normalised legacy term (1), " "NOT raw on-disk hdr.term (0) — else the standby's logOk rejects forever"); t42_teardown(c); char cmd[512]; snprintf(cmd, sizeof(cmd), "rm -rf %s", tmp); system(cmd); printf("test_wal_stream_term_from_index_not_raw_hdr: PASS\n"); } /* ── Task 3.5: Snapshot chunk transfer ──────────────────────── */ static snapshot_transfer_t g_snaptest_chunks[16]; static int g_snaptest_chunk_count = 0; static void snaptest_reset(void) { memset(g_snaptest_chunks, 0, sizeof(g_snaptest_chunks)); g_snaptest_chunk_count = 0; } static int snaptest_hook(endpoint_t *ep, xce_request_t *req) { (void)ep; if (g_snaptest_chunk_count >= 16) return -1; memcpy(&g_snaptest_chunks[g_snaptest_chunk_count], req->body.snapshottransfermsg, sizeof(snapshot_transfer_t)); g_snaptest_chunk_count++; return 0; } static void test_snapshot_transfer_two_chunks(void) { char tmpdir[] = "/tmp/xcectest_snaptrans_XXXXXX"; TEST_ASSERT_NOT_NULL(mkdtemp(tmpdir)); char snap_path[512]; snprintf(snap_path, sizeof(snap_path), "%s/snap-0000000000000064.capnp", tmpdir); FILE *f = fopen(snap_path, "wb"); TEST_ASSERT_NOT_NULL(f); char header[16] = "XCESNAP\0"; uint64_t lsn = 100; memcpy(header + 8, &lsn, 8); fwrite(header, 1, 16, f); char zero[4096]; memset(zero, 0, sizeof(zero)); for (int i = 16; i < 131072; i += 4096) fwrite(zero, 1, (i + 4096 > 131072) ? (size_t)(131072 - i) : 4096, f); fclose(f); extern int (*xce_snapshot_chunk_hook)(endpoint_t *, xce_request_t *); xce_snapshot_chunk_hook = snaptest_hook; snaptest_reset(); extern int ha_send_snapshot_to_standby(endpoint_t *, const char *, uint64_t); int rc = ha_send_snapshot_to_standby(NULL, tmpdir, lsn); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_EQUAL_INT(2, g_snaptest_chunk_count); TEST_ASSERT_EQUAL_UINT64(100, g_snaptest_chunks[0].snapshot_lsn); TEST_ASSERT_EQUAL_UINT32(0, g_snaptest_chunks[0].chunk_index); TEST_ASSERT_EQUAL_UINT64(0, g_snaptest_chunks[0].chunk_offset); TEST_ASSERT_EQUAL_UINT64(131072, g_snaptest_chunks[0].total_size); TEST_ASSERT_EQUAL_INT(0, g_snaptest_chunks[0].is_last); TEST_ASSERT_EQUAL_UINT64(131072, g_snaptest_chunks[1].total_size); TEST_ASSERT_EQUAL_UINT32(1, g_snaptest_chunks[1].chunk_index); TEST_ASSERT_EQUAL_UINT64(65536, g_snaptest_chunks[1].chunk_offset); TEST_ASSERT_EQUAL_INT(1, g_snaptest_chunks[1].is_last); xce_snapshot_chunk_hook = NULL; char cmd[512]; snprintf(cmd, sizeof(cmd), "rm -rf %s", tmpdir); system(cmd); printf("test_snapshot_transfer_two_chunks: PASS\n"); } static void test_snapshot_transfer_small_less_than_one_chunk(void) { char tmpdir[] = "/tmp/xcectest_snaptrans_XXXXXX"; TEST_ASSERT_NOT_NULL(mkdtemp(tmpdir)); char snap_path[512]; snprintf(snap_path, sizeof(snap_path), "%s/snap-000000000000000a.capnp", tmpdir); FILE *f = fopen(snap_path, "wb"); TEST_ASSERT_NOT_NULL(f); char buf[64]; memset(buf, 0, sizeof(buf)); memcpy(buf, "XCESNAP\0", 8); uint64_t lsn = 10; memcpy(buf + 8, &lsn, 8); fwrite(buf, 1, sizeof(buf), f); fclose(f); extern int (*xce_snapshot_chunk_hook)(endpoint_t *, xce_request_t *); xce_snapshot_chunk_hook = snaptest_hook; snaptest_reset(); extern int ha_send_snapshot_to_standby(endpoint_t *, const char *, uint64_t); int rc = ha_send_snapshot_to_standby(NULL, tmpdir, lsn); TEST_ASSERT_EQUAL_INT(0, rc); TEST_ASSERT_EQUAL_INT(1, g_snaptest_chunk_count); TEST_ASSERT_EQUAL_UINT64(10, g_snaptest_chunks[0].snapshot_lsn); TEST_ASSERT_EQUAL_UINT32(0, g_snaptest_chunks[0].chunk_index); TEST_ASSERT_EQUAL_UINT64(64, g_snaptest_chunks[0].total_size); TEST_ASSERT_EQUAL_INT(1, g_snaptest_chunks[0].is_last); xce_snapshot_chunk_hook = NULL; char cmd[512]; snprintf(cmd, sizeof(cmd), "rm -rf %s", tmpdir); system(cmd); printf("test_snapshot_transfer_small_less_than_one_chunk: PASS\n"); } /* ── Task 3.5c: Standby snapshot receive handler ──────────────────────────── */ /* Helper: initialise a minimal ha_state_t for snapshot receive tests */ static void snap_recv_ha_init(ha_state_t *ha) { memset(ha, 0, sizeof(*ha)); ha_config_t cfg; memset(&cfg, 0, sizeof(cfg)); cfg.heartbeat_timeout_ms = 10000; cfg.election_timeout_ms = 1500; ha_init(ha, &cfg, 0); } /* Helper: write a minimal valid snapshot file to path (XCESNAP header + body) */ static void write_test_snapshot(const char *path, uint64_t lsn, size_t body_size) { FILE *f = fopen(path, "wb"); TEST_ASSERT_NOT_NULL_MESSAGE(f, "Failed to create test snapshot file"); char header[16]; memcpy(header, "XCESNAP\0", 8); memcpy(header + 8, &lsn, 8); fwrite(header, 1, 16, f); if (body_size > 16) { char *zeros = calloc(1, body_size - 16); fwrite(zeros, 1, body_size - 16, f); free(zeros); } fclose(f); } /* Test 3.5c-A: receive 2 chunks, verify snap.part assembled and renamed correctly, * applied_lsn set to checkpoint_lsn. * Requirements: 4.2, 4.3 */ static void test_snapshot_recv_two_chunks_assembles_and_renames(void) { char tmpdir[] = "/tmp/xcectest_snaprecv_XXXXXX"; TEST_ASSERT_NOT_NULL(mkdtemp(tmpdir)); /* Create snap_dir subdirectory (matches persist->snap_path layout) */ char snap_dir[512]; snprintf(snap_dir, sizeof(snap_dir), "%s/snap", tmpdir); TEST_ASSERT_EQUAL_INT(0, mkdir(snap_dir, 0700)); /* Build a fake 128-byte snapshot body (two 64-byte chunks). * payload_bytes contains the raw file bytes for each chunk. * We skip writing an actual snapshot source file — the receiver * writes snap.part and renames it; we use the resulting file to verify. */ uint64_t snap_lsn = 100; size_t total_size = 128; uint8_t chunk0[64]; uint8_t chunk1[64]; memset(chunk0, 0xAA, 64); memset(chunk1, 0xBB, 64); /* Embed XCESNAP header in chunk0 so load_snapshot can parse it */ memcpy(chunk0, "XCESNAP\0", 8); memcpy(chunk0 + 8, &snap_lsn, 8); persist_t *p = calloc(1, sizeof(persist_t)); TEST_ASSERT_NOT_NULL(p); persist_config_t pcfg = persist_config_defaults(); persist_set_workdir(p, tmpdir, "ctrl-snaprecv"); persist_init(p, &pcfg, NULL); persist_start(p); ha_state_t ha; snap_recv_ha_init(&ha); /* Chunk 0 */ snapshot_transfer_t c0; memset(&c0, 0, sizeof(c0)); c0.snapshot_lsn = snap_lsn; c0.chunk_index = 0; c0.chunk_offset = 0; c0.total_size = (uint64_t)total_size; c0.is_last = 0; int rc = ha_handle_snapshot_chunk(&ha, p, snap_dir, &c0, chunk0, 64); TEST_ASSERT_EQUAL_INT(0, rc); /* snap.part must exist after first chunk */ char part_path[600]; snprintf(part_path, sizeof(part_path), "%s/snap.part", snap_dir); struct stat st; TEST_ASSERT_EQUAL_INT_MESSAGE(0, stat(part_path, &st), "snap.part should exist after chunk 0"); TEST_ASSERT_EQUAL_UINT64(64, (uint64_t)st.st_size); /* Chunk 1 (isLast=true) */ snapshot_transfer_t c1; memset(&c1, 0, sizeof(c1)); c1.snapshot_lsn = snap_lsn; c1.chunk_index = 1; c1.chunk_offset = 64; c1.total_size = (uint64_t)total_size; c1.is_last = 1; rc = ha_handle_snapshot_chunk(&ha, p, snap_dir, &c1, chunk1, 64); TEST_ASSERT_EQUAL_INT(0, rc); /* snap.part must be gone (renamed) */ TEST_ASSERT_NOT_EQUAL_MESSAGE(0, stat(part_path, &st), "snap.part should be gone after isLast"); /* Final file snap-{lsn:016x}.bin must exist */ char final_path[600]; snprintf(final_path, sizeof(final_path), "%s/snap-%016" PRIx64 ".bin", snap_dir, snap_lsn); TEST_ASSERT_EQUAL_INT_MESSAGE(0, stat(final_path, &st), "snap-{lsn}.bin should exist after isLast"); TEST_ASSERT_EQUAL_UINT64((uint64_t)total_size, (uint64_t)st.st_size); /* applied_lsn must be set to checkpoint_lsn (= snap_lsn = 100) */ uint64_t got_lsn = (uint64_t)atomic_load_explicit(&ha.applied_lsn, memory_order_acquire); TEST_ASSERT_EQUAL_UINT64(snap_lsn, got_lsn); ha_fini(&ha); persist_stop(p); persist_fini(p); free(p); char cmd[512]; snprintf(cmd, sizeof(cmd), "rm -rf %s", tmpdir); system(cmd); printf("test_snapshot_recv_two_chunks_assembles_and_renames: PASS\n"); } /* Test 3.5c-B: size mismatch on isLast returns error (verify size == total_size). * Requirements: 4.3 */ static void test_snapshot_recv_size_mismatch_fails(void) { char tmpdir[] = "/tmp/xcectest_snaprecv2_XXXXXX"; TEST_ASSERT_NOT_NULL(mkdtemp(tmpdir)); char snap_dir[512]; snprintf(snap_dir, sizeof(snap_dir), "%s/snap", tmpdir); TEST_ASSERT_EQUAL_INT(0, mkdir(snap_dir, 0700)); persist_t *p = calloc(1, sizeof(persist_t)); TEST_ASSERT_NOT_NULL(p); persist_config_t pcfg = persist_config_defaults(); persist_set_workdir(p, tmpdir, "ctrl-snaprecv2"); persist_init(p, &pcfg, NULL); persist_start(p); ha_state_t ha; snap_recv_ha_init(&ha); uint64_t snap_lsn = 200; uint8_t chunk0[64]; memset(chunk0, 0xCC, 64); /* Claim total_size=200 but only send 64 bytes with is_last=true */ snapshot_transfer_t c0; memset(&c0, 0, sizeof(c0)); c0.snapshot_lsn = snap_lsn; c0.chunk_index = 0; c0.chunk_offset = 0; c0.total_size = 200; /* mismatch — only 64 bytes will be written */ c0.is_last = 1; int rc = ha_handle_snapshot_chunk(&ha, p, snap_dir, &c0, chunk0, 64); /* Must fail — size mismatch */ TEST_ASSERT_NOT_EQUAL_MESSAGE(0, rc, "size mismatch must return error"); /* applied_lsn must remain 0 (not advanced on failure) */ uint64_t got_lsn = (uint64_t)atomic_load_explicit(&ha.applied_lsn, memory_order_acquire); TEST_ASSERT_EQUAL_UINT64(0, got_lsn); ha_fini(&ha); persist_stop(p); persist_fini(p); free(p); char cmd[512]; snprintf(cmd, sizeof(cmd), "rm -rf %s", tmpdir); system(cmd); printf("test_snapshot_recv_size_mismatch_fails: PASS\n"); } /* Test 3.5c-C: NULL snap_dir or NULL chunk returns error gracefully */ static void test_snapshot_recv_null_args_fail(void) { ha_state_t ha; snap_recv_ha_init(&ha); int rc = ha_handle_snapshot_chunk(&ha, NULL, NULL, NULL, NULL, 0); TEST_ASSERT_NOT_EQUAL(0, rc); ha_fini(&ha); printf("test_snapshot_recv_null_args_fail: PASS\n"); } /* Test 6.4-A: 128KB snapshot in 2 equal chunks (65536 bytes each) — standby * receives both chunks, verifies total_size matches, snap.part is atomically * renamed to snap-{lsn}.bin, persist_load_snapshot is called successfully, * and applied_lsn equals the checkpoint_lsn embedded in the snapshot. * * This is the primary task-6.4 coverage test for Req 4.2, 4.3. * Uses the same 65536-byte chunk size as the sender test * (test_snapshot_transfer_two_chunks) to exercise the full receive path. */ static void test_snapshot_recv_128kb_two_chunks(void) { char tmpdir[] = "/tmp/xcectest_snap128kb_XXXXXX"; TEST_ASSERT_NOT_NULL(mkdtemp(tmpdir)); char snap_dir[512]; snprintf(snap_dir, sizeof(snap_dir), "%s/snap", tmpdir); TEST_ASSERT_EQUAL_INT(0, mkdir(snap_dir, 0700)); persist_t *p = calloc(1, sizeof(persist_t)); TEST_ASSERT_NOT_NULL(p); persist_config_t pcfg = persist_config_defaults(); persist_set_workdir(p, tmpdir, "ctrl-snap128kb"); persist_init(p, &pcfg, NULL); persist_start(p); ha_state_t ha; snap_recv_ha_init(&ha); /* 128KB snapshot split into two 65536-byte chunks */ const uint64_t snap_lsn = 0x64; /* LSN 100 — matches sender test */ const size_t chunk_size = 65536; const size_t total_size = 131072; /* 128 * 1024 */ /* Allocate and zero-fill chunks */ uint8_t *chunk0 = calloc(1, chunk_size); uint8_t *chunk1 = calloc(1, chunk_size); TEST_ASSERT_NOT_NULL(chunk0); TEST_ASSERT_NOT_NULL(chunk1); /* Embed valid XCESNAP header in chunk0 so persist_load_snapshot succeeds */ memcpy(chunk0, "XCESNAP\0", 8); memcpy(chunk0 + 8, &snap_lsn, 8); /* ── Chunk 0: not last ── */ snapshot_transfer_t c0; memset(&c0, 0, sizeof(c0)); c0.snapshot_lsn = snap_lsn; c0.chunk_index = 0; c0.chunk_offset = 0; c0.total_size = (uint64_t)total_size; c0.is_last = 0; int rc = ha_handle_snapshot_chunk(&ha, p, snap_dir, &c0, chunk0, chunk_size); TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, "chunk 0 must succeed"); /* snap.part must exist and be exactly chunk_size bytes */ char part_path[600]; snprintf(part_path, sizeof(part_path), "%s/snap.part", snap_dir); struct stat st; TEST_ASSERT_EQUAL_INT_MESSAGE(0, stat(part_path, &st), "snap.part must exist after chunk 0"); TEST_ASSERT_EQUAL_UINT64_MESSAGE((uint64_t)chunk_size, (uint64_t)st.st_size, "snap.part must be chunk_size bytes after chunk 0"); /* ── Chunk 1: isLast=true ── */ snapshot_transfer_t c1; memset(&c1, 0, sizeof(c1)); c1.snapshot_lsn = snap_lsn; c1.chunk_index = 1; c1.chunk_offset = (uint64_t)chunk_size; c1.total_size = (uint64_t)total_size; c1.is_last = 1; rc = ha_handle_snapshot_chunk(&ha, p, snap_dir, &c1, chunk1, chunk_size); TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, "chunk 1 (isLast) must succeed"); /* snap.part must be gone (atomically renamed) */ TEST_ASSERT_NOT_EQUAL_MESSAGE(0, stat(part_path, &st), "snap.part must be gone after isLast rename"); /* Final file snap-{lsn:016x}.bin must exist with total_size bytes */ char final_path[600]; snprintf(final_path, sizeof(final_path), "%s/snap-%016" PRIx64 ".bin", snap_dir, snap_lsn); TEST_ASSERT_EQUAL_INT_MESSAGE(0, stat(final_path, &st), "snap-{lsn}.bin must exist after isLast"); TEST_ASSERT_EQUAL_UINT64_MESSAGE((uint64_t)total_size, (uint64_t)st.st_size, "snap-{lsn}.bin must be exactly total_size bytes"); /* applied_lsn must equal checkpoint_lsn (snap_lsn) after successful load */ uint64_t got_lsn = (uint64_t)atomic_load_explicit(&ha.applied_lsn, memory_order_acquire); TEST_ASSERT_EQUAL_UINT64_MESSAGE(snap_lsn, got_lsn, "applied_lsn must equal checkpoint_lsn after load"); free(chunk0); free(chunk1); ha_fini(&ha); persist_stop(p); persist_fini(p); free(p); char cmd[512]; snprintf(cmd, sizeof(cmd), "rm -rf %s", tmpdir); system(cmd); printf("test_snapshot_recv_128kb_two_chunks: PASS\n"); } /* Test 3.5d: reconnected standby discards snap.part and restarts from chunk 0. * * Design §Snapshot Transfer: "Mid-transfer reconnect always restarts snapshot * from chunk 0 — the standby discards its partial snap.part file on reconnect." * The discard (unlink) is done by the ctrlhello handler before the first chunk * of a new transfer arrives. This test verifies that: * 1. After receiving chunk 0 (creating snap.part), if the caller unlinks * snap.part (simulating reconnect/reset) and then calls ha_handle_snapshot_chunk * again with chunk_index=0, the function creates a fresh snap.part * (not appending to a previous stale file). * 2. The resulting snap.part contains only the new chunk's bytes (not a * concatenation of old + new data). * * Requirement: 4.4 (ha-repl-hardening) — retry by reconnecting initiates new transfer. */ static void test_snapshot_recv_reconnect_discards_snappart(void) { char tmpdir[] = "/tmp/xcectest_snapreconn_XXXXXX"; TEST_ASSERT_NOT_NULL(mkdtemp(tmpdir)); char snap_dir[512]; snprintf(snap_dir, sizeof(snap_dir), "%s/snap", tmpdir); TEST_ASSERT_EQUAL_INT(0, mkdir(snap_dir, 0700)); persist_t *p = calloc(1, sizeof(persist_t)); TEST_ASSERT_NOT_NULL(p); persist_config_t pcfg = persist_config_defaults(); persist_set_workdir(p, tmpdir, "ctrl-snapreconn"); persist_init(p, &pcfg, NULL); persist_start(p); ha_state_t ha; snap_recv_ha_init(&ha); uint64_t snap_lsn = 300; uint8_t first[64]; uint8_t second[64]; memset(first, 0xAA, 64); memset(second, 0xBB, 64); /* ── First transfer attempt: send chunk 0 ── */ snapshot_transfer_t c0; memset(&c0, 0, sizeof(c0)); c0.snapshot_lsn = snap_lsn; c0.chunk_index = 0; c0.chunk_offset = 0; c0.total_size = 64; /* single-chunk transfer */ c0.is_last = 0; /* not last — simulates partial in-flight transfer */ int rc = ha_handle_snapshot_chunk(&ha, p, snap_dir, &c0, first, 64); TEST_ASSERT_EQUAL_INT(0, rc); /* snap.part must exist and contain exactly 64 bytes (first chunk) */ char part_path[600]; snprintf(part_path, sizeof(part_path), "%s/snap.part", snap_dir); struct stat st; TEST_ASSERT_EQUAL_INT_MESSAGE(0, stat(part_path, &st), "snap.part must exist after first chunk"); TEST_ASSERT_EQUAL_UINT64_MESSAGE(64, (uint64_t)st.st_size, "snap.part must be 64 bytes after first chunk"); /* ── Simulate reconnect: caller discards snap.part ── */ TEST_ASSERT_EQUAL_INT_MESSAGE(0, unlink(part_path), "unlink snap.part must succeed (simulating reconnect)"); /* ── Second transfer attempt: restart from chunk 0 ── */ snapshot_transfer_t c0b; memset(&c0b, 0, sizeof(c0b)); c0b.snapshot_lsn = snap_lsn; c0b.chunk_index = 0; c0b.chunk_offset = 0; c0b.total_size = 64; c0b.is_last = 0; /* still not last — just verifying discard */ rc = ha_handle_snapshot_chunk(&ha, p, snap_dir, &c0b, second, 64); TEST_ASSERT_EQUAL_INT(0, rc); /* snap.part must exist and contain exactly 64 bytes (second chunk only, * not 128 bytes which would indicate old data was appended) */ TEST_ASSERT_EQUAL_INT_MESSAGE(0, stat(part_path, &st), "snap.part must exist after reconnect chunk 0"); TEST_ASSERT_EQUAL_UINT64_MESSAGE(64, (uint64_t)st.st_size, "snap.part must be 64 bytes (not 128) — stale data discarded"); /* Verify the content is the second chunk (0xBB bytes), not the first (0xAA) */ FILE *f = fopen(part_path, "rb"); TEST_ASSERT_NOT_NULL_MESSAGE(f, "snap.part must be readable"); uint8_t buf[64]; size_t n = fread(buf, 1, 64, f); fclose(f); TEST_ASSERT_EQUAL_UINT64_MESSAGE(64, (uint64_t)n, "must read 64 bytes from snap.part"); TEST_ASSERT_EQUAL_UINT8_MESSAGE(0xBB, buf[0], "snap.part content must be second chunk (0xBB), not first (0xAA)"); ha_fini(&ha); persist_stop(p); persist_fini(p); free(p); char cmd[512]; snprintf(cmd, sizeof(cmd), "rm -rf %s", tmpdir); system(cmd); printf("test_snapshot_recv_reconnect_discards_snappart: PASS\n"); } /* ── Task 5.1: handle_ctrlhello — connection setup only ───────────────────── * * Verify that after the ha-repl-hardening simplification, handle_ctrlhello * does connection bookkeeping only and enqueues NO replication messages. * * Requirements: 1.1, 1.2 */ static int g_ctrlhello_repl_msg_count = 0; /* Count any xce_request calls (via the wal stream hook) that arrive during * handle_ctrlhello. After simplification this must remain 0. * Note: the WAL streaming hook is for ha_stream_wal_to_standby; if * handle_ctrlhello were still calling xce_request directly the test would * need a separate seam — but since the simplified code has none, we simply * verify no crash and correct state. */ static void test_ctrlhello_no_repl_messages_enqueued(void) { /* Build a minimal cluster_t in CTRL_STANDBY role. * Intentionally leave persist.repl, persist.wal_path, and * standby_endpoints uninitialised (zero/NULL) — the simplified * handle_ctrlhello must not touch any of these fields. * The old code would crash or corrupt memory if it reached * repl_add_standby / WAL streaming / push_config_to_standbys * with NULL repl state. */ cluster_t c; memset(&c, 0, sizeof(c)); ha_config_t hcfg; memset(&hcfg, 0, sizeof(hcfg)); hcfg.heartbeat_timeout_ms = 10000; hcfg.election_timeout_ms = 1500; ha_init(&c.ha, &hcfg, 0); c.ha.primary_ctrl_fd = -1; atomic_store_explicit(&c.ha.role, (int32_t)CTRL_STANDBY, memory_order_release); /* Give myconf a non-zero config_version_tag so the old config-push branch * would have fired (standby.config_tag=0 != primary_tag=0xCAFE). */ myconf_t conf; memset(&conf, 0, sizeof(conf)); conf.config_version_tag = 0xCAFEULL; c.myconf = &conf; /* Give persist a non-zero committed_lsn so the old WAL-streaming branch * would have fired (standby.applied_lsn=0 < committed_lsn=100). */ atomic_store_explicit(&c.persist.committed_lsn, 100ULL, memory_order_release); /* Leave persist.wal_path as NULL / empty — old code would have opened NULL. */ /* Build a minimal CtrlHelloRequest: standby reports lsn=0, tag=0 */ ctrl_hello_req_t hello; memset(&hello, 0, sizeof(hello)); hello.name = "standby-test"; hello.applied_lsn = 0; hello.config_tag = 0; xce_request_t req; memset(&req, 0, sizeof(req)); req.body_which = Request_body_ctrlhellomsg; req.body.ctrlhellomsg = &hello; endpoint_t ep; memset(&ep, 0, sizeof(ep)); ep.fd = 77; /* Reset repl message counter */ g_ctrlhello_repl_msg_count = 0; /* Call the simplified handle_ctrlhello — must return 0 without crash */ int rc = handle_ctrlhello(&c, &ep, &req); TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, "handle_ctrlhello must return 0 (Req 1.1)"); /* Connection bookkeeping must have happened */ TEST_ASSERT_EQUAL_INT_MESSAGE(77, c.ha.primary_ctrl_fd, "handle_ctrlhello must set primary_ctrl_fd = ep->fd (Req 1.1)"); /* ha_on_heartbeat_received must have been called (last_activity updated) */ TEST_ASSERT_TRUE_MESSAGE(c.ha.last_activity > 0, "handle_ctrlhello must call ha_on_heartbeat_received (Req 1.2)"); /* Idempotency: second call must not overwrite primary_ctrl_fd */ ep.fd = 99; handle_ctrlhello(&c, &ep, &req); TEST_ASSERT_EQUAL_INT_MESSAGE(77, c.ha.primary_ctrl_fd, "handle_ctrlhello: second call must not overwrite primary_ctrl_fd (Req 1.1)"); /* Verify no replication messages were counted. * With the simplified implementation, handle_ctrlhello never calls * xce_request() — so g_ctrlhello_repl_msg_count stays 0. If xce_request * were called for OPCode_replConfigPush or OPCode_walCatchupEntry the * test binary would crash (NULL ep infrastructure) before reaching here. */ TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_ctrlhello_repl_msg_count, "handle_ctrlhello must enqueue no OPCode_replConfigPush or " "OPCode_walCatchupEntry messages (Req 1.1, 1.2)"); ha_fini(&c.ha); printf("test_ctrlhello_no_repl_messages_enqueued: PASS\n"); } /* Declare handle_ctrlhello for this translation unit (defined in config_repl.c) */ extern int handle_ctrlhello(cluster_t *c, endpoint_t *ep, xce_request_t *req); #endif /* XCE_TESTING */ /* ── dispatch ───────────────────────────────────── */ void test_config_repl(void) { RUN_TEST(test_conf_read_current_filename_missing); RUN_TEST(test_conf_read_current_filename_empty_file); RUN_TEST(test_conf_read_current_filename_path_separator); RUN_TEST(test_conf_read_current_filename_bad_prefix); RUN_TEST(test_conf_read_current_filename_wrong_hex_length); RUN_TEST(test_conf_save_creates_pointer_file); RUN_TEST(test_conf_save_filename_has_12_hex_suffix); RUN_TEST(test_conf_trim_keeps_only_n_versions); RUN_TEST(test_conf_get_saved_tag_no_config); RUN_TEST(test_conf_get_saved_tag_after_save); RUN_TEST(test_conf_check_stale_disabled); RUN_TEST(test_conf_check_stale_fresh); RUN_TEST(test_conf_check_stale_aged); RUN_TEST(test_config_contains_reserved_name_absent); RUN_TEST(test_sanitise_cn_printable); RUN_TEST(test_sanitise_cn_control_chars); RUN_TEST(test_sanitise_cn_truncated); RUN_TEST(test_conf_save_config_path_consistency); /* Task 4.1: WAL_CONFIG_MARKER type + reference-only codec */ RUN_TEST(test_config_marker_enum_value_is_50); RUN_TEST(test_config_marker_encode_decode_roundtrip); RUN_TEST(test_config_marker_decode_truncated_fails); RUN_TEST(test_config_marker_encode_short_buffer_fails); /* Task 4.2: append marker via unified log + derive committed config */ RUN_TEST(test_config_marker_append_via_unified_log); RUN_TEST(test_committed_marker_derives_after_commit); RUN_TEST(test_uncommitted_marker_not_yet_committed_config); RUN_TEST(test_divergent_marker_truncated_like_job_entry); /* Task 4.2 r1: live wiring + compaction-safe derive */ RUN_TEST(test_config_commit_change_live_path_commits_and_derives); RUN_TEST(test_committed_marker_survives_checkpoint_trim); RUN_TEST(test_config_activate_match_activates_and_unfences); RUN_TEST(test_config_activate_mismatch_fences_and_does_not_activate); RUN_TEST(test_config_activate_obtained_bytes_verified_before_activate); RUN_TEST(test_config_activate_recovery_uses_latest_committed_marker); /* Task 4.3 r1 remediation: production integration (push-path sole activation, * no-marker fallback, serve-fence enforced by dispatch gate + promotion). */ RUN_TEST(test_push_path_mismatch_does_not_preactivate); RUN_TEST(test_config_activate_no_marker_falls_back_and_loads); RUN_TEST(test_serve_fence_enforced_by_dispatch_gate_and_promotion); /* Task 4.3 r2 remediation: commit-advance retry clears a push-before-commit * serve-fence (Finding D / R7.9). */ RUN_TEST(test_commit_advance_retry_clears_push_before_commit_fence); #ifdef XCE_TESTING RUN_TEST(test_load_primary_config_missing_returns_error); RUN_TEST(test_wait_all_with_timeout_slow_threads); RUN_TEST(test_wait_all_with_timeout_fast_threads); /* Task 2.1: OPCode_walCatchupEntry handler */ RUN_TEST(test_wal_catchup_entry_advances_applied_lsn); RUN_TEST(test_wal_catchup_entry_error_skipped_no_crash); RUN_TEST(test_wal_catchup_entry_null_entry_no_crash); /* Task 2.2: batch WAL catch-up entries */ RUN_TEST(test_wal_catchup_batch_advances_applied_lsn); RUN_TEST(test_wal_catchup_partial_batch); RUN_TEST(test_wal_catchup_batch_error_entry_skipped); RUN_TEST(test_wal_catchup_batch_zero_entries_fallback); /* Task 6.2: additional WAL catch-up and batch handling tests */ RUN_TEST(test_wal_catchup_batch_256_entries); RUN_TEST(test_wal_catchup_batch_applied_lsn_only_after_full_batch); /* Task 2.3: OPCode_nop heartbeat reply carries stateReport */ RUN_TEST(test_nop_reply_carries_state_report); RUN_TEST(test_nop_reply_zero_state_fresh_standby); /* Task 3.4: ha_stream_wal_to_standby batch WAL streaming */ RUN_TEST(test_wal_stream_500_entries_two_batches); RUN_TEST(test_wal_stream_partial_batch); RUN_TEST(test_wal_stream_empty_range_no_batches); RUN_TEST(test_wal_stream_term_from_index_not_raw_hdr); /* bug-1028 */ /* Task 3.5: snapshot chunk transfer (sender side) */ RUN_TEST(test_snapshot_transfer_two_chunks); RUN_TEST(test_snapshot_transfer_small_less_than_one_chunk); /* Task 3.5c: standby snapshot receive handler */ RUN_TEST(test_snapshot_recv_two_chunks_assembles_and_renames); RUN_TEST(test_snapshot_recv_size_mismatch_fails); RUN_TEST(test_snapshot_recv_null_args_fail); /* Task 6.4-A: 128KB snapshot in 2 chunks — full-size receive path */ RUN_TEST(test_snapshot_recv_128kb_two_chunks); /* Task 3.5d: reconnect discards snap.part */ RUN_TEST(test_snapshot_recv_reconnect_discards_snappart); /* Task 5.1: handle_ctrlhello connection setup only — no repl messages */ RUN_TEST(test_ctrlhello_no_repl_messages_enqueued); #endif /* Task A: raw capnp shims */ RUN_TEST(test_replconfig_req_ssl_encode); RUN_TEST(test_replconfig_reply_raw_encode); RUN_TEST(test_upgrade_config_req_extract_raw); }