diff --git a/.kiro/specs/ha-commit-barrier-rollout/tasks.md b/.kiro/specs/ha-commit-barrier-rollout/tasks.md index d973c96b..e6ceef9e 100644 --- a/.kiro/specs/ha-commit-barrier-rollout/tasks.md +++ b/.kiro/specs/ha-commit-barrier-rollout/tasks.md @@ -66,7 +66,7 @@ - _Requirements: 3.1, 3.3, 3.4_ - [ ] 5. Configuration change commit barrier -- [ ] 5.1 Write RED tests for configAck arithmetic and the upgrade failure path +- [x] 5.1 Write RED tests for configAck arithmetic and the upgrade failure path - Unit tests (repl.c configAck branch via `repl_wait_majority_ack`): a standby's `acked_config_lsn` from a previous tag does not count toward the current push; n_peers=3 with zero covering acks → `-ETIMEDOUT` (the `n_active == 0 → success` short-circuit dies on this branch); self + one covering standby (majority of n_peers=3) → 0; n_peers=1 → immediate 0; update `test_repl_majority.c` configAck cases to the per-push arithmetic - Integration tests (config_repl.c): `push_config_to_standbys` returns `-ETIMEDOUT` with acks withheld and 0 with majority acks, push count via out-param; on failure the `upgradeConfig` requester receives an error reply (send_upgrade_error shape), `committed_config_tag` is unmoved, and `restart_requested` stays clear; on success the existing reply + restart behavior is unchanged - Observable completion: new cases FAIL (red) against the current logged-only result (config_repl.c:916-929) and unconditional success ACK (config_repl.c:1062-1073) diff --git a/.wolf/cerebrum.md b/.wolf/cerebrum.md index 22be7e3a..287ca937 100644 --- a/.wolf/cerebrum.md +++ b/.wolf/cerebrum.md @@ -9,6 +9,10 @@ ## Key Learnings + +- [2026-06-12] Census parsing caveat: controller_unit_test logs (tinylog) and Unity result lines BOTH go to stdout and can splice mid-line under the multi-threaded config-push tests — a distinct-Unity-line count silently drops spliced results. Robust census = union of Unity result names AND the `test_x: PASS` printf lines, cross-checked against the per-group `N Tests M Failures` summary blocks (which are splice-immune). +- [2026-06-12] Config-barrier test trick: when a test needs a covering config ACK for a tag only known AFTER conf_save_config hashes the re-encoded blob (upgradeConfig path), pre-ack with acked_config_lsn = UINT64_MAX — covers any push tag under both legacy (>0) and per-push (>= tag) counting. +- [2026-06-12] A zeroed stack cbuf_t DEADLOCKS on cbuf_put (ncap==0 → len==ncap → cond_wait forever); any fixture driving do_upgrade_config_with_bytes' restart path must cbuf_init(&c->wbuf, …) first. - Census EXACT method (2026-06-12): count DISTINCT Unity `file:line:test:PASS` triples (`grep -aoE` + sort -u), not raw `2mPASS` occurrences — tinylog interleaving garbles ~9 verdict lines per run and duplicates others; distinct-triple count matched the 494-baseline arithmetic exactly (495 after task 4.3's +1). - repl registry reconnect semantics: repl_remove_standby vacates the slot (ep=NULL, keeps last_disconnected_lsn guard); repl_add_standby reuses the first empty slot and memsets it (acked_lsn=0). With both standbys removed, majority_candidate_locked zero-fills to n_peers-1 → candidate 0 → committed_lsn frozen (RETAIN entries safely pending); after re-add, ONE covering ack = 2-of-3 majority. - repl_wait_majority_ack walAck track is engine-only since barrier-rollout 2.2: entry_lsn==0 returns 0 immediately (documented once-per-lifetime LSN-0 first entry), everything else delegates to commit_submit unconditionally — ce==NULL surfaces as -EINVAL from commit_submit (unreachable in production; cluster_init_phase2 always wires the engine). The counting loop that remains in the function is configAck-ONLY (no self-count, n_active==0 shortcut is Req 8.5 config behavior). persist_wal_sync_append's residual non-engine branch can only store committed_lsn for the LSN-0 no-op case. (2026-06-12, task 2.2) diff --git a/.wolf/memory.md b/.wolf/memory.md index 1092926c..20f16b36 100644 --- a/.wolf/memory.md +++ b/.wolf/memory.md @@ -962,3 +962,18 @@ | 08:03 | Edited src/controller/testsuite/test_commit_barrier_reply.c | 2→5 lines | ~72 | | 15:15 | Task 4.3 ha-commit-barrier-rollout: fact-retention-across-quorum-loss integration test (GREEN on landing) — quorum lost via repl_remove_standby x2, 4 worker facts via on_job_start/on_job_finish (cluster.c/do_notify not in test link), RETAIN pending + on-disk exactly-once scan, reconnect + covering ACK advances committed_lsn, no re-delivery, retained-depth=4 printed | src/controller/testsuite/test_commit_barrier_reply.c | census 495 distinct Unity PASS (494+1), ZERO FAIL | ~55k | | 08:29 | Edited .kiro/specs/ha-commit-barrier-rollout/tasks.md | inline fix | ~30 | +| 08:43 | Edited src/controller/config_repl.h | modified acked_out() | ~244 | +| 08:43 | Edited src/controller/config_repl.c | added 1 condition(s) | ~219 | +| 08:43 | Edited src/controller/config_repl.c | added 1 condition(s) | ~57 | +| 08:43 | Edited src/controller/config_repl.c | 2→2 lines | ~40 | +| 08:44 | Edited src/controller/ha.c | 2→3 lines | ~59 | +| 08:44 | Edited src/controller/join.c | 2→2 lines | ~27 | +| 08:45 | Edited src/controller/testsuite/test_repl_majority.c | modified arithmetic() | ~158 | +| 08:45 | Edited src/controller/testsuite/test_repl_majority.c | modified entry_lsn() | ~121 | +| 08:45 | Edited src/controller/testsuite/test_repl_majority.c | 4→5 lines | ~90 | +| 08:49 | Edited src/controller/testsuite/test_commit_barrier_rollout.c | expanded (+6 lines) | ~339 | +| 08:50 | Edited src/controller/testsuite/test_commit_barrier_rollout.c | added 2 condition(s) | ~5302 | +| 08:51 | Edited src/controller/testsuite/test_commit_barrier_rollout.c | expanded (+10 lines) | ~210 | +| 08:51 | Edited src/controller/testsuite/test_commit_barrier_rollout.c | 3→8 lines | ~109 | +| 16:10 | Task 5.1 ha-commit-barrier-rollout: RED tests for configAck per-push arithmetic + upgrade failure path; behavior-preserving push_config_to_standbys signature change (+int *acked_out, return semantics unchanged) | src/controller/{config_repl.c,config_repl.h,ha.c,join.c}, testsuite/{test_commit_barrier_rollout.c,test_repl_majority.c} | 8 new cases (6 RED + 2 pins), isolated group 29 tests/6 fails, full suite: only those 6 fail (summary blocks), repl_majority green, prod controller builds | ~110k | +| 09:31 | Edited .kiro/specs/ha-commit-barrier-rollout/tasks.md | inline fix | ~22 | diff --git a/src/controller/config_repl.c b/src/controller/config_repl.c index c17b77b7..8bdc5456 100644 --- a/src/controller/config_repl.c +++ b/src/controller/config_repl.c @@ -864,8 +864,15 @@ int wait_all_with_timeout_for_test(pthread_t *threads, push_ctx_t *ctxs, #endif int push_config_to_standbys(cluster_t *c, const uint8_t *bytes, size_t blen, - const uint8_t sha256[32], uint64_t new_tag) + const uint8_t sha256[32], uint64_t new_tag, + int *acked_out) { + /* acked_out: per-standby transport push count (task 5.1 signature; + * behavior-preserving — the return value keeps the legacy push-count + * semantics until task 5.2 re-plumbs it to the barrier result). */ + if (acked_out) + *acked_out = 0; + if (!c->ha.standby_endpoints) return 0; @@ -911,6 +918,9 @@ int push_config_to_standbys(cluster_t *c, const uint8_t *bytes, size_t blen, free(ctxs); free(threads); + if (acked_out) + *acked_out = acked; + /* Req 8.1–8.3: wait for majority-quorum OP_upgradeConfigAck. * On quorum: advance committed_config_tag atomically (Req 8.5). */ /* entry_lsn = 0: the configAck branch ignores it (walAck-only adapter @@ -1057,7 +1067,7 @@ static int do_upgrade_config_with_bytes(cluster_t *c, endpoint_t *ep, /* 7. Push to standbys (best-effort; see push_config_to_standbys comment) */ int acked = push_config_to_standbys(c, out_bytes, out_len, computed, - c->myconf->config_version_tag); + c->myconf->config_version_tag, NULL); free(out_bytes); /* 8. ACK the admin tool */ diff --git a/src/controller/config_repl.h b/src/controller/config_repl.h index 7a10b04a..6f850159 100644 --- a/src/controller/config_repl.h +++ b/src/controller/config_repl.h @@ -112,12 +112,17 @@ int handle_repl_config_push(cluster_t *c, endpoint_t *ep, xce_request_t *req); * push_config_to_standbys: fan-out a new config blob to all connected standby * controllers in parallel. Best-effort — standbys that miss the push will * pull on reconnect via handle_config_request(). + * acked_out (may be NULL): receives the number of standbys that successfully + * received the push at the transport level (per-standby push count). * Returns the number of standbys that successfully received the push. + * (ha-commit-barrier-rollout task 5.2 moves the majority-ACK barrier result + * into the return value; the push count is then only available via acked_out.) * Only meaningful when CONF_ENABLE_UPGRADE_CONFIG is set, but always declared * so callers can reference it without an #if guard. */ int push_config_to_standbys(cluster_t *c, const uint8_t *bytes, size_t blen, - const uint8_t sha256[32], uint64_t new_tag); + const uint8_t sha256[32], uint64_t new_tag, + int *acked_out); /* * handle_upgrade_config: handle an OP_upgradeConfig request from an admin tool. diff --git a/src/controller/ha.c b/src/controller/ha.c index 80da7a5c..4912f862 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -75,7 +75,8 @@ typedef struct { extern int cluster_init_phase2(cluster_t *c); extern void ha_send_ctrlhello_to_all_standbys(cluster_t *c); extern int push_config_to_standbys(cluster_t *c, const uint8_t *bytes, size_t blen, - const uint8_t sha256[32], uint64_t new_tag); + const uint8_t sha256[32], uint64_t new_tag, + int *acked_out); #endif static void on_vote_connected(endpoint_t *ep); @@ -568,11 +569,12 @@ static void raft_cb_become_leader(void *ctx, uint64_t term) conf_read_raw(c->myconf->conffile, &cfg_bytes, &cfg_len) == 0) { push_config_to_standbys(c, cfg_bytes, cfg_len, c->myconf->config_sha256, - c->myconf->config_version_tag); + c->myconf->config_version_tag, NULL); free(cfg_bytes); } else { push_config_to_standbys(c, NULL, 0, NULL, - c->myconf ? c->myconf->config_version_tag : 0); + c->myconf ? c->myconf->config_version_tag : 0, + NULL); } } #endif @@ -1407,11 +1409,12 @@ int ha_dispatch_repl_action_for_standby(ha_state_t *ha, conf_read_raw(c->myconf->conffile, &cfg_bytes, &cfg_len) == 0) { push_config_to_standbys(c, cfg_bytes, cfg_len, c->myconf->config_sha256, - c->myconf->config_version_tag); + c->myconf->config_version_tag, NULL); free(cfg_bytes); } else { push_config_to_standbys(c, NULL, 0, NULL, - c->myconf ? c->myconf->config_version_tag : 0); + c->myconf ? c->myconf->config_version_tag : 0, + NULL); } #endif } @@ -2280,7 +2283,7 @@ apply_role: ha_push_config_hook(c); #else if (c != NULL) - push_config_to_standbys(c, NULL, 0, NULL, 0); + push_config_to_standbys(c, NULL, 0, NULL, 0, NULL); #endif /* Send CtrlHelloRequest to all standbys to trigger WAL catch-up (Req 7.2) */ diff --git a/src/controller/join.c b/src/controller/join.c index 6d2b4650..ee08f7cb 100644 --- a/src/controller/join.c +++ b/src/controller/join.c @@ -184,7 +184,7 @@ static int handle_controller_join(cluster_t* c, endpoint_t* ep, /* Immediately push current config to the newly-joined standby so it * doesn't have to wait for the next scheduled push. */ if (ha_is_primary(&c->ha)) - push_config_to_standbys(c, NULL, 0, NULL, 0); + push_config_to_standbys(c, NULL, 0, NULL, 0, NULL); } return 0; diff --git a/src/controller/testsuite/test_commit_barrier_rollout.c b/src/controller/testsuite/test_commit_barrier_rollout.c index fd20d6b8..784f6711 100644 --- a/src/controller/testsuite/test_commit_barrier_rollout.c +++ b/src/controller/testsuite/test_commit_barrier_rollout.c @@ -77,11 +77,17 @@ * CTRL_PRIMARY_COMMITTING is intentionally NOT tested (deleted wholesale * by ha-single-leader-role). * + * Task 5.1 adds the config-barrier cases (Req 4.1-4.4) — configAck + * per-push arithmetic, push_config_to_standbys barrier-result contract, + * and the upgradeConfig requester failure/success paths; see the task-5.1 + * section banner below for the full RED rationale. + * * Requirements: 1.1, 1.2, 1.3 (task 2.1); 2.1, 2.2, 3.1 (task 1.1); - * 3.2 (task 4.1) + * 3.2 (task 4.1); 4.1, 4.2, 4.3, 4.4 (task 5.1) */ #include #include +#include /* SHA256 (task 5.1 config blob hashing) */ #include #include #include @@ -90,16 +96,21 @@ #include #include #include +#include /* mkdir (task 5.1 upgrade fixture) */ #include #include #include #include +#include /* raw reply parsing (task 5.1) */ + #include "xce/cluster_def.h" /* HOOK_DETAIL_*, XCE_CONTROLLER_HOOK_JOB */ #include "xce/errnox.h" /* XCE_ERR_NOT_READY (task 2.1 passthrough) */ #include "xce/types.h" /* job_t */ #include "commit.h" +#include "config_repl.h" /* push_config_to_standbys, handle_upgrade_config + * (+ netio.h, request.h, myconf.h, clusterconfig.h) */ #include "controller.capnp.h" /* OPCode_* (task 4.1 gate cases) */ #include "ha.h" /* ha_state_t, ha_config_t, zarray (task 4.1) */ #include "hook.h" /* HOOK_AFTER_EVENT */ @@ -987,6 +998,500 @@ static void test_gate_with_quorum_admits_all_opcodes(void) printf("test_gate_with_quorum_admits_all_opcodes: PASS\n"); } +/* ════════════════════════════════════════════════════════════════════════ + * Task 5.1 — config barrier: configAck per-push arithmetic + upgrade + * failure path (Req 4.1, 4.2, 4.3, 4.4; design §Config Barrier). + * + * UNIT half (repl.c configAck branch via repl_wait_majority_ack): the + * hardened branch must take the in-flight push tag through the EXISTING + * entry_lsn parameter (added by ha-commit-engine; no second parameter) + * and count standbys with acked_config_lsn >= push_tag; majority = + * 1 (self) + count > n_peers / 2 over configured members (engine + * n_peers, the fixture's wired commit_engine_init value); n_peers <= 1 → + * immediate success. RED carriers against the current branch + * (repl.c:444-510, re-verified this session): it ignores entry_lsn, + * counts acked_config_lsn > 0 ("ACKed anything ever"), uses denominator + * n_active, and short-circuits n_active == 0 → success. + * + * INTEGRATION half (config_repl.c): push_config_to_standbys must return + * the barrier result (-ETIMEDOUT with acks withheld / 0 with majority) + * with the transport push count via the acked_out out-param (the task-5.1 + * behavior-preserving signature change keeps the legacy push-count RETURN + * until task 5.2 — both rc assertions are RED against it), and the + * upgradeConfig requester must observe the barrier outcome: on failure an + * error reply in the send_upgrade_error shape (status < 0, nodata) with + * committed_config_tag unmoved and restart_requested clear (RED against + * the unconditional success ACK + restart, config_repl.c:1073-1106 + * post-edit anchors); on success the existing reply + restart behavior is + * unchanged (PIN — both standbys pre-ack with UINT64_MAX so the case is + * green under both the legacy counting and the per-push arithmetic). + * + * Fixtures: the unit cases reuse barrier_setup (stub persist_t + repl + * registry + wired engine — the engine carries n_peers for the configAck + * denominator). The integration cases use a minimal cluster_t with stack + * standby endpoints (zeroed, no iomgr — pushes and replies stay on + * ep->outq, the test_config_repl.c:2210 / test_commit_barrier_reply.c + * capture precedent) and parse replies with the raw capnp reader. + * ════════════════════════════════════════════════════════════════════════ */ + +/* ── Unit: configAck per-push arithmetic (Req 4.1, 4.2) ─────────────────── */ + +/* A standby's acked_config_lsn from a PREVIOUS push tag must not count + * toward the current push: both standbys acked tag 0x50, the in-flight + * push is 0x100 → no covering ack → -ETIMEDOUT. RED: the current branch + * counts acked_config_lsn > 0, sees 2 >= quorum 2 and returns 0. */ +static void test_configack_previous_tag_ack_does_not_count(void) +{ + int rc; + + /* Arrange: n_peers=3, both standbys acked an OLDER push tag */ + barrier_setup(3, 2, 0); + repl_on_config_ack(&g_bp.repl, &g_bep_a, 0x50); + repl_on_config_ack(&g_bp.repl, &g_bep_b, 0x50); + + /* Act: wait on push tag 0x100 (entry_lsn carries the push tag) */ + rc = repl_wait_majority_ack(&g_bp.repl, 0x100, 150, + (uint8_t)OPCode_upgradeConfigAck); + + /* Assert */ + TEST_ASSERT_EQUAL_INT_MESSAGE(-ETIMEDOUT, rc, + "a previous push's config ack must not count toward the current tag"); + + barrier_teardown(); + printf("test_configack_previous_tag_ack_does_not_count: PASS\n"); +} + +/* n_peers=3 with ZERO covering acks (here: zero connected standbys) must + * time out — a 3-member cluster can never reach majority on self alone. + * RED: the current `n_active == 0 → success` short-circuit returns 0. */ +static void test_configack_no_active_standbys_times_out(void) +{ + int rc; + + /* Arrange: n_peers=3, no standby connected at all */ + barrier_setup(3, 0, 0); + + /* Act */ + rc = repl_wait_majority_ack(&g_bp.repl, 0x100, 150, + (uint8_t)OPCode_upgradeConfigAck); + + /* Assert */ + TEST_ASSERT_EQUAL_INT_MESSAGE(-ETIMEDOUT, rc, + "n_peers=3 with zero covering config acks must never succeed " + "(the n_active == 0 success short-circuit must die on this branch)"); + + barrier_teardown(); + printf("test_configack_no_active_standbys_times_out: PASS\n"); +} + +/* Self + one covering standby is a majority of n_peers=3 (1 + 1 = 2 > 1). + * RED: the current branch uses denominator n_active with NO self-count — + * quorum (2+2)/2 = 2 vs acked 1 → it times out. */ +static void test_configack_self_plus_one_covering_is_majority_of_three(void) +{ + int rc; + + /* Arrange: n_peers=3, two standbys, one acks the in-flight tag */ + barrier_setup(3, 2, 0); + repl_on_config_ack(&g_bp.repl, &g_bep_a, 0x100); + + /* Act */ + rc = repl_wait_majority_ack(&g_bp.repl, 0x100, 400, + (uint8_t)OPCode_upgradeConfigAck); + + /* Assert */ + TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, + "self + one covering standby must satisfy majority of n_peers=3"); + + barrier_teardown(); + printf("test_configack_self_plus_one_covering_is_majority_of_three: PASS\n"); +} + +/* n_peers=1: single-member cluster — immediate success without consuming + * the timeout. PIN: green today (via the n_active == 0 short-circuit) and + * green after task 5.2 (explicit n_peers <= 1 rule). */ +static void test_configack_single_member_immediate_success(void) +{ + int rc; + uint64_t t0, elapsed_ms; + + /* Arrange */ + barrier_setup(1, 0, 0); + + /* Act */ + t0 = mono_ms(); + rc = repl_wait_majority_ack(&g_bp.repl, 0x100, 500, + (uint8_t)OPCode_upgradeConfigAck); + elapsed_ms = mono_ms() - t0; + + /* Assert */ + TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, + "single-member cluster must confirm a config push immediately"); + TEST_ASSERT_TRUE_MESSAGE(elapsed_ms < 200, + "single-member success must not consume the ack timeout"); + + barrier_teardown(); + printf("test_configack_single_member_immediate_success: PASS\n"); +} + +/* ── Integration fixture: minimal cluster_t for the config push path ────── */ + +/* Statics survive Unity's longjmp on assertion failure. */ +static cluster_t g_cb5_c; +static commit_engine_t g_cb5_eng; +static endpoint_t g_cb5_sb_ep[2]; /* standby push endpoints */ +static ep_private_t g_cb5_sb_priv[2]; +static endpoint_t g_cb5_req_ep; /* upgradeConfig requester */ +static ep_private_t g_cb5_req_priv; +static zarray_t g_cb5_standbys; +static myconf_t g_cb5_myconf; +static char g_cb5_tmpdir[256]; +static int g_cb5_have_tmpdir; + +static void cb5_setup(void) +{ + int i; + + memset(&g_cb5_c, 0, sizeof(g_cb5_c)); + memset(g_cb5_sb_ep, 0, sizeof(g_cb5_sb_ep)); + memset(g_cb5_sb_priv, 0, sizeof(g_cb5_sb_priv)); + memset(&g_cb5_req_ep, 0, sizeof(g_cb5_req_ep)); + memset(&g_cb5_req_priv, 0, sizeof(g_cb5_req_priv)); + memset(&g_cb5_myconf, 0, sizeof(g_cb5_myconf)); + g_cb5_have_tmpdir = 0; + + g_cb5_c.myhost = &g_cb5_c.myhost_info; + g_cb5_c.myhost_info.name = (char *)"ctrl-cb5"; + g_cb5_c.myhost_info.key = (char *)""; + + /* Zeroed stack endpoints, no iomgr: sends stay queued on ep->outq + * (endpoint_update_epoll no-ops; zeroed pthread mutexes are valid). */ + g_cb5_req_ep.fd = -1; + g_cb5_req_ep.data = &g_cb5_req_priv; + + TEST_ASSERT_EQUAL_INT(0, zarray_init(&g_cb5_standbys, 4, NULL)); + for (i = 0; i < 2; i++) { + g_cb5_sb_ep[i].fd = -1; + g_cb5_sb_ep[i].data = &g_cb5_sb_priv[i]; + zarray_store(&g_cb5_standbys, &g_cb5_sb_ep[i]); + } + g_cb5_c.ha.standby_endpoints = &g_cb5_standbys; + + /* Replicator registry mirrors the push endpoints; wired engine carries + * n_peers=3 for the configAck majority denominator. */ + TEST_ASSERT_EQUAL_INT(0, repl_init(&g_cb5_c.persist.repl)); + TEST_ASSERT_TRUE(repl_add_standby(&g_cb5_c.persist.repl, + &g_cb5_sb_ep[0], "cb5_a", 7500) >= 0); + TEST_ASSERT_TRUE(repl_add_standby(&g_cb5_c.persist.repl, + &g_cb5_sb_ep[1], "cb5_b", 7501) >= 0); + TEST_ASSERT_EQUAL_INT(0, + commit_engine_init(&g_cb5_eng, &g_cb5_c.persist, 3, 8)); + repl_set_commit_engine(&g_cb5_c.persist.repl, &g_cb5_eng); + + /* The current do_upgrade_config_with_bytes restart path does + * cbuf_put(&c->wbuf, …) — must be a real cbuf or it blocks forever. */ + TEST_ASSERT_EQUAL_INT(0, cbuf_init(&g_cb5_c.wbuf, 4, sizeof(int))); +} + +/* Upgrade-path extension: tmpdir-backed myconf so conf_save_config works + * (it writes to //conf — the parent must exist). */ +static void cb5_setup_upgrade(void) +{ + char inst[300]; + + cb5_setup(); + snprintf(g_cb5_tmpdir, sizeof(g_cb5_tmpdir), "/tmp/xce_cb51_XXXXXX"); + TEST_ASSERT_NOT_NULL(mkdtemp(g_cb5_tmpdir)); + g_cb5_have_tmpdir = 1; + snprintf(inst, sizeof(inst), "%s/ctrl", g_cb5_tmpdir); + TEST_ASSERT_EQUAL_INT(0, mkdir(inst, 0700)); + + g_cb5_myconf.workdir = g_cb5_tmpdir; + g_cb5_myconf.name = (char *)"ctrl"; + g_cb5_myconf.conf_keep_n = 2; + g_cb5_c.myconf = &g_cb5_myconf; +} + +/* Drain and free every packet queued on an endpoint's outq. */ +static void cb5_drain_ep(endpoint_t *ep) +{ + data_item_t *it; + + while ((it = endpoint_deq(ep)) != NULL) { + io_packet_t *pkt = ITPTR_TO_PACKET(it); + + free_packet(pkt); + free(pkt); + } +} + +static void cb5_teardown(void) +{ + cb5_drain_ep(&g_cb5_sb_ep[0]); + cb5_drain_ep(&g_cb5_sb_ep[1]); + cb5_drain_ep(&g_cb5_req_ep); + commit_engine_fini(&g_cb5_eng); + repl_fini(&g_cb5_c.persist.repl); + zarray_fini(&g_cb5_standbys); + cbuf_fini(&g_cb5_c.wbuf); + free(g_cb5_myconf.conffile); + g_cb5_myconf.conffile = NULL; + if (g_cb5_have_tmpdir) { + char cmd[300]; + + snprintf(cmd, sizeof(cmd), "rm -rf %s", g_cb5_tmpdir); + system(cmd); + g_cb5_have_tmpdir = 0; + } +} + +/* ── Reply observation (raw capnp parse of ep->outq packets) ────────────── */ + +typedef struct { + int n_replies; /* replymsg packets seen */ + int32_t last_status; /* status of the last replymsg observed */ + int last_data_which; /* RequestReply_data_which discriminant */ +} cb5_reply_obs_t; + +static void cb5_scan_replies(endpoint_t *ep, cb5_reply_obs_t *obs) +{ + data_item_t *it; + + while ((it = endpoint_deq(ep)) != NULL) { + io_packet_t *pkt = ITPTR_TO_PACKET(it); + struct capn c; + struct Request rq; + Request_ptr rp; + + capn_init_mem(&c, (const uint8_t *)pkt->data, pkt->sz, 0); + rp.p = capn_getp(capn_root(&c), 0, 1); + read_Request(&rq, rp); + if (rq.body_which == Request_body_replymsg) { + struct RequestReply rr; + + read_RequestReply(&rr, rq.body.replymsg); + obs->n_replies += 1; + obs->last_status = rr.status; + obs->last_data_which = (int)rr.data_which; + } + capn_free(&c); + free_packet(pkt); + free(pkt); + } +} + +/* Build a minimal valid serialised ClusterConfig blob (caller frees). */ +static uint8_t *cb5_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 *)"cb5-upgrade"; + cfg.port = 9990; + 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; +} + +/* upgradeConfig request scaffolding (statics survive Unity longjmp). + * handle_upgrade_config reads the raw_* fields xce_decode would have + * populated; the test owns the buffers, so xce_free_request is NOT used. */ +static upgrade_config_req_t g_cb5_ucr; +static xce_request_t g_cb5_req; + +static void cb5_arrange_upgrade_req(uint8_t *blob, size_t blen, + uint8_t sha[32]) +{ + memset(&g_cb5_ucr, 0, sizeof(g_cb5_ucr)); + memset(&g_cb5_req, 0, sizeof(g_cb5_req)); + g_cb5_ucr.raw_sha256 = sha; + g_cb5_ucr.raw_bytes = blob; + g_cb5_ucr.raw_bytes_len = blen; + g_cb5_req.opcode = OPCode_upgradeConfig; + g_cb5_req.seq = 1; + g_cb5_req.body_which = Request_body_upgradeconfigrequestmsg; + g_cb5_req.body.upgradeconfigrequestmsg = &g_cb5_ucr; +} + +/* ── push_config_to_standbys: barrier result + push count (Req 4.2, 4.3) ── */ + +/* ACKs withheld: the push barrier must report -ETIMEDOUT (return value), + * the transport push count via acked_out, and the tag must never commit. + * RED: the current return value IS the push count (2), and the internal + * majority wait only logs its timeout. */ +static void test_push_config_withheld_acks_returns_etimedout(void) +{ + static const uint8_t blob[] = "cb5-push-blob"; + uint8_t sha[32]; + int pushed = -1; + int rc; + + /* Arrange */ + cb5_setup(); + SHA256(blob, sizeof(blob) - 1, sha); + + /* Act: no standby ever acks tag 0xAB12 */ + rc = push_config_to_standbys(&g_cb5_c, blob, sizeof(blob) - 1, sha, + 0xAB12, &pushed); + + /* Assert */ + TEST_ASSERT_EQUAL_INT_MESSAGE(-ETIMEDOUT, rc, + "withheld config ACKs must surface as -ETIMEDOUT from the push (Req 4.2)"); + TEST_ASSERT_EQUAL_INT_MESSAGE(2, pushed, + "acked_out must report the per-standby transport push count"); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(0, + atomic_load_explicit(&g_cb5_c.ha.committed_config_tag, + memory_order_acquire), + "an unacked config must never become committed (Req 4.3)"); + + cb5_teardown(); + printf("test_push_config_withheld_acks_returns_etimedout: PASS\n"); +} + +/* Majority acks (self + 1 covering standby of n_peers=3): the push must + * return 0 and commit the tag. RED twice over: the return value is the + * push count (2 != 0), and the legacy internal wait (quorum 2 over + * n_active, no self-count) times out so the tag never advances. */ +static void test_push_config_majority_acks_returns_zero_and_commits(void) +{ + static const uint8_t blob[] = "cb5-push-blob"; + uint8_t sha[32]; + int pushed = -1; + int rc; + + /* Arrange: one standby has already acked the in-flight tag */ + cb5_setup(); + SHA256(blob, sizeof(blob) - 1, sha); + repl_on_config_ack(&g_cb5_c.persist.repl, &g_cb5_sb_ep[0], 0xAB12); + + /* Act */ + rc = push_config_to_standbys(&g_cb5_c, blob, sizeof(blob) - 1, sha, + 0xAB12, &pushed); + + /* Assert */ + TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, + "majority config ACK must surface as success from the push (Req 4.1)"); + TEST_ASSERT_EQUAL_INT_MESSAGE(2, pushed, + "acked_out must report the per-standby transport push count"); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(0xAB12, + atomic_load_explicit(&g_cb5_c.ha.committed_config_tag, + memory_order_acquire), + "majority-acked config must advance committed_config_tag (Req 4.1)"); + + cb5_teardown(); + printf("test_push_config_majority_acks_returns_zero_and_commits: PASS\n"); +} + +/* ── upgradeConfig requester: failure and success paths (Req 4.1-4.4) ───── */ + +/* Barrier failure (ACKs withheld) must reach the requester as an error + * reply in the send_upgrade_error shape (status < 0, nodata), leave + * committed_config_tag unmoved, and keep restart_requested clear. + * RED: today the handler unconditionally ACKs status=0 with the + * upgradeConfigReply payload and requests the restart. */ +static void test_upgrade_config_barrier_failure_error_reply_no_restart(void) +{ + uint8_t *blob; + size_t blen = 0; + uint8_t sha[32]; + cb5_reply_obs_t obs = { 0 }; + + /* Arrange */ + cb5_setup_upgrade(); + blob = cb5_make_config_blob(&blen); + SHA256(blob, blen, sha); + cb5_arrange_upgrade_req(blob, blen, sha); + + /* Act: no standby ever acks the new tag */ + (void)handle_upgrade_config(&g_cb5_c, &g_cb5_req_ep, &g_cb5_req); + + /* Assert: error reply, no commit, no restart */ + cb5_scan_replies(&g_cb5_req_ep, &obs); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, obs.n_replies, + "the requester must receive exactly one reply"); + TEST_ASSERT_TRUE_MESSAGE(obs.last_status < 0, + "a failed config commit must reach the requester as an error reply" + " (Req 4.2)"); + TEST_ASSERT_EQUAL_INT_MESSAGE((int)RequestReply_data_nodata, + obs.last_data_which, + "the failure reply must use the send_upgrade_error shape (nodata)"); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(0, + atomic_load_explicit(&g_cb5_c.ha.committed_config_tag, + memory_order_acquire), + "a failed upgrade must leave committed_config_tag unmoved (Req 4.3)"); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, atomic_load(&g_cb5_c.restart_requested), + "a failed upgrade must not request the activation restart (Req 4.4)"); + + free(blob); + cb5_teardown(); + printf("test_upgrade_config_barrier_failure_error_reply_no_restart: PASS\n"); +} + +/* Success path PIN: with a full majority of covering ACKs the existing + * behavior is unchanged — status-0 reply with the upgradeConfigReply + * payload, committed_config_tag at the saved tag, restart requested. + * Both standbys pre-ack UINT64_MAX (covers any tag — the new tag is only + * known after conf_save_config hashes the re-encoded blob), so the case + * is green under both the legacy counting and the per-push arithmetic. */ +static void test_upgrade_config_majority_ack_success_unchanged(void) +{ + uint8_t *blob; + size_t blen = 0; + uint8_t sha[32]; + int rc; + cb5_reply_obs_t obs = { 0 }; + + /* Arrange: full covering pre-ack from both standbys */ + cb5_setup_upgrade(); + blob = cb5_make_config_blob(&blen); + SHA256(blob, blen, sha); + cb5_arrange_upgrade_req(blob, blen, sha); + repl_on_config_ack(&g_cb5_c.persist.repl, &g_cb5_sb_ep[0], UINT64_MAX); + repl_on_config_ack(&g_cb5_c.persist.repl, &g_cb5_sb_ep[1], UINT64_MAX); + + /* Act */ + rc = handle_upgrade_config(&g_cb5_c, &g_cb5_req_ep, &g_cb5_req); + + /* Assert: existing success reply + restart behavior unchanged */ + TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, + "a majority-acked upgrade must succeed"); + cb5_scan_replies(&g_cb5_req_ep, &obs); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, obs.n_replies, + "the requester must receive exactly one reply"); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, obs.last_status, + "the success ACK must carry status 0 (Req 4.1)"); + TEST_ASSERT_EQUAL_INT_MESSAGE((int)RequestReply_data_upgradeconfigreplymsg, + obs.last_data_which, + "the success ACK must carry the upgradeConfigReply payload"); + TEST_ASSERT_TRUE_MESSAGE(g_cb5_myconf.config_version_tag != 0, + "conf_save_config must have derived the new version tag"); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(g_cb5_myconf.config_version_tag, + atomic_load_explicit(&g_cb5_c.ha.committed_config_tag, + memory_order_acquire), + "a majority-acked config must be committed at the saved tag (Req 4.1)"); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&g_cb5_c.restart_requested), + "a successful upgrade must request the activation restart"); + + free(blob); + cb5_teardown(); + printf("test_upgrade_config_majority_ack_success_unchanged: PASS\n"); +} + /* ── Test group entry point ──────────────────────────────────────────────── */ void test_commit_barrier_rollout(void) @@ -1016,5 +1521,15 @@ void test_commit_barrier_rollout(void) RUN_TEST(test_gate_no_quorum_admits_clrpreason_fact); RUN_TEST(test_gate_no_quorum_still_rejects_request_opcodes); RUN_TEST(test_gate_with_quorum_admits_all_opcodes); + /* Task 5.1 — config barrier: configAck arithmetic + upgrade failure + * path (Req 4.1, 4.2, 4.3, 4.4) */ + RUN_TEST(test_configack_previous_tag_ack_does_not_count); + RUN_TEST(test_configack_no_active_standbys_times_out); + RUN_TEST(test_configack_self_plus_one_covering_is_majority_of_three); + RUN_TEST(test_configack_single_member_immediate_success); + RUN_TEST(test_push_config_withheld_acks_returns_etimedout); + RUN_TEST(test_push_config_majority_acks_returns_zero_and_commits); + RUN_TEST(test_upgrade_config_barrier_failure_error_reply_no_restart); + RUN_TEST(test_upgrade_config_majority_ack_success_unchanged); UNITY_END(); } diff --git a/src/controller/testsuite/test_repl_majority.c b/src/controller/testsuite/test_repl_majority.c index 1b890521..57692ff3 100644 --- a/src/controller/testsuite/test_repl_majority.c +++ b/src/controller/testsuite/test_repl_majority.c @@ -13,9 +13,16 @@ * confirmation advances committed_lsn through the engine. They are RED * against the unwired legacy counting (`acked_lsn > 0`, n_active * denominator, zero-active success short-circuit) and go green with tasks - * 5.2/5.3. The configAck expectations are untouched (config barrier is - * owned by ha-commit-barrier-rollout); only the mechanical entry_lsn=0 - * argument was added to the call. + * 5.2/5.3. + * + * ha-commit-barrier-rollout task 5.1: the configAck case now follows the + * per-push arithmetic (design §Config Barrier): the call passes the config + * PUSH TAG as entry_lsn (no longer a vacuous 0 — the hardened branch counts + * standbys with acked_config_lsn >= push_tag, majority 1 (self) + count > + * n_peers / 2 over configured members). The expectation below (-ETIMEDOUT + * with zero covering acks) holds under BOTH the legacy counting and the + * per-push arithmetic; the RED carriers for the arithmetic change live in + * test_commit_barrier_rollout.c (task 5.1 section). * * Requirements: ha-commit-engine 1.1, 1.2, 2.1, 4.2, 4.4, 6.3 * (original repl coverage: 5.1, 5.4, 5.5, 5.8, 8.2) @@ -213,8 +220,11 @@ static void test_wait_majority_ack_catchup_excluded_from_quorum(void) } /* ── Test 5: OP_walAck vs OP_upgradeConfigAck dispatched independently ───── - * walAck half is engine-wired per-LSN; the configAck expectation is - * byte-identical legacy behavior (config barrier owned downstream). */ + * walAck half is engine-wired per-LSN; the configAck half passes a real + * push tag as entry_lsn (per-push arithmetic, barrier-rollout task 5.1): + * n_peers=2, one active standby with NO covering config ack → timeout under + * both the legacy counting (acked_config_lsn 0, not > 0) and the hardened + * arithmetic (count(acked_config_lsn >= tag) = 0; 1 + 0 > 2/2 is false). */ static struct ack_args g_a5_0; static void test_wait_majority_ack_ack_op_discriminates(void) @@ -233,8 +243,9 @@ static void test_wait_majority_ack_ack_op_discriminates(void) TEST_ASSERT_EQUAL_UINT64_MESSAGE(1, m_committed(), "walAck confirmation must advance committed_lsn per-LSN"); - /* upgradeConfigAck with no pending ACKs → timeout */ - int rc_cfg = repl_wait_majority_ack(&g_mp.repl, 0, 50, + /* upgradeConfigAck for push tag 0xC0FFEE with no covering config ACKs + * → timeout (entry_lsn carries the push tag — per-push arithmetic) */ + int rc_cfg = repl_wait_majority_ack(&g_mp.repl, 0xC0FFEE, 50, (uint8_t)OPCode_upgradeConfigAck); TEST_ASSERT_EQUAL_INT(-ETIMEDOUT, rc_cfg);