diff --git a/.kiro/specs/ha-commit-engine/tasks.md b/.kiro/specs/ha-commit-engine/tasks.md index 20bafa9a..c0baad00 100644 --- a/.kiro/specs/ha-commit-engine/tasks.md +++ b/.kiro/specs/ha-commit-engine/tasks.md @@ -53,7 +53,7 @@ - _Requirements: 3.1, 3.2, 3.3, 6.1, 6.2_ - [ ] 5. Integration: wire the engine behind the job-submit sync path -- [ ] 5.1 Write RED integration tests for the ACK feed and the sync-append path +- [x] 5.1 Write RED integration tests for the ACK feed and the sync-append path - Tests in `test_commit_engine.c`: `repl_on_ack` with a wired engine updates `acked_lsn` AND triggers commit evaluation (majority → `committed_lsn` advances, waiter wakes); `repl_start_catchup`/`repl_complete_catchup`/`repl_remove_standby`/`repl_add_standby` trigger re-evaluation; `persist_wal_sync_append` with wired engine returns 0 only on per-entry majority, returns `-ETIMEDOUT` when standbys' `acked_lsn` covers only an EARLIER entry (stale-ACK regression — connectivity/old ACKs never substitute), leaves `committed_lsn` unmoved on timeout, and for n_peers=1 returns 0 with `committed_lsn` advanced (replaces the old `-ENODEV` degraded path) - Update `testsuite/test_repl_majority.c` walAck cases to wire an engine and assert per-LSN semantics; configAck cases byte-identical - Observable completion: new/updated cases FAIL (red) against unwired production code @@ -88,3 +88,7 @@ - (2.1) WAIT/cap/gate tests (8-16) read `g_ce.inflight_waits` under `g_ce.lock` — task 2.2 must keep that field accurate while waiters block. Timing floors have ~20ms slack vs sleeps (reviewer advisory: watch for flake on loaded boxes). Unity assertions are main-thread-only after joins — keep that discipline in future threaded tests. - (3.1/3.2) Joint completion: task 1.2's commit_on_membership_change (commit.c:334-346) already implemented the full 3.2 contract, making 3.1's RED unachievable without dishonest tests. Tests 17-20 landed green as regression coverage; reviewer confirmed mutation-killing power (each test fails against the relevant mutant) and zero production changes. Lesson for later RED tasks: when a prior task's "minimal" allowance fully implements a later contract, complete the pair jointly under review rather than weakening tests. - (4.1) REVIEWER ADVISORY for 4.2: the gate-side fix (ha_state_t.repl back-pointer + catch-up exclusion through ha_has_write_quorum) has NO RED test — 4.1's tests would all pass even if 4.2 only implemented the helper. 4.2 MUST add a wired gate-level case (n_peers=3, catch-up-only standby → NOT_READY via ha_dispatch_gate) as its own green evidence. Tests 26-28 pin the NULL-back-pointer legacy fallback and n_peers<=1 contracts 4.2 must preserve. Stub repl_active_standby_count returns -1 (repl.c:557) — replace, don't wrap. +- (5.1) **LSN-0 edge for 5.2/5.3 (reviewer-verified)**: commit_submit rejects lsn==0 with -EINVAL (commit.c:229), but the FIRST-EVER WAL entry is LSN 0 (current_lsn starts at 0; written_lsn = current_lsn - 1 at persist.c:1020). When 5.3 passes written_lsn, the first sync append on a fresh cluster would get -EINVAL. 5.3 must handle written_lsn==0 (options: treat entry 0 as valid in the engine, or shift LSN numbering, or special-case — decide per design). Tests 32-34 sidestep via a priming async append; production must not. +- (5.1) persist.c currently passes entry_lsn=0 to the adapter — if 5.2 lands walAck delegation before 5.3 passes written_lsn, the wired sync path returns -EINVAL. Land 5.2's delegation gated on ce!=NULL AND entry_lsn!=0 falling back to legacy, OR land 5.2+5.3 wiring in lockstep. +- (5.1) Baseline failure root causes identified: test_committed_lsn_does_not_advance_on_timeout (test_repl_unit.c:294) expects the OLD contract (rc==0 + sync_rc out-param) but persist_wal_sync_append now rejects-on-timeout — STALE TEST EXPECTATION, fix the test at 6.1, not the code. test_p2_integration.c:test_single_node_committed_lsn_unchanged pins the -ENODEV degraded path that 5.3 replaces — update it at 5.3/6.1. +- (5.1) Repo hygiene (pre-existing, out of scope): untracked src/lib/testsuite/test_wal_wire.c is referenced by committed src/lib/meson.build:176 — should be added to git eventually. diff --git a/.wolf/memory.md b/.wolf/memory.md index 7237973b..be1124c8 100644 --- a/.wolf/memory.md +++ b/.wolf/memory.md @@ -910,3 +910,20 @@ | 2026-06-11 | ha-commit-engine task 4.2: replaced repl_active_standby_count -1 stub with registry-truth count (ep!=NULL && !in_catch_up under r->lock); added ha_state_t.repl back-pointer (ha.h); ha_has_write_quorum uses helper when wired, legacy standby_endpoints fallback when NULL; added wired gate test 29 (RED: Expected -10013 Was 0 → GREEN) | src/controller/repl.c, ha.c, ha.h, testsuite/test_commit_engine.c | 29 tests 0 failures; full census = exactly 3 bug-411 baseline failures unchanged | ~60k | | 16:40 | Adversarial review ha-commit-engine task 4.2 (quorum-accounting fix): boundary clean, mutant run proves test 29 killing power (Expected -10013 Was 0), 29/29 group PASS, full census = exactly 3 bug-411 baseline failures, no sanitizer output — APPROVED | ha.c, ha.h, repl.c, test_commit_engine.c | approved | ~30k | | 15:28 | Edited .kiro/specs/ha-commit-engine/tasks.md | inline fix | ~42 | +| 15:37 | Edited src/include/repl.h | modified void() | ~75 | +| 15:38 | Edited src/include/repl.h | expanded (+8 lines) | ~172 | +| 15:38 | Edited src/controller/repl.c | modified repl_wait_majority_ack() | ~130 | +| 15:38 | Edited src/controller/repl.c | modified repl_set_commit_engine() | ~150 | +| 15:38 | Edited src/controller/config_repl.c | 2→4 lines | ~73 | +| 15:38 | Edited src/controller/persist.c | 2→5 lines | ~99 | +| 15:38 | Edited src/controller/testsuite/test_repl_unit.c | 2→2 lines | ~27 | +| 15:39 | Edited src/controller/testsuite/test_repl_unit.c | 2→2 lines | ~30 | +| 15:39 | Edited src/controller/testsuite/test_commit_engine.c | added error handling | ~269 | +| 15:40 | Edited src/controller/testsuite/test_commit_engine.c | added error handling | ~3260 | +| 15:40 | Edited src/controller/testsuite/test_commit_engine.c | added error handling | ~96 | +| 15:40 | Edited src/controller/testsuite/test_commit_engine.c | expanded (+6 lines) | ~151 | +| 15:42 | Created src/controller/testsuite/test_repl_majority.c | — | ~2989 | +| 5.1-RED | ha-commit-engine task 5.1: RED tests for ACK feed + sync-append path; repl_wait_majority_ack gains entry_lsn param (legacy body byte-identical); inert repl_set_commit_engine stub | test_commit_engine.c, test_repl_majority.c, repl.h, repl.c, config_repl.c, persist.c, test_repl_unit.c | building | ~60k | +| 5.1-RED-done | RED captured: 9 new failures (4 commit_engine #30/31/33/34, 5 repl_majority walAck) + 3 baselines exact; build green; test 32 pins positive contract (green in RED by design) | census /tmp/census_final.log | ok | ~90k | +| 16:08 | Edited .kiro/specs/ha-commit-engine/tasks.md | inline fix | ~22 | +| 16:08 | Edited .kiro/specs/ha-commit-engine/tasks.md | modified hygiene() | ~517 | diff --git a/src/controller/config_repl.c b/src/controller/config_repl.c index 5965f038..c17b77b7 100644 --- a/src/controller/config_repl.c +++ b/src/controller/config_repl.c @@ -913,7 +913,9 @@ int push_config_to_standbys(cluster_t *c, const uint8_t *bytes, size_t blen, /* Req 8.1–8.3: wait for majority-quorum OP_upgradeConfigAck. * On quorum: advance committed_config_tag atomically (Req 8.5). */ - int rc = repl_wait_majority_ack(&c->persist.repl, CONF_PUSH_TIMEOUT_MS, + /* entry_lsn = 0: the configAck branch ignores it (walAck-only adapter + * parameter, ha-commit-engine task 5.1). */ + int rc = repl_wait_majority_ack(&c->persist.repl, 0, CONF_PUSH_TIMEOUT_MS, (uint8_t)OPCode_upgradeConfigAck); if (rc == 0) { atomic_store_explicit(&c->ha.committed_config_tag, new_tag, diff --git a/src/controller/persist.c b/src/controller/persist.c index 2e1f4e7e..82444d5d 100644 --- a/src/controller/persist.c +++ b/src/controller/persist.c @@ -1029,7 +1029,10 @@ int persist_wal_sync_append(persist_t* p, uint32_t type, if (!has_standbys) { sync_err = -ENODEV; } else { - sync_err = repl_wait_majority_ack(&p->repl, p->config.sync_timeout_ms, + /* entry_lsn = 0: mechanical signature update only (ha-commit-engine + * task 5.1) — the legacy body ignores it. Task 5.3 passes + * written_lsn so the wired engine evaluates per-entry majority. */ + sync_err = repl_wait_majority_ack(&p->repl, 0, p->config.sync_timeout_ms, (uint8_t)OPCode_walAck); if (sync_err == 0) atomic_store_explicit(&p->committed_lsn, written_lsn, memory_order_release); diff --git a/src/controller/repl.c b/src/controller/repl.c index 8e2063c2..89b3bc9b 100644 --- a/src/controller/repl.c +++ b/src/controller/repl.c @@ -383,8 +383,15 @@ void repl_on_config_ack(replicator_t *r, repl_ep_t ep, uint64_t acked_config_lsn * Returns 0 on quorum, -ETIMEDOUT on timeout. * ============================================================ */ -int repl_wait_majority_ack(replicator_t *r, uint32_t timeout_ms, uint8_t ack_op) +int repl_wait_majority_ack(replicator_t *r, uint64_t entry_lsn, + uint32_t timeout_ms, uint8_t ack_op) { + /* entry_lsn is accepted but unused by this legacy counting body + * (ha-commit-engine task 5.1 signature change): task 5.2 delegates the + * walAck branch to commit_submit(ce, entry_lsn, COMMIT_MODE_WAIT, ...) + * when the engine is wired. Behavior here stays byte-identical. */ + (void)entry_lsn; + if (r == NULL) return -EINVAL; @@ -556,6 +563,18 @@ int repl_needs_snapshot(replicator_t *r, uint64_t standby_lsn, * (ha.c) for write admission. Takes ONLY r->lock — never the commit-engine * mutex (lock-ordering rule: gate path holds no locks when calling in). * (ha-commit-engine Req 3.1, 3.2, 3.3) */ +/* repl_set_commit_engine — INERT stub (ha-commit-engine task 5.1). + * Stores nothing, so the engine is never wired and the task-5.1 RED tests + * fail on behavior (ACK and membership events never reach the engine), + * not on linking. Task 5.2 adds the replicator_t.ce field, stores the + * pointer here, and feeds commit_on_ack / commit_on_membership_change + * from repl_on_ack and the membership mutators. */ +void repl_set_commit_engine(replicator_t *r, struct commit_engine *ce) +{ + (void)r; + (void)ce; +} + int repl_active_standby_count(replicator_t *r) { int i; diff --git a/src/controller/testsuite/test_commit_engine.c b/src/controller/testsuite/test_commit_engine.c index d47b4453..c2a6756c 100644 --- a/src/controller/testsuite/test_commit_engine.c +++ b/src/controller/testsuite/test_commit_engine.c @@ -42,8 +42,21 @@ * and ha_dispatch_gate write-quorum behavior. RED against the inert repl.c * stub (returns -1); green with the task-4.2 implementation. * - * Requirements: 1.1, 1.2, 2.1, 2.2, 2.3, 3.1, 3.2, 3.3, 4.1, 4.2, 4.3, 4.4, - * 5.1, 6.1, 6.2, 6.3, 7.1, 7.2, 7.3 + * Task 5.1 extends the file with integration tests for the ACK feed and the + * job-submit sync-append path: repl_on_ack with a wired engine + * (repl_set_commit_engine) updates acked_lsn AND triggers commit evaluation; + * the four membership mutators (repl_add_standby / repl_remove_standby / + * repl_start_catchup / repl_complete_catchup) trigger re-evaluation; and + * persist_wal_sync_append over a real persist_t (tmpdir fixture, pattern + * from test_p2_integration.c) honors the POST-wiring contract: 0 only on + * per-entry majority, -ETIMEDOUT on stale ACKs covering only an earlier + * entry, committed_lsn unmoved past the entry on timeout, and n_peers=1 + * returning 0 with committed_lsn advanced (replaces -ENODEV degraded path). + * RED against the inert repl_set_commit_engine stub and the unwired + * production sync-append path; green with tasks 5.2 + 5.3. + * + * Requirements: 1.1, 1.2, 1.3, 2.1, 2.2, 2.3, 3.1, 3.2, 3.3, 4.1, 4.2, 4.3, + * 4.4, 5.1, 6.1, 6.2, 6.3, 7.1, 7.2, 7.3 */ #include #include @@ -62,6 +75,8 @@ #include "ha.h" /* ha_state_t, ha_dispatch_gate fixture */ #include "persist.h" #include "repl.h" +#include "typemap.h" /* wal_catchup_entry_t (task 5.1 send hook) */ +#include "wal_types.h" /* WAL_JOB_SUBMIT (task 5.1 sync-append cases) */ /* ha_dispatch_gate — defined in ha.c (same pattern as test_ha_repl_ack.c) */ int ha_dispatch_gate(ha_state_t *ha, int opcode); @@ -1100,6 +1115,300 @@ static void test_gate_wired_registry_catchup_standby_rejects_then_admits(void) printf("test_gate_wired_registry_catchup_standby_rejects_then_admits: PASS\n"); } +/* ════════════════════════════════════════════════════════════════════════ + * Task 5.1 — integration: ACK feed and the job-submit sync-append path. + * (Req 1.1, 1.2, 1.3, 2.1, 4.2, 4.4, 6.3; design §ACK Feed, + * §Job-Submit Adapter) + * + * Tests 30-31 wire the engine into the stub-fixture replicator via + * repl_set_commit_engine and drive the PRODUCTION entry points + * (repl_on_ack, membership mutators) instead of commit_on_ack directly. + * Tests 32-34 run persist_wal_sync_append against a real persist_t + * (tmpdir fixture, test_p2_integration.c pattern) with an engine wired + * over it, asserting the post-task-5.3 return contract. + * ════════════════════════════════════════════════════════════════════════ */ + +/* Test-injectable send hook (defined in repl.c under XCE_TESTING) — keeps + * repl_broadcast from dereferencing the dummy endpoints. */ +extern int (*repl_broadcast_send_hook)(void *ep, + wal_catchup_entry_t **entries, + int n_entries); + +static int ce_drop_send_hook(void *ep, wal_catchup_entry_t **entries, + int n_entries) +{ + (void)ep; + (void)entries; + (void)n_entries; + return 0; /* pretend success, drop the data */ +} + +/* Heap persist over a private tmpdir (test_p2_integration.c pattern). + * Static pointer: survives Unity's longjmp on assertion failure. */ +static persist_t *g_sp; + +static persist_t *sync_persist_setup(const char *suffix, uint32_t sync_timeout_ms) +{ + persist_t *p; + persist_config_t cfg; + char tmpdir[256]; + + p = calloc(1, sizeof(persist_t)); + TEST_ASSERT_NOT_NULL(p); + + cfg = persist_config_defaults(); + cfg.wal_sync_ms = 5000; + cfg.sync_timeout_ms = sync_timeout_ms; + + snprintf(tmpdir, sizeof(tmpdir), "/tmp/xce_ce51_%s_XXXXXX", suffix); + TEST_ASSERT_NOT_NULL(mkdtemp(tmpdir)); + + persist_set_workdir(p, tmpdir, "ctrl"); + TEST_ASSERT_EQUAL_INT(0, persist_init(p, &cfg, NULL)); + TEST_ASSERT_EQUAL_INT(0, persist_start(p)); + return p; +} + +static void sync_persist_teardown(persist_t *p) +{ + if (p == NULL) { + return; + } + persist_stop(p); + persist_fini(p); + free(p); +} + +/* Background ACK through the PRODUCTION feed (repl_on_ack), not the + * commit_on_ack shortcut — exercises the task-5.2 wiring. */ +typedef struct { + replicator_t *r; + void *ep; + uint64_t lsn; + uint32_t delay_ms; +} repl_ack_arg_t; + +static repl_ack_arg_t g_rack; + +static void *repl_ack_thread_fn(void *arg) +{ + repl_ack_arg_t *a = (repl_ack_arg_t *)arg; + + if (a->delay_ms > 0) { + sleep_ms(a->delay_ms); + } + repl_on_ack(a->r, a->ep, a->lsn); + return NULL; +} + +/* ── Test 30: repl_on_ack with a wired engine feeds commit evaluation ───── + * The production ACK entry point must BOTH update the registry watermark + * AND trigger engine evaluation: majority → committed_lsn advances and the + * pending waiter wakes with success. (Req 1.1, 4.2) */ +static void test_repl_on_ack_wired_updates_watermark_and_commits(void) +{ + uint64_t watermark; + + /* Arrange: 3-member cluster, entry 1 appended; engine wired into the + * replicator the way cluster init will (repl_set_commit_engine) */ + engine_setup(3, 2, 1); + repl_set_commit_engine(&g_p.repl, &g_ce); + append_entries(1); + spawn_waiters(1, 1, 1000); + + /* Act: standby 0 ACKs entry 1 through the PRODUCTION feed */ + sleep_ms(80); + repl_on_ack(&g_p.repl, &g_eps[0], 1); + join_waiters(1); + + /* Assert: the registry watermark was updated (pre-existing contract)... */ + watermark = atomic_load(&g_p.repl.standbys[0].acked_lsn); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(1, watermark, + "repl_on_ack must update the standby's acked_lsn"); + /* ...AND the wired engine evaluated the majority: commit advanced... */ + TEST_ASSERT_EQUAL_UINT64_MESSAGE(1, committed(), + "wired repl_on_ack must trigger commit evaluation (majority -> advance)"); + /* ...and the pending waiter woke with success */ + TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_wargs[0].rc, + "waiter must complete once the wired ACK feed reaches majority"); + + engine_teardown(); + printf("test_repl_on_ack_wired_updates_watermark_and_commits: PASS\n"); +} + +/* ── Test 31: all four membership mutators trigger re-evaluation ────────── + * For each of repl_add_standby / repl_start_catchup / repl_complete_catchup + * / repl_remove_standby: a majority-covering watermark is planted DIRECTLY + * in the registry (no ACK feed), so committed_lsn can only advance if the + * mutator itself triggers commit_on_membership_change re-evaluation. + * (Req 4.2) */ +static void test_membership_mutators_trigger_commit_reevaluation(void) +{ + /* Arrange: 3-member cluster, entries 1..4 appended, ONE standby (sb0) + * registered and wired; sb0's watermark alone forms majority with self */ + engine_setup(3, 1, 4); + repl_set_commit_engine(&g_p.repl, &g_ce); + append_entries(4); + + /* Act + Assert per mutator. Plant the watermark silently, confirm no + * advancement happened, then attribute the advance to the mutator. */ + + /* repl_add_standby (standby connect) re-evaluates */ + atomic_store(&g_p.repl.standbys[0].acked_lsn, 1); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(0, committed(), + "planted watermark alone must not advance committed_lsn"); + TEST_ASSERT_TRUE(repl_add_standby(&g_p.repl, &g_eps[1], "sb1", 7001) >= 0); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(1, committed(), + "repl_add_standby must trigger commit re-evaluation"); + + /* repl_start_catchup (catch-up entry) re-evaluates */ + atomic_store(&g_p.repl.standbys[0].acked_lsn, 2); + TEST_ASSERT_EQUAL_UINT64(1, committed()); + repl_start_catchup(&g_p.repl, &g_eps[1], 0); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(2, committed(), + "repl_start_catchup must trigger commit re-evaluation"); + + /* repl_complete_catchup (catch-up completion) re-evaluates */ + atomic_store(&g_p.repl.standbys[0].acked_lsn, 3); + TEST_ASSERT_EQUAL_UINT64(2, committed()); + repl_complete_catchup(&g_p.repl, &g_eps[1]); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(3, committed(), + "repl_complete_catchup must trigger commit re-evaluation"); + + /* repl_remove_standby (disconnect) re-evaluates */ + atomic_store(&g_p.repl.standbys[0].acked_lsn, 4); + TEST_ASSERT_EQUAL_UINT64(3, committed()); + repl_remove_standby(&g_p.repl, &g_eps[1]); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(4, committed(), + "repl_remove_standby must trigger commit re-evaluation"); + + engine_teardown(); + printf("test_membership_mutators_trigger_commit_reevaluation: PASS\n"); +} + +/* ── Test 32: sync append returns 0 on per-entry majority (wired) ───────── + * Contract-pinning positive case: a standby watermark covering THE entry + * commits it — return 0, sync status 0, committed_lsn at the entry. + * (Req 1.1, 1.2) */ +static int g_ep_sa, g_ep_sb; + +static void test_sync_append_wired_commits_on_per_entry_majority(void) +{ + pthread_t th; + int rc, sync_rc; + + /* Arrange: real persist, 2 standbys, engine n_peers=3 wired; one + * priming async append so the sync entry lands at LSN 1 */ + g_sp = sync_persist_setup("maj", 2000); + TEST_ASSERT_TRUE(repl_add_standby(&g_sp->repl, &g_ep_sa, "sb_a", 7100) >= 0); + TEST_ASSERT_TRUE(repl_add_standby(&g_sp->repl, &g_ep_sb, "sb_b", 7101) >= 0); + TEST_ASSERT_EQUAL_INT(0, commit_engine_init(&g_ce, g_sp, 3, 8)); + repl_set_commit_engine(&g_sp->repl, &g_ce); + + repl_broadcast_send_hook = ce_drop_send_hook; + TEST_ASSERT_EQUAL_INT(0, persist_wal_async(g_sp, WAL_JOB_UPDATE, "p", 1)); + + /* Act: standby A's ACK covering the sync entry (LSN 1) arrives mid-wait */ + g_rack = (repl_ack_arg_t){ &g_sp->repl, &g_ep_sa, 1, 30 }; + TEST_ASSERT_EQUAL_INT(0, + pthread_create(&th, NULL, repl_ack_thread_fn, &g_rack)); + sync_rc = -1; + rc = persist_wal_sync_append(g_sp, WAL_JOB_SUBMIT, "job", 3, &sync_rc); + pthread_join(th, NULL); + repl_broadcast_send_hook = NULL; + + /* Assert: per-entry majority → confirmed, commit index at the entry */ + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, sync_rc, + "sync status must be 0 on per-entry majority"); + TEST_ASSERT_EQUAL_UINT64_MESSAGE( + 1, atomic_load(&g_sp->committed_lsn), + "committed_lsn must advance to the majority-acked entry"); + + commit_engine_fini(&g_ce); + sync_persist_teardown(g_sp); + g_sp = NULL; + printf("test_sync_append_wired_commits_on_per_entry_majority: PASS\n"); +} + +/* ── Test 33: stale-ACK regression — earlier-entry ACKs never commit ────── + * Both standbys' watermarks cover only an EARLIER entry. The dishonest + * `acked_lsn > 0` counting confirms the new entry off those stale ACKs; + * the engine contract is -ETIMEDOUT with committed_lsn unmoved past the + * entry: connectivity / old ACKs never substitute for per-entry + * acknowledgment. (Req 1.2, 1.3, 2.1, 4.4) */ +static void test_sync_append_wired_stale_acks_time_out(void) +{ + int rc, sync_rc; + + /* Arrange: real persist (short sync timeout), 2 standbys, engine + * n_peers=3 wired; entries 0..1 appended and BOTH standbys ACK + * watermark 1 BEFORE the new sync entry exists */ + g_sp = sync_persist_setup("stale", 60); + TEST_ASSERT_TRUE(repl_add_standby(&g_sp->repl, &g_ep_sa, "sb_a", 7110) >= 0); + TEST_ASSERT_TRUE(repl_add_standby(&g_sp->repl, &g_ep_sb, "sb_b", 7111) >= 0); + TEST_ASSERT_EQUAL_INT(0, commit_engine_init(&g_ce, g_sp, 3, 8)); + repl_set_commit_engine(&g_sp->repl, &g_ce); + + repl_broadcast_send_hook = ce_drop_send_hook; + TEST_ASSERT_EQUAL_INT(0, persist_wal_async(g_sp, WAL_JOB_UPDATE, "p", 1)); + TEST_ASSERT_EQUAL_INT(0, persist_wal_async(g_sp, WAL_JOB_UPDATE, "q", 1)); + repl_on_ack(&g_sp->repl, &g_ep_sa, 1); + repl_on_ack(&g_sp->repl, &g_ep_sb, 1); + + /* Act: sync append entry LSN 2 — no ACK for it ever arrives */ + sync_rc = 0; + rc = persist_wal_sync_append(g_sp, WAL_JOB_SUBMIT, "job", 3, &sync_rc); + repl_broadcast_send_hook = NULL; + + /* Assert: stale ACKs (watermark 1 > 0) must NOT confirm entry 2... */ + TEST_ASSERT_EQUAL_INT_MESSAGE(-ETIMEDOUT, rc, + "stale earlier-entry ACKs must never confirm a new entry"); + TEST_ASSERT_EQUAL_INT(-ETIMEDOUT, sync_rc); + /* ...and the commit index never advances past the unacked entry */ + TEST_ASSERT_TRUE_MESSAGE(atomic_load(&g_sp->committed_lsn) < 2, + "committed_lsn must stay unmoved past the entry on timeout"); + + commit_engine_fini(&g_ce); + sync_persist_teardown(g_sp); + g_sp = NULL; + printf("test_sync_append_wired_stale_acks_time_out: PASS\n"); +} + +/* ── Test 34: n_peers=1 → sync append succeeds with committed advanced ──── + * Standalone self-quorum replaces the old `-ENODEV` degraded path: the + * single-member primary confirms locally (sync status 0, NOT -ENODEV) and + * committed_lsn advances to the entry. (Req 6.3) */ +static void test_sync_append_wired_single_member_commits_locally(void) +{ + int rc, sync_rc; + + /* Arrange: real persist, NO standbys, engine n_peers=1 wired; one + * priming async append so the sync entry lands at LSN 1 */ + g_sp = sync_persist_setup("solo", 2000); + TEST_ASSERT_EQUAL_INT(0, commit_engine_init(&g_ce, g_sp, 1, 8)); + repl_set_commit_engine(&g_sp->repl, &g_ce); + TEST_ASSERT_EQUAL_INT(0, persist_wal_async(g_sp, WAL_JOB_UPDATE, "p", 1)); + + /* Act */ + sync_rc = -1; + rc = persist_wal_sync_append(g_sp, WAL_JOB_SUBMIT, "job", 3, &sync_rc); + + /* Assert: local confirmation, no degraded -ENODEV signal... */ + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, sync_rc, + "n_peers=1 must self-commit (0), not report -ENODEV degraded mode"); + /* ...and the standalone commit index advances to the entry */ + TEST_ASSERT_EQUAL_UINT64_MESSAGE( + 1, atomic_load(&g_sp->committed_lsn), + "n_peers=1 sync append must advance committed_lsn to the entry"); + + commit_engine_fini(&g_ce); + sync_persist_teardown(g_sp); + g_sp = NULL; + printf("test_sync_append_wired_single_member_commits_locally: PASS\n"); +} + /* ── Test group entry point ──────────────────────────────────────────────── */ void test_commit_engine(void) @@ -1138,5 +1447,11 @@ void test_commit_engine(void) RUN_TEST(test_gate_reaccepts_after_reconnect_without_reset); /* Task 4.2 — wired gate: registry truth at ha_dispatch_gate */ RUN_TEST(test_gate_wired_registry_catchup_standby_rejects_then_admits); + /* Task 5.1 — integration: ACK feed + job-submit sync-append path */ + RUN_TEST(test_repl_on_ack_wired_updates_watermark_and_commits); + RUN_TEST(test_membership_mutators_trigger_commit_reevaluation); + RUN_TEST(test_sync_append_wired_commits_on_per_entry_majority); + RUN_TEST(test_sync_append_wired_stale_acks_time_out); + RUN_TEST(test_sync_append_wired_single_member_commits_locally); UNITY_END(); } diff --git a/src/controller/testsuite/test_repl_majority.c b/src/controller/testsuite/test_repl_majority.c index 5141abd4..1b890521 100644 --- a/src/controller/testsuite/test_repl_majority.c +++ b/src/controller/testsuite/test_repl_majority.c @@ -1,14 +1,24 @@ /* - * test_repl_majority.c — RED-phase tests for majority-commit replication. + * test_repl_majority.c — majority-commit replication tests. * * Tests for: - * repl_wait_majority_ack(replicator_t *, uint32_t timeout_ms, uint8_t ack_op) + * repl_wait_majority_ack(replicator_t *, uint64_t entry_lsn, + * uint32_t timeout_ms, uint8_t ack_op) * repl_broadcast majority semantics (pending_ack_count, over-limit eviction) * - * These functions/fields do not exist yet (task 7.2, 7.3). - * This file will fail to link until they are implemented. + * ha-commit-engine task 5.1: the walAck cases are wired to a commit engine + * (stub persist_t fixture + repl_set_commit_engine, pattern from + * test_commit_engine.c) and assert PER-LSN semantics — entry N is confirmed + * only when self + a majority of active standbys have acked_lsn >= N, and + * 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. * - * Requirements: 5.1, 5.4, 5.5, 5.8, 8.2 + * 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) */ #include #include @@ -18,31 +28,66 @@ #include #include #include -#include "repl.h" +#include "commit.h" #include "config.h" #include "controller.capnp.h" /* OPCode_walAck, OPCode_upgradeConfigAck */ +#include "persist.h" +#include "repl.h" /* MAX_PENDING_ACKS: defined in repl.h after task 7.2; fallback for RED phase */ #ifndef MAX_PENDING_ACKS #define MAX_PENDING_ACKS 16 #endif -/* ── Forward declarations of functions-under-test (not yet implemented) ───── */ - -/* - * repl_wait_majority_ack: wait until majority of active standbys have ACKed - * for the given ack_op. Returns 0 on quorum, -ETIMEDOUT on timeout. - * Single-node (no active standbys) returns 0 immediately. - */ -extern int repl_wait_majority_ack(replicator_t *r, uint32_t timeout_ms, - uint8_t ack_op); - /* ── Helpers ───────────────────────────────────────────────────────────────── */ /* Dummy endpoints — repl.c only compares pointers */ static int ep0, ep1, ep2; -/* ACK a standby from another thread */ +/* ── Engine-wired fixture (ha-commit-engine task 5.1) ────────────────────── + * Stub persist_t whose live parts are committed_lsn, current_lsn, and the + * embedded replicator registry (test_commit_engine.c pattern). Statics + * survive Unity's longjmp on assertion failure. */ +static persist_t g_mp; +static commit_engine_t g_meng; + +static int *g_meps[3] = { &ep0, &ep1, &ep2 }; + +/* Arrange a fresh wired fixture: n_peers cluster members (self counted), + * n_standbys connected standbys, entries 1..appended_lsn appended locally + * (persist invariant: current_lsn = appended_lsn + 1). */ +static void majority_setup(int n_peers, int n_standbys, uint64_t appended_lsn) +{ + int i; + char name[16]; + + memset(&g_mp, 0, sizeof(g_mp)); + TEST_ASSERT_EQUAL_INT(0, repl_init(&g_mp.repl)); + for (i = 0; i < n_standbys; i++) { + snprintf(name, sizeof(name), "sb%d", i); + TEST_ASSERT_EQUAL_INT( + i, repl_add_standby(&g_mp.repl, g_meps[i], name, + 7000 + (uint32_t)i)); + } + g_mp.current_lsn = appended_lsn + 1; + atomic_store(&g_mp.committed_lsn, 0); + + TEST_ASSERT_EQUAL_INT(0, commit_engine_init(&g_meng, &g_mp, n_peers, 8)); + repl_set_commit_engine(&g_mp.repl, &g_meng); +} + +static void majority_teardown(void) +{ + commit_engine_fini(&g_meng); + repl_fini(&g_mp.repl); +} + +static uint64_t m_committed(void) +{ + return atomic_load(&g_mp.committed_lsn); +} + +/* ACK a standby from another thread (production feed: repl_on_ack) */ struct ack_args { replicator_t *r; void *ep; @@ -62,122 +107,138 @@ static void *ack_thread(void *arg) return NULL; } -/* ── Test 1: 2/3 ACKs → majority, returns 0 ───────────────────────────────── */ +/* ── Test 1: per-entry majority — 2/3 standbys ACK the entry → 0 ────────── + * 4-member cluster (self + 3): entry 1 confirmed when 2 standbys' ACK + * watermarks cover LSN 1; confirmation advances committed_lsn through the + * wired engine. (ha-commit-engine Req 1.1, 1.2) */ /* Static so ack_args survive Unity longjmp if assert fires before join */ -static replicator_t g_r1; static struct ack_args g_a1_0, g_a1_1; static void test_wait_majority_ack_two_of_three_returns_zero(void) { - repl_init(&g_r1); - repl_add_standby(&g_r1, &ep0, "sb0", 7000); - repl_add_standby(&g_r1, &ep1, "sb1", 7001); - repl_add_standby(&g_r1, &ep2, "sb2", 7002); + majority_setup(4, 3, 1); - /* 2 standbys ACK — majority of 3 active standbys is ceil((3+1)/2) = 2 ACKs + self = quorum */ + /* 2 of 3 standbys ACK watermark 1 — self + 2 = 3 >= majority(3) of 4 */ pthread_t t0, t1; - g_a1_0 = (struct ack_args){ &g_r1, &ep0, 1, 10 }; - g_a1_1 = (struct ack_args){ &g_r1, &ep1, 1, 20 }; + g_a1_0 = (struct ack_args){ &g_mp.repl, &ep0, 1, 10 }; + g_a1_1 = (struct ack_args){ &g_mp.repl, &ep1, 1, 20 }; pthread_create(&t0, NULL, ack_thread, &g_a1_0); pthread_create(&t1, NULL, ack_thread, &g_a1_1); - int rc = repl_wait_majority_ack(&g_r1, 2000, (uint8_t)OPCode_walAck); + int rc = repl_wait_majority_ack(&g_mp.repl, 1, 2000, (uint8_t)OPCode_walAck); /* Join threads before asserting — prevents use-after-return if assert longjmps */ pthread_join(t0, NULL); pthread_join(t1, NULL); TEST_ASSERT_EQUAL_INT(0, rc); + /* Per-LSN semantics: confirmation means the ENGINE committed entry 1 */ + TEST_ASSERT_EQUAL_UINT64_MESSAGE(1, m_committed(), + "majority confirmation must advance committed_lsn to the entry"); - repl_fini(&g_r1); + majority_teardown(); printf("test_wait_majority_ack_two_of_three_returns_zero: PASS\n"); } -/* ── Test 2: timeout before majority ACKs → -ETIMEDOUT ────────────────────── */ - -static void test_wait_majority_ack_timeout_returns_etimedout(void) +/* ── Test 2: stale earlier-entry ACKs → -ETIMEDOUT (per-LSN regression) ─── + * Standbys' watermarks cover only entry 1; waiting on entry 2 must time + * out — `acked_lsn > 0` ("ACKed anything ever") counting would dishonestly + * confirm it. committed_lsn never moves past the unacked entry. + * (ha-commit-engine Req 1.2, 1.3, 2.1, 4.4) */ +static void test_wait_majority_ack_stale_acks_time_out(void) { - replicator_t r; - repl_init(&r); - repl_add_standby(&r, &ep0, "sb0", 7000); - repl_add_standby(&r, &ep1, "sb1", 7001); - repl_add_standby(&r, &ep2, "sb2", 7002); - /* No ACKs arrive — should time out */ + majority_setup(4, 3, 2); - int rc = repl_wait_majority_ack(&r, 50, (uint8_t)OPCode_walAck); - TEST_ASSERT_EQUAL_INT(-ETIMEDOUT, rc); + /* Both ACKs cover only entry 1 — stale relative to entry 2 */ + repl_on_ack(&g_mp.repl, &ep0, 1); + repl_on_ack(&g_mp.repl, &ep1, 1); - repl_fini(&r); - printf("test_wait_majority_ack_timeout_returns_etimedout: PASS\n"); + int rc = repl_wait_majority_ack(&g_mp.repl, 2, 100, (uint8_t)OPCode_walAck); + TEST_ASSERT_EQUAL_INT_MESSAGE(-ETIMEDOUT, rc, + "earlier-entry ACK watermarks must never confirm a later entry"); + TEST_ASSERT_TRUE_MESSAGE(m_committed() < 2, + "committed_lsn must stay unmoved past the unacked entry"); + + majority_teardown(); + printf("test_wait_majority_ack_stale_acks_time_out: PASS\n"); } -/* ── Test 3: single-node (no standbys) → returns 0 immediately ────────────── */ - +/* ── Test 3: single-member cluster → 0 immediately, committed advances ──── + * n_peers=1 self-quorum: local confirmation through the engine advances + * committed_lsn to the entry. (ha-commit-engine Req 6.3) */ static void test_wait_majority_ack_no_standbys_returns_zero(void) { - replicator_t r; - repl_init(&r); - /* No standbys added — primary alone is majority */ + majority_setup(1, 0, 1); - int rc = repl_wait_majority_ack(&r, 100, (uint8_t)OPCode_walAck); + int rc = repl_wait_majority_ack(&g_mp.repl, 1, 100, (uint8_t)OPCode_walAck); TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(1, m_committed(), + "single-member self-quorum must advance committed_lsn to the entry"); - repl_fini(&r); + majority_teardown(); printf("test_wait_majority_ack_no_standbys_returns_zero: PASS\n"); } -/* ── Test 4: in_catch_up standby excluded from quorum denominator ──────────── */ - -static replicator_t g_r4; +/* ── Test 4: in_catch_up standby's ACK never counts toward the entry ────── + * 3-member cluster: the catch-up standby's covering watermark is excluded; + * only the active standby's ACK (self + 1 = majority of 3) confirms. + * (ha-commit-engine Req 1.2, 4.2) */ static struct ack_args g_a4_0; static void test_wait_majority_ack_catchup_excluded_from_quorum(void) { - repl_init(&g_r4); - repl_add_standby(&g_r4, &ep0, "sb0", 7000); /* active */ - repl_add_standby(&g_r4, &ep1, "sb1", 7001); /* will be put in catch-up */ - repl_add_standby(&g_r4, &ep2, "sb2", 7002); /* active */ + majority_setup(3, 2, 1); - repl_start_catchup(&g_r4, &ep1, 0); + /* sb1 enters catch-up, then ACKs a covering watermark — excluded */ + repl_start_catchup(&g_mp.repl, &ep1, 0); + repl_on_ack(&g_mp.repl, &ep1, 1); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(0, m_committed(), + "catch-up standby's ACK must not commit the entry"); + /* sb0 (active) ACKs — self + sb0 reach majority */ pthread_t t0; - g_a4_0 = (struct ack_args){ &g_r4, &ep0, 1, 10 }; + g_a4_0 = (struct ack_args){ &g_mp.repl, &ep0, 1, 10 }; pthread_create(&t0, NULL, ack_thread, &g_a4_0); - int rc = repl_wait_majority_ack(&g_r4, 2000, (uint8_t)OPCode_walAck); + int rc = repl_wait_majority_ack(&g_mp.repl, 1, 2000, (uint8_t)OPCode_walAck); pthread_join(t0, NULL); TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(1, m_committed(), + "active-standby majority must advance committed_lsn to the entry"); - repl_fini(&g_r4); + majority_teardown(); printf("test_wait_majority_ack_catchup_excluded_from_quorum: PASS\n"); } -/* ── Test 5: OP_walAck vs OP_upgradeConfigAck dispatched independently ─────── */ - -static replicator_t g_r5; +/* ── 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). */ static struct ack_args g_a5_0; static void test_wait_majority_ack_ack_op_discriminates(void) { - repl_init(&g_r5); - repl_add_standby(&g_r5, &ep0, "sb0", 7000); + majority_setup(2, 1, 1); pthread_t t0; - g_a5_0 = (struct ack_args){ &g_r5, &ep0, 1, 10 }; + g_a5_0 = (struct ack_args){ &g_mp.repl, &ep0, 1, 10 }; pthread_create(&t0, NULL, ack_thread, &g_a5_0); - int rc_wal = repl_wait_majority_ack(&g_r5, 2000, (uint8_t)OPCode_walAck); + int rc_wal = repl_wait_majority_ack(&g_mp.repl, 1, 2000, + (uint8_t)OPCode_walAck); pthread_join(t0, NULL); TEST_ASSERT_EQUAL_INT(0, rc_wal); + 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_r5, 50, (uint8_t)OPCode_upgradeConfigAck); + int rc_cfg = repl_wait_majority_ack(&g_mp.repl, 0, 50, + (uint8_t)OPCode_upgradeConfigAck); TEST_ASSERT_EQUAL_INT(-ETIMEDOUT, rc_cfg); - repl_fini(&g_r5); + majority_teardown(); printf("test_wait_majority_ack_ack_op_discriminates: PASS\n"); } @@ -230,7 +291,7 @@ void test_repl_majority(void) { UNITY_BEGIN(); RUN_TEST(test_wait_majority_ack_two_of_three_returns_zero); - RUN_TEST(test_wait_majority_ack_timeout_returns_etimedout); + RUN_TEST(test_wait_majority_ack_stale_acks_time_out); RUN_TEST(test_wait_majority_ack_no_standbys_returns_zero); RUN_TEST(test_wait_majority_ack_catchup_excluded_from_quorum); RUN_TEST(test_wait_majority_ack_ack_op_discriminates); diff --git a/src/controller/testsuite/test_repl_unit.c b/src/controller/testsuite/test_repl_unit.c index c69ca7c2..a7f2514e 100644 --- a/src/controller/testsuite/test_repl_unit.c +++ b/src/controller/testsuite/test_repl_unit.c @@ -390,7 +390,7 @@ static void test_wal_ack_reply_updates_acked_lsn_and_signals(void) /* repl_wait_majority_ack: 1 active standby, quorum=(1+2)/2=1, * acked_total = 1(self) + 1(standby) = 2 >= 1 → returns 0. */ - int rc = repl_wait_majority_ack(&r, 200, (uint8_t)OPCode_walAck); + int rc = repl_wait_majority_ack(&r, 0, 200, (uint8_t)OPCode_walAck); pthread_join(th, NULL); TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, @@ -453,7 +453,7 @@ static void test_wal_nack_does_not_update_acked_lsn(void) * quorum (1), so repl_wait_majority_ack returns 0, not -ETIMEDOUT. * A timeout would only occur if quorum required at least one standby ACK, * which only happens when n_active >= 2 (e.g. 2+ standbys registered). */ - int rc = repl_wait_majority_ack(&r, 50, (uint8_t)OPCode_walAck); + int rc = repl_wait_majority_ack(&r, 0, 50, (uint8_t)OPCode_walAck); TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, "repl_wait_majority_ack must return 0 with 1 standby: " "self-count(1) >= quorum(1) even without standby ACK (Req 4.3)"); diff --git a/src/include/repl.h b/src/include/repl.h index f11e0962..1cf10ca3 100644 --- a/src/include/repl.h +++ b/src/include/repl.h @@ -18,6 +18,10 @@ /* Endpoint is an opaque pointer — full type in netio.h (avoid heavy include) */ typedef void* repl_ep_t; +/* Commit engine (controller/include/commit.h) — forward-declared so repl.h + * keeps zero controller-header dependencies (ha-commit-engine task 5). */ +struct commit_engine; + /* Per-standby connection slot */ typedef struct { repl_ep_t ep; /* NULL = slot empty */ @@ -70,8 +74,16 @@ void repl_on_config_ack(replicator_t *r, repl_ep_t ep, uint64_t acked_config_lsn /* Sync replication (stub in P2-1; wired in P2-3) */ int repl_wait_ack(replicator_t *r, uint64_t target_lsn, uint32_t timeout_ms); -/* Majority-quorum ACK wait (task 7.3) */ -int repl_wait_majority_ack(replicator_t *r, uint32_t timeout_ms, uint8_t ack_op); +/* Majority-quorum ACK wait (task 7.3; ha-commit-engine task 5 adapter). + * entry_lsn: the WAL entry the caller waits on. The walAck branch delegates + * to the commit engine for per-LSN evaluation when wired (task 5.2); the + * legacy counting body ignores it. configAck callers pass 0 (ignored). */ +int repl_wait_majority_ack(replicator_t *r, uint64_t entry_lsn, + uint32_t timeout_ms, uint8_t ack_op); + +/* Wire the commit engine into the replicator so ACK and membership events + * are fed to it (ha-commit-engine task 5.2; NULL = not wired). */ +void repl_set_commit_engine(replicator_t *r, struct commit_engine *ce); /* WAL trim guard */ uint64_t repl_trim_lsn(replicator_t *r, uint64_t checkpoint_lsn, uint32_t snap_interval_s);