feat(ha-commit-barrier-rollout): config barrier with production ACK ingestion (task 5.2 + r1)
configAck branch: per-push majority (acked_config_lsn >= push_tag, 1+count > n_peers/2 over r->ce->n_peers, n_peers<=1 immediate, ce==NULL -EINVAL); legacy >0 counting and n_active==0 shortcut dead. push_config_to_standbys returns the barrier result; upgrade failure -> send_upgrade_error (rc forwarded), no success ACK, no restart, committed_config_tag success-only. Remediation r1 (two CRITICAL review findings): - F1: wire production config-ACK ingestion — replconfigpushreplymsg decode branch in repl_ack_decode_and_dispatch -> repl_on_config_ack on the promoted-vote ep; + defensive cluster.c OPCode_reply case. Closes the unlanded half of ha-config-bootstrap 8.1; without this the barrier would brick every multi-member config upgrade. No schema change. - F2: keep the barrier off the iomgr thread — export push_config_transport_only (non-blocking) for the 5 ha.c + join.c streaming sites; barrier path now vtask-only (do_upgrade_config). Wire-path test drives real capnp ACK bytes through the production decode branch. 30/30 rollout, 6/6 repl_majority, 46/46 config_repl; census 0 FAIL; pre-existing fixture leak only (less than baseline).
This commit is contained in:
parent
9cdeba5816
commit
5a2fae1d37
12 changed files with 1034 additions and 399 deletions
|
|
@ -1492,6 +1492,175 @@ static void test_upgrade_config_majority_ack_success_unchanged(void)
|
|||
printf("test_upgrade_config_majority_ack_success_unchanged: PASS\n");
|
||||
}
|
||||
|
||||
/* ════════════════════════════════════════════════════════════════════════
|
||||
* Task 5.2 remediation r1 — wire-path config ACK ingestion (Req 4.1, 4.2).
|
||||
*
|
||||
* The barrier consumes acked_config_lsn, which only repl_on_config_ack
|
||||
* advances — so the standby's OP_replConfigPush reply
|
||||
* (replconfigpushreplymsg, config_repl.c:744-752) must be decoded on the
|
||||
* primary and fed to repl_on_config_ack. This case drives the ACK over
|
||||
* the WIRE: it builds the standby's capnp reply bytes and pushes them
|
||||
* through peer_ctrl_msg_handler → repl_ack_decode_and_dispatch (the
|
||||
* production decode entry for promoted peer endpoints), asserting the
|
||||
* decoded tag lands in the replicator slot and completes a pending
|
||||
* barrier. This also completes the unlanded half of ha-config-bootstrap
|
||||
* task 8.1 (production ingestion of the config-push ACK), which until
|
||||
* now was only exercised by direct repl_on_config_ack injection.
|
||||
* ════════════════════════════════════════════════════════════════════════ */
|
||||
|
||||
/* ha.c test seams (test_ha_connection_reuse.c precedent) */
|
||||
extern int peer_ctrl_msg_handler_for_test(endpoint_t *ep, io_packet_t *pkt);
|
||||
extern endpoint_t *vote_ep_ctx_build_for_test(endpoint_t *ep,
|
||||
ha_state_t *ha,
|
||||
ha_elect_ctx_t *ctx,
|
||||
int peer_idx,
|
||||
uint64_t my_term,
|
||||
uint64_t my_tag,
|
||||
uint64_t my_lsn,
|
||||
const char *my_name);
|
||||
extern void vote_ep_ctx_set_cluster_for_test(endpoint_t *ep, cluster_t *c);
|
||||
|
||||
/* Build the standby's config-push ACK exactly as xce_reply() puts it on
|
||||
* the wire (config_repl.c:744-752): Request{opcode=reply,
|
||||
* body.replymsg=RequestReply{data.replconfigpushreplymsg=
|
||||
* ReplConfigPushReply{accepted, versionTag}}}. The handler takes
|
||||
* ownership (free_packet + free). */
|
||||
static io_packet_t *cb5_build_config_ack_pkt(int accepted, uint64_t tag)
|
||||
{
|
||||
struct capn cs;
|
||||
struct capn_segment *seg;
|
||||
struct ReplConfigPushReply ack_capn;
|
||||
ReplConfigPushReply_ptr ack_ptr;
|
||||
struct RequestReply rr_capn;
|
||||
RequestReply_ptr rr_ptr;
|
||||
struct Request rq_capn;
|
||||
Request_ptr rq_ptr;
|
||||
io_packet_t *pkt;
|
||||
|
||||
capn_init_malloc(&cs);
|
||||
seg = capn_root(&cs).seg;
|
||||
|
||||
memset(&ack_capn, 0, sizeof(ack_capn));
|
||||
ack_capn.accepted = accepted ? 1 : 0;
|
||||
ack_capn.versionTag = tag;
|
||||
ack_ptr = new_ReplConfigPushReply(seg);
|
||||
write_ReplConfigPushReply(&ack_capn, ack_ptr);
|
||||
|
||||
memset(&rr_capn, 0, sizeof(rr_capn));
|
||||
rr_capn.seq = 1;
|
||||
rr_capn.opcode = OPCode_replConfigPush;
|
||||
rr_capn.status = 0;
|
||||
rr_capn.data_which = RequestReply_data_replconfigpushreplymsg;
|
||||
rr_capn.data.replconfigpushreplymsg = ack_ptr;
|
||||
rr_ptr = new_RequestReply(seg);
|
||||
write_RequestReply(&rr_capn, rr_ptr);
|
||||
|
||||
memset(&rq_capn, 0, sizeof(rq_capn));
|
||||
rq_capn.opcode = OPCode_reply;
|
||||
rq_capn.seq = 1;
|
||||
rq_capn.body_which = Request_body_replymsg;
|
||||
rq_capn.body.replymsg = rr_ptr;
|
||||
rq_ptr = new_Request(seg);
|
||||
write_Request(&rq_capn, rq_ptr);
|
||||
capn_setp(capn_root(&cs), 0, rq_ptr.p);
|
||||
|
||||
pkt = calloc(1, sizeof(*pkt));
|
||||
TEST_ASSERT_NOT_NULL_MESSAGE(pkt, "config ACK pkt alloc");
|
||||
pkt->sz = capn_size(&cs);
|
||||
TEST_ASSERT_TRUE_MESSAGE(alloc_packet(pkt) >= 0, "config ACK pkt data alloc");
|
||||
capn_write_mem(&cs, (uint8_t *)pkt->data, (size_t)pkt->sz, 0);
|
||||
capn_free(&cs);
|
||||
|
||||
return pkt;
|
||||
}
|
||||
|
||||
typedef struct {
|
||||
endpoint_t *ep;
|
||||
io_packet_t *pkt;
|
||||
unsigned delay_ms;
|
||||
} cb5_wire_ack_args_t;
|
||||
|
||||
/* Deliver the wire ACK to the promoted-peer message handler after a delay,
|
||||
* while the main thread is parked inside the barrier wait. */
|
||||
static void *cb5_wire_ack_thread(void *arg)
|
||||
{
|
||||
cb5_wire_ack_args_t *a = (cb5_wire_ack_args_t *)arg;
|
||||
|
||||
usleep(a->delay_ms * 1000);
|
||||
peer_ctrl_msg_handler_for_test(a->ep, a->pkt);
|
||||
return NULL;
|
||||
}
|
||||
|
||||
/* statics survive Unity longjmp */
|
||||
static endpoint_t g_cb5_wire_ep;
|
||||
static cb5_wire_ack_args_t g_cb5_wire_args;
|
||||
|
||||
/* A config ACK delivered as WIRE BYTES through the production decode entry
|
||||
* (peer_ctrl_msg_handler → repl_ack_decode_and_dispatch) must reach
|
||||
* repl_on_config_ack: the replicator slot's acked_config_lsn advances to
|
||||
* the decoded tag and a pending configAck barrier completes (Req 4.1).
|
||||
* RED (remediation r1): repl_ack_decode_and_dispatch has no
|
||||
* replconfigpushreplymsg branch — the reply is silently dropped, the
|
||||
* barrier times out, rc=-ETIMEDOUT. */
|
||||
static void test_config_ack_wire_path_completes_barrier(void)
|
||||
{
|
||||
const uint64_t push_tag = 0xC0FFEE;
|
||||
pthread_t th;
|
||||
int rc;
|
||||
uint64_t acked = 0;
|
||||
int found = 0;
|
||||
|
||||
/* Arrange: cb5 cluster (wired engine n_peers=3, two never-acking
|
||||
* standbys) + a third, promoted-peer endpoint whose vote_ep_ctx routes
|
||||
* peer_ctrl_msg_handler to repl_ack_decode_and_dispatch (role PRIMARY). */
|
||||
cb5_setup();
|
||||
memset(&g_cb5_wire_ep, 0, sizeof(g_cb5_wire_ep));
|
||||
g_cb5_wire_ep.fd = -1;
|
||||
vote_ep_ctx_build_for_test(&g_cb5_wire_ep, &g_cb5_c.ha, NULL, 0,
|
||||
1, 0, 0, "ctrl-cb5");
|
||||
vote_ep_ctx_set_cluster_for_test(&g_cb5_wire_ep, &g_cb5_c);
|
||||
atomic_store_explicit(&g_cb5_c.ha.role, (int32_t)CTRL_PRIMARY,
|
||||
memory_order_release);
|
||||
TEST_ASSERT_TRUE(repl_add_standby(&g_cb5_c.persist.repl,
|
||||
&g_cb5_wire_ep, "cb5_wire", 7502) >= 0);
|
||||
|
||||
/* Act: park in the barrier (self + 1 covering = 2 > 3/2 quorum), then
|
||||
* deliver the standby's ACK as raw capnp bytes through the handler. */
|
||||
g_cb5_wire_args.ep = &g_cb5_wire_ep;
|
||||
g_cb5_wire_args.pkt = cb5_build_config_ack_pkt(1, push_tag);
|
||||
g_cb5_wire_args.delay_ms = 50;
|
||||
TEST_ASSERT_EQUAL_INT(0, pthread_create(&th, NULL, cb5_wire_ack_thread,
|
||||
&g_cb5_wire_args));
|
||||
rc = repl_wait_majority_ack(&g_cb5_c.persist.repl, push_tag, 1000,
|
||||
(uint8_t)OPCode_upgradeConfigAck);
|
||||
pthread_join(th, NULL);
|
||||
|
||||
/* Assert: barrier completed off the wire-decoded ACK … */
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc,
|
||||
"a wire-delivered config ACK must complete a pending barrier"
|
||||
" (production ingestion path — Req 4.1)");
|
||||
|
||||
/* … because repl_on_config_ack advanced the slot to the decoded tag */
|
||||
for (int i = 0; i < MAX_STANDBYS; i++) {
|
||||
if (g_cb5_c.persist.repl.standbys[i].ep == (void *)&g_cb5_wire_ep) {
|
||||
acked = atomic_load_explicit(
|
||||
&g_cb5_c.persist.repl.standbys[i].acked_config_lsn,
|
||||
memory_order_acquire);
|
||||
found = 1;
|
||||
break;
|
||||
}
|
||||
}
|
||||
TEST_ASSERT_TRUE_MESSAGE(found, "wire ep must be registered in the replicator");
|
||||
TEST_ASSERT_EQUAL_UINT64_MESSAGE(push_tag, acked,
|
||||
"the decoded versionTag must land in acked_config_lsn via"
|
||||
" repl_on_config_ack");
|
||||
|
||||
free(g_cb5_wire_ep.data); /* vctx from vote_ep_ctx_build_for_test */
|
||||
g_cb5_wire_ep.data = NULL;
|
||||
cb5_teardown();
|
||||
printf("test_config_ack_wire_path_completes_barrier: PASS\n");
|
||||
}
|
||||
|
||||
/* ── Test group entry point ──────────────────────────────────────────────── */
|
||||
|
||||
void test_commit_barrier_rollout(void)
|
||||
|
|
@ -1531,5 +1700,8 @@ void test_commit_barrier_rollout(void)
|
|||
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);
|
||||
/* Task 5.2 remediation r1 — production wire-path ACK ingestion
|
||||
* (completes the unlanded half of ha-config-bootstrap task 8.1) */
|
||||
RUN_TEST(test_config_ack_wire_path_completes_barrier);
|
||||
UNITY_END();
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue