lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

test(ha-commit-barrier-rollout): RED commit-first-then-reply tests per opcode (task 3.1)

Browse source 
New test_commit_barrier_reply.c drives do_jobcontrol_op /
do_queuecontrol_op / do_hostcontrol_op / handle_worker_join directly
(pthread harness, ep->outq reply capture, wired engine n_peers=3).
12 cases: 8 RED (reply-before-commit + bug-447 pins), 4 validation
pins green. DISCOVERY bug-447 (reviewer-confirmed): key_find found-rc
leak means queue/host control never append WAL in production — 3.2's
rc normalization is the fix point. Census 484 = 472 + 12; zero
baseline regressions.
This commit is contained in:
Rongsong Shen 2026-06-12 10:00:38 +00:00
parent 8f688eace1
commit e1c3c173f4
8 changed files with 1036 additions and 140 deletions
Download patch file Download diff file Expand all files Collapse all files

4
src/controller/meson.build
View file

@ -122,7 +122,11 @@ controller_unit_test_src = [
'testsuite/test_ha_liveness.c',
'testsuite/test_commit_engine.c',
'testsuite/test_commit_barrier_rollout.c',
'testsuite/test_commit_barrier_reply.c',
'commit.c',
'job.c',
'queue.c',
'join.c',
'startup_validation.c',
'worker.c',
'persist.c',

2
src/controller/testsuite/controller_test.c
View file

@ -52,6 +52,7 @@ extern void test_ha_rdhup_integration(void);
extern void test_ha_liveness(void);
extern void test_commit_engine(void);
extern void test_commit_barrier_rollout(void);
extern void test_commit_barrier_reply(void);
void setUp(void) { return; }
@ -173,6 +174,7 @@ int main(int argc, char* argv[])
TEST(test_ha_liveness);
TEST(test_commit_engine);
TEST(test_commit_barrier_rollout);
TEST(test_commit_barrier_reply);
return UNITY_END();
}

800
src/controller/testsuite/test_commit_barrier_reply.c Normal file
View file

@ -0,0 +1,800 @@
/*
* test_commit_barrier_reply.c — RED-phase integration tests for
* commit-first-then-reply per opcode (ha-commit-barrier-rollout task 3.1).
* Requirements 1.1 (success reply only after majority commit), 1.2 (commit
* timeout → -ETIMEDOUT error reply), 1.3 (barrier rejection → error reply,
* never success), 1.4 (no success-interpretable reply pre-commit).
*
* Drives the four request-shaped handlers DIRECTLY on a worker pthread
* (vtask-context stand-in; production dispatch: cluster.c:325-363) against
* a real tmpdir persist_t embedded in a test cluster_t with registered
* persistence hooks, a wired commit engine (n_peers=3), a stub standby
* registry (2 dummy standbys), ACKs driven through repl_on_ack:
* do_jobcontrol_op — not-started-job branch → WAL_JOB_FAIL (SIGKILL)
* do_queuecontrol_op — QUEUE_CLOSE → WAL_QUEUE_CTRL
* do_hostcontrol_op — HOST_CLOSE → WAL_WORKER_CTRL/UPDATE
* handle_worker_join — first-time join → WAL_WORKER_CTRL/JOIN
* (via handle_join — handle_worker_join is static)
*
* Reply observation harness: there is no xce_reply test hook in src/lib
* (out of boundary), so replies are observed where the production send
* path deposits them — xce_reply() encodes the reply Request and
* endpoint_enq()s it onto ep->outq (request.c:334-368, netio.c:796). The
* requester endpoint is a zeroed static endpoint_t with no iomgr:
* endpoint_update_epoll() is a no-op and the zeroed pthread mutexes are
* valid locks on Linux, so packets stay queued. Tests drain via
* endpoint_deq() and parse with the raw capnp reader (capn_init_mem →
* read_Request → read_RequestReply), the raw-packet parse pattern of
* test_config_repl.c — least invasive, zero production changes.
*
* RED expectations against the current reply-before-commit handlers
* (anchors re-verified 2026-06-12: job.c:932, queue.c:130, worker.c:700,
* join.c:107-109):
* - *_withheld_acks_replies_error / *_success_reply_only_after_commit:
* the handlers reply immediately, before any covering ACK → the
* Req 1.1/1.2/1.4 ordering assertions fail. Additional RED finding:
* queue_ctrl/worker_ctrl/job_ctrl leak key_find's found rc (== 1), so
* queue/host control skip their AFTER_EVENT hook (NO WAL entry) and
* reply status 1 — pinned by the append assertion, 3.2 must close it.
* - *_error_reply_immediate: local validation failures already reply
* immediately with their existing codes — contract pins that stay
* green before AND after the task-3.2 restructure.
*
* Fixtures reused from test_commit_barrier_rollout.c (engine + stub
* registry + drop send hook + LSN-0 priming append) and test_worker_join.c
* (kvmgrs, resources, hostattrs, hostconfig; its collect_confvars stub
* also serves this file's join path).
*/
/* rtti.h must come first — it defines 'var' and 'typeinfo_t' used by kvmgr.h */
#include <rtti.h>
#include <errno.h>
#include <pthread.h>
#include <signal.h>
#include <stdatomic.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>
#include <capnp_c.h>
#include <unity/unity.h>
#include "controller.capnp.h"
#include "xce/cluster_def.h"
#include "xce/types.h"
#include "cluster.h"
#include "clusterconfig.h"
#include "commit.h"
#include "hook.h"
#include "jobj.h"
#include "kvmgr.h"
#include "myconf.h"
#include "netio.h"
#include "persist.h"
#include "qobj.h"
#include "repl.h"
#include "resobj.h"
#include "request.h"
#include "typemap.h"
#include "wal_types.h"
#include "wobj.h"
#include "xstring.h"
#include "zarray.h"
/* ── Link stubs ──────────────────────────────────────────────────────────
* job.c / join.c reference these cross-module functions (opupd.c, sched.c,
* apiserver.c — not part of the unit-test link). None of them sits on the
* four code paths under test (jobctrl not-started branch, queue/host ctrl,
* worker join); stubbed so the binary links, same precedent as
* test_worker_join.c's collect_confvars stub. */
/* clang-format off */
int notify(cluster_t *c, endpoint_t *e, xce_notify_data_t *d)
{ (void)c; (void)e; (void)d; return 0; }
worker_t *get_ep_worker(cluster_t *c, endpoint_t *e)
{ (void)c; (void)e; return NULL; }
int update_host_runusage(xce_job_runusage_t **r, char *h, int n,
xce_rusage_t **u)
{ (void)r; (void)h; (void)n; (void)u; return 0; }
int update_pids(job_pidinfo_t **i, int n, int *p)
{ (void)i; (void)n; (void)p; return 0; }
int update_pgids(job_pidinfo_t **i, int n, int *p)
{ (void)i; (void)n; (void)p; return 0; }
int on_scheduler_join(cluster_t *c) { (void)c; return 0; }
int add_apiserver(cluster_t *c, char *n, endpoint_t *e)
{ (void)c; (void)n; (void)e; return 0; }
int on_apiserver_join(cluster_t *c, endpoint_t *e)
{ (void)c; (void)e; return 0; }
/* clang-format on */
/* Test-injectable send hook (repl.c, XCE_TESTING) — keeps repl_broadcast
* from dereferencing the dummy standby endpoints. */
extern int (*repl_broadcast_send_hook)(void *ep,
wal_catchup_entry_t **entries,
int n_entries);
static int cbrp_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 */
}
static uint64_t cbrp_mono_ms(void)
{
struct timespec ts;
clock_gettime(CLOCK_MONOTONIC, &ts);
return (uint64_t)ts.tv_sec * 1000u + (uint64_t)ts.tv_nsec / 1000000u;
}
static void cbrp_sleep_ms(uint32_t ms)
{
struct timespec ts;
ts.tv_sec = ms / 1000u;
ts.tv_nsec = (long)(ms % 1000u) * 1000000L;
nanosleep(&ts, NULL);
}
/* ── Reply observation (see file header for the harness rationale) ───── */
typedef struct {
int n_replies; /* replymsg packets seen by this scan */
int32_t last_status; /* status of the last replymsg observed */
uint64_t committed_at_obs; /* committed_lsn sampled at observation time */
} cbrp_reply_obs_t;
static persist_t *g_obs_persist; /* sampled into committed_at_obs */
/* Drain every packet queued on ep->outq; record replymsg status. Non-reply
* packets (e.g. on_worker_join's config push) are parsed and discarded. */
static void cbrp_scan_replies(endpoint_t *ep, cbrp_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->committed_at_obs = atomic_load(&g_obs_persist->committed_lsn);
}
capn_free(&c);
free_packet(pkt);
free(pkt);
}
}
/* Poll until at least one replymsg is observed or the deadline expires. */
static void cbrp_poll_reply(endpoint_t *ep, uint32_t deadline_ms,
cbrp_reply_obs_t *obs)
{
uint64_t t0 = cbrp_mono_ms();
for (;;) {
cbrp_scan_replies(ep, obs);
if (obs->n_replies > 0 || cbrp_mono_ms() - t0 >= deadline_ms)
return;
cbrp_sleep_ms(5);
}
}
/* Poll until the WAL grows past lsn_before (the operation's append became
* visible) or the deadline expires; returns the appended watermark
* (current_lsn - 1) or 0 on timeout. */
static uint64_t cbrp_poll_append(persist_t *p, uint64_t lsn_before,
uint32_t deadline_ms)
{
uint64_t t0 = cbrp_mono_ms();
for (;;) {
uint64_t cur;
pthread_mutex_lock(&p->lock);
cur = p->current_lsn;
pthread_mutex_unlock(&p->lock);
if (cur > lsn_before)
return cur - 1;
if (cbrp_mono_ms() - t0 >= deadline_ms)
return 0;
cbrp_sleep_ms(5);
}
}
/* ── Handler-on-a-pthread runner (vtask-context stand-in) ───────────── */
typedef int (*cbrp_op_fn_t)(cluster_t *, endpoint_t *, xce_request_t *);
typedef struct {
cbrp_op_fn_t fn;
xce_request_t *req;
int rc;
} cbrp_op_arg_t;
/* Statics survive Unity's longjmp on assertion failure. */
static cbrp_op_arg_t g_op_arg;
static pthread_t g_op_tid;
static cluster_t g_hc; /* harness cluster (defined below) */
static endpoint_t g_req_ep; /* requester endpoint (capture, defined below) */
static void *cbrp_op_thread_fn(void *arg)
{
cbrp_op_arg_t *a = arg;
a->rc = a->fn(&g_hc, &g_req_ep, a->req);
return NULL;
}
/* ── Cluster harness ─────────────────────────────────────────────────── */
/* Resource names/types required by init_worker → apply_join_state
* (add_hostattrs_resvals, worker.c) — mirror of test_worker_join.c. */
#define CBRP_N_RES 9
static const char *cbrp_res_names[CBRP_N_RES] = {
"type", "model", "cpuf", "maxcpus", "maxmem",
"maxswap", "maxtmp", "ndisks", "slots"
};
static const int cbrp_res_valtypes[CBRP_N_RES] = {
XCE_VAL_STRING, XCE_VAL_STRING, XCE_VAL_NUMERIC, XCE_VAL_NUMERIC,
XCE_VAL_NUMERIC, XCE_VAL_NUMERIC, XCE_VAL_NUMERIC, XCE_VAL_NUMERIC,
XCE_VAL_NUMERIC
};
/* Statics survive Unity's longjmp on assertion failure.
* (g_hc / g_req_ep are tentatively defined above for the thread runner.) */
static commit_engine_t g_hce; /* wired commit engine */
static int g_hep_a, g_hep_b;/* dummy standby endpoints */
static ep_private_t g_req_priv;
static xce_peerinfo_t g_req_from;
static char g_hc_tmpdir[256];
/* myconf scaffolding for find_matched_hostconfig (worker join path) */
static myconf_t g_hc_myconf;
static cluster_config_t g_hc_clusterconf;
static xce_hostconfig_t g_hc_hconfig;
static xce_hostconfig_t *g_hc_hconfigs[1];
static void cbrp_setup(uint32_t sync_timeout_ms)
{
persist_config_t cfg;
kvmgr_params_t str_params = { .key_type = KVMGR_KEY_STRING,
.nslots = 64, .delta = 16 };
kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT,
.nslots = 64, .delta = 16 };
int i;
memset(&g_hc, 0, sizeof(g_hc));
memset(&g_req_ep, 0, sizeof(g_req_ep));
memset(&g_req_priv, 0, sizeof(g_req_priv));
memset(&g_req_from, 0, sizeof(g_req_from));
/* Requester endpoint: zeroed, no iomgr — xce_reply packets stay on
* ep->outq (see file header). */
g_req_ep.fd = -1;
g_req_ep.data = &g_req_priv;
g_req_from.name = (char *)"requester";
g_req_from.key = (char *)"";
g_hc.myhost = &g_hc.myhost_info;
g_hc.myhost_info.name = (char *)"ctrl-test";
g_hc.myhost_info.key = (char *)"";
/* ha zero-init: CTRL_PRIMARY == 0 — handle_join's standby gates pass */
TEST_ASSERT_EQUAL_INT(0, xce_hook_init(&g_hc.hooksys,
XCE_CONTROLLER_HOOK_MAX));
g_hc.jobmgr = makeInstance(&xkvmgr_type, &int_params);
g_hc.qmgr = makeInstance(&xkvmgr_type, &str_params);
g_hc.workermgr = makeInstance(&xkvmgr_type, &str_params);
g_hc.resmgr = makeInstance(&xkvmgr_type, &str_params);
TEST_ASSERT_NOT_NULL(g_hc.jobmgr);
TEST_ASSERT_NOT_NULL(g_hc.qmgr);
TEST_ASSERT_NOT_NULL(g_hc.workermgr);
TEST_ASSERT_NOT_NULL(g_hc.resmgr);
zarray_init(&g_hc.sharedinst, 64, NULL);
for (i = 0; i < CBRP_N_RES; i++) {
xce_resource_t rdef;
memset(&rdef, 0, sizeof(rdef));
rdef.name = (char *)cbrp_res_names[i];
rdef.valtype = cbrp_res_valtypes[i];
rdef.idx = -1;
TEST_ASSERT_NOT_NULL_MESSAGE(makeResource(g_hc.resmgr, &rdef),
cbrp_res_names[i]);
}
/* hostconfig "default" so find_matched_hostconfig succeeds for joins */
memset(&g_hc_myconf, 0, sizeof(g_hc_myconf));
memset(&g_hc_clusterconf, 0, sizeof(g_hc_clusterconf));
memset(&g_hc_hconfig, 0, sizeof(g_hc_hconfig));
g_hc_hconfig.name = (char *)"default";
g_hc_hconfig.hosttype = (char *)"!";
g_hc_hconfig.hostmodel = (char *)"!";
g_hc_hconfig.mxj = 2;
g_hc_hconfigs[0] = &g_hc_hconfig;
g_hc_clusterconf.nhconfigs = 1;
g_hc_clusterconf.hconfigs = g_hc_hconfigs;
g_hc_myconf.cluster_config = &g_hc_clusterconf;
g_hc.myconf = &g_hc_myconf;
/* Real persist over a private tmpdir; cluster back-pointer set by
* persist_init so persist_start registers the production hooks. */
cfg = persist_config_defaults();
cfg.wal_sync_ms = 5000;
cfg.sync_timeout_ms = sync_timeout_ms;
snprintf(g_hc_tmpdir, sizeof(g_hc_tmpdir), "/tmp/xce_cbrp31_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(g_hc_tmpdir));
persist_set_workdir(&g_hc.persist, g_hc_tmpdir, "ctrl");
TEST_ASSERT_EQUAL_INT(0, persist_init(&g_hc.persist, &cfg, &g_hc));
TEST_ASSERT_EQUAL_INT(0, persist_start(&g_hc.persist));
/* 3-member cluster: 2 stub standbys, engine wired, ACKs withheld
* until a test explicitly delivers them via repl_on_ack. */
TEST_ASSERT_TRUE(
repl_add_standby(&g_hc.persist.repl, &g_hep_a, "sb_a", 7400) >= 0);
TEST_ASSERT_TRUE(
repl_add_standby(&g_hc.persist.repl, &g_hep_b, "sb_b", 7401) >= 0);
TEST_ASSERT_EQUAL_INT(0, commit_engine_init(&g_hce, &g_hc.persist, 3, 8));
repl_set_commit_engine(&g_hc.persist.repl, &g_hce);
repl_broadcast_send_hook = cbrp_drop_send_hook;
g_obs_persist = &g_hc.persist;
/* Priming append so operation entries land at LSN >= 1 (the engine
* rejects lsn == 0 — accepted interim LSN-0 edge, ha-commit-engine). */
TEST_ASSERT_EQUAL_INT(
0, persist_wal_async(&g_hc.persist, WAL_JOB_UPDATE, "p", 1));
TEST_ASSERT_EQUAL_UINT64(1, g_hc.persist.current_lsn);
}
static void cbrp_teardown(void)
{
cbrp_reply_obs_t drain = { 0 };
cbrp_scan_replies(&g_req_ep, &drain); /* free any leftover packets */
repl_broadcast_send_hook = NULL;
commit_engine_fini(&g_hce);
persist_stop(&g_hc.persist);
persist_fini(&g_hc.persist);
zarray_fini(&g_hc.sharedinst);
if (g_hc.resmgr) dereference(g_hc.resmgr);
if (g_hc.workermgr) dereference(g_hc.workermgr);
if (g_hc.qmgr) dereference(g_hc.qmgr);
if (g_hc.jobmgr) dereference(g_hc.jobmgr);
xce_hook_fini(&g_hc.hooksys);
}
/* ── Per-opcode entity + request builders ────────────────────────────── */
/* job 42, submitted but never started (allocs == NULL): SIGKILL takes the
* not-started branch of job_ctrl → mark_job_fail → WAL_JOB_FAIL. */
static xce_signal_t g_sigkill = { .signo = SIGKILL, .force = 0 };
static xce_jobctrl_t g_jobctrl;
static xce_request_t g_jobctrl_req;
static job_submit_data_t g_job_submit; /* job_init requires submit != NULL */
static void cbrp_arrange_jobctrl(uint64_t jobid, int create_job)
{
if (create_job) {
xce_job_t jdef;
static int jobobj_ready; /* jobj module tables — once per process */
if (!jobobj_ready) {
TEST_ASSERT_EQUAL_INT(0, job_obj_init(64, 64));
jobobj_ready = 1;
}
memset(&g_job_submit, 0, sizeof(g_job_submit));
memset(&jdef, 0, sizeof(jdef));
jdef.jobid = jobid;
jdef.state = JOB_STAT_PEND;
jdef.submit = &g_job_submit;
TEST_ASSERT_NOT_NULL(makeJob(g_hc.jobmgr, &jdef));
}
memset(&g_jobctrl, 0, sizeof(g_jobctrl));
g_jobctrl.action = JobControl_u_siginfo;
g_jobctrl.jobid = jobid;
g_jobctrl.u.siginfo = &g_sigkill;
memset(&g_jobctrl_req, 0, sizeof(g_jobctrl_req));
g_jobctrl_req.opcode = OPCode_jobcontrol;
g_jobctrl_req.seq = 11;
g_jobctrl_req.from = &g_req_from;
g_jobctrl_req.body_which = Request_body_jobctrlmsg;
g_jobctrl_req.body.jobctrlmsg = &g_jobctrl;
}
/* queue "q1": QUEUE_CLOSE → status |= QUEUE_STAT_CLOSED → WAL_QUEUE_CTRL */
static xce_queuectrl_t g_queuectrl;
static xce_request_t g_queuectrl_req;
static void cbrp_arrange_queuectrl(const char *qname, int create_queue)
{
if (create_queue) {
xce_queue_t qdef;
memset(&qdef, 0, sizeof(qdef));
qdef.queue = (char *)qname;
TEST_ASSERT_NOT_NULL(makeQueue(g_hc.qmgr, &qdef));
}
memset(&g_queuectrl, 0, sizeof(g_queuectrl));
g_queuectrl.action = QUEUE_CLOSE;
g_queuectrl.queue = (char *)qname;
memset(&g_queuectrl_req, 0, sizeof(g_queuectrl_req));
g_queuectrl_req.opcode = OPCode_queuecontrol;
g_queuectrl_req.seq = 12;
g_queuectrl_req.from = &g_req_from;
g_queuectrl_req.body_which = Request_body_queuectrlmsg;
g_queuectrl_req.body.queuectrlmsg = &g_queuectrl;
}
/* worker "hostA": HOST_CLOSE → status |= XCE_WORKER_CLOSED →
* WAL_WORKER_CTRL detail WORKER_UPDATE */
static xce_hostctrl_t g_hostctrl;
static xce_request_t g_hostctrl_req;
static void cbrp_arrange_hostctrl(const char *host, int create_worker)
{
if (create_worker) {
xce_worker_t wdef;
hostattrs_t *attrs = calloc(1, sizeof(*attrs));
TEST_ASSERT_NOT_NULL(attrs);
attrs->hosttype = makeString("LINUX");
attrs->hostmodel = makeString("x86");
attrs->cpufactor = 1.0f;
attrs->maxcpus = 4;
memset(&wdef, 0, sizeof(wdef));
wdef.name = (char *)host;
wdef.attrs = attrs;
TEST_ASSERT_NOT_NULL(makeWorker(g_hc.workermgr, &wdef));
}
memset(&g_hostctrl, 0, sizeof(g_hostctrl));
g_hostctrl.action = HOST_CLOSE;
g_hostctrl.hostname = (char *)host;
memset(&g_hostctrl_req, 0, sizeof(g_hostctrl_req));
g_hostctrl_req.opcode = OPCode_hostcontrol;
g_hostctrl_req.seq = 13;
g_hostctrl_req.from = &g_req_from;
g_hostctrl_req.body_which = Request_body_hostctrlmsg;
g_hostctrl_req.body.hostctrlmsg = &g_hostctrl;
}
/* worker join: handle_join with role HostRole_worker reaches the static
* handle_worker_join; first-time join → init_worker → AFTER_EVENT hook →
* WAL_WORKER_CTRL detail WORKER_JOIN. */
static xce_joindata_t g_joindata;
static hostattrs_t g_join_attrs;
static xce_peerinfo_t g_join_from;
static xce_request_t g_join_req;
static void cbrp_arrange_worker_join(const char *host)
{
memset(&g_join_attrs, 0, sizeof(g_join_attrs));
g_join_attrs.hosttype = makeString("LINUX");
g_join_attrs.hostmodel = makeString("x86");
g_join_attrs.cpufactor = 1.0f;
g_join_attrs.maxcpus = 4;
g_join_attrs.maxmem = 16;
g_join_attrs.maxswap = 32;
g_join_attrs.maxtmp = 64;
g_join_attrs.ndisks = 1;
memset(&g_joindata, 0, sizeof(g_joindata));
g_joindata.role = HostRole_worker;
g_joindata.data_which = JoinData_data_attrs;
g_joindata.data.attrs = &g_join_attrs;
memset(&g_join_from, 0, sizeof(g_join_from));
g_join_from.name = (char *)host;
g_join_from.key = (char *)"";
memset(&g_join_req, 0, sizeof(g_join_req));
g_join_req.opcode = OPCode_join;
g_join_req.seq = 14;
g_join_req.from = &g_join_from;
g_join_req.body_which = Request_body_joinmsg;
g_join_req.body.joinmsg = &g_joindata;
}
/* ── Shared case drivers ─────────────────────────────────────────────── */
/* Withheld-ACK case (Req 1.2, 1.4): the requester reply must carry
* -ETIMEDOUT and no success reply may be observable while the operation's
* entry exceeds committed_lsn.
* RED: the current handlers reply SUCCESS immediately. */
static void cbrp_drive_withheld(cbrp_op_fn_t fn, xce_request_t *req,
uint32_t sync_timeout_ms)
{
cbrp_reply_obs_t obs = { 0 };
uint64_t lsn_before, watermark, committed_final;
pthread_mutex_lock(&g_hc.persist.lock);
lsn_before = g_hc.persist.current_lsn;
pthread_mutex_unlock(&g_hc.persist.lock);
/* Act: handler on a worker pthread (post-3.2 it blocks at the barrier) */
g_op_arg.fn = fn;
g_op_arg.req = req;
g_op_arg.rc = -9999;
TEST_ASSERT_EQUAL_INT(0, pthread_create(&g_op_tid, NULL,
cbrp_op_thread_fn, &g_op_arg));
/* The reply may legitimately take up to the commit timeout */
cbrp_poll_reply(&g_req_ep, sync_timeout_ms + 1500, &obs);
pthread_join(g_op_tid, NULL);
watermark = cbrp_poll_append(&g_hc.persist, lsn_before, 1000);
committed_final = atomic_load(&g_hc.persist.committed_lsn);
/* Assert: the operation must produce its WAL entry — nothing to
* commit-confirm otherwise (Req 1.1). RED finding: queue_ctrl and
* worker_ctrl leak key_find's found rc (== 1) into the `if (rc == 0)`
* hook gate → NO WAL entry, status-1 reply (header; 3.2 closes it). */
TEST_ASSERT_TRUE_MESSAGE(watermark > 0,
"request-shaped control op must append its WAL entry (queue_ctrl/"
"worker_ctrl currently skip the AFTER_EVENT hook: rc==1 leak)");
/* ...which never committed (ACKs withheld) */
TEST_ASSERT_TRUE_MESSAGE(committed_final < watermark,
"committed_lsn must stay below the entry with ACKs withheld");
/* ...the requester got exactly an error reply */
TEST_ASSERT_TRUE_MESSAGE(obs.n_replies > 0,
"the requester must receive a reply");
TEST_ASSERT_TRUE_MESSAGE(
!(obs.last_status == 0 && obs.committed_at_obs < watermark),
"no success reply may be observable while the operation's entry "
"exceeds committed_lsn (Req 1.4)");
TEST_ASSERT_EQUAL_INT_MESSAGE(-ETIMEDOUT, obs.last_status,
"withheld ACKs past the commit timeout must surface -ETIMEDOUT "
"to the requester (Req 1.2)");
}
/* Covering-ACK case (Req 1.1, 1.4): the success reply must be observable
* only after committed_lsn covers the operation's entry.
* RED: the current handlers reply success before any ACK arrives. */
static void cbrp_drive_acked(cbrp_op_fn_t fn, xce_request_t *req)
{
cbrp_reply_obs_t early = { 0 };
cbrp_reply_obs_t final = { 0 };
uint64_t lsn_before, watermark;
pthread_mutex_lock(&g_hc.persist.lock);
lsn_before = g_hc.persist.current_lsn;
pthread_mutex_unlock(&g_hc.persist.lock);
g_op_arg.fn = fn;
g_op_arg.req = req;
g_op_arg.rc = -9999;
TEST_ASSERT_EQUAL_INT(0, pthread_create(&g_op_tid, NULL,
cbrp_op_thread_fn, &g_op_arg));
/* Wait until the operation's append is visible, then hold the handler
* blocked for a beat and sample the reply capture BEFORE any ACK. */
watermark = cbrp_poll_append(&g_hc.persist, lsn_before, 2000);
cbrp_sleep_ms(150);
cbrp_scan_replies(&g_req_ep, &early);
/* Deliver the covering ACK, then collect the (now permitted) reply */
repl_on_ack(&g_hc.persist.repl, &g_hep_a, watermark);
cbrp_poll_reply(&g_req_ep, 2000, &final);
pthread_join(g_op_tid, NULL);
/* Assert: the operation must produce its WAL entry (see the rc==1
* leak note in cbrp_drive_withheld — same RED finding applies) */
TEST_ASSERT_TRUE_MESSAGE(watermark > 0,
"request-shaped control op must append its WAL entry (queue_ctrl/"
"worker_ctrl currently skip the AFTER_EVENT hook: rc==1 leak)");
/* THE ordering gate: nothing the requester can read as confirmation
* may exist before the covering ACK (Req 1.1, 1.4) */
TEST_ASSERT_EQUAL_INT_MESSAGE(0, early.n_replies,
"no reply may be observable before committed_lsn covers the "
"operation's entry (Req 1.1, 1.4)");
/* After the ACK: success reply, entry committed */
TEST_ASSERT_TRUE_MESSAGE(final.n_replies > 0,
"the requester must receive the reply once the entry commits");
TEST_ASSERT_EQUAL_INT_MESSAGE(0, final.last_status,
"the post-commit reply must report success (Req 1.1)");
TEST_ASSERT_TRUE_MESSAGE(final.committed_at_obs >= watermark,
"the success reply must be observable only with committed_lsn "
"covering the operation's entry (Req 1.1)");
}
/* Local-validation-failure pin (Req 1.4 boundary, task brief): the
* existing error reply stays immediate — no barrier wait, no WAL entry.
* Green before AND after the task-3.2 restructure (latency-bounded). */
static void cbrp_drive_validation_error(cbrp_op_fn_t fn, xce_request_t *req,
int32_t expect_status,
int check_exact_status)
{
cbrp_reply_obs_t obs = { 0 };
uint64_t lsn_before, lsn_after, t0, elapsed_ms;
pthread_mutex_lock(&g_hc.persist.lock);
lsn_before = g_hc.persist.current_lsn;
pthread_mutex_unlock(&g_hc.persist.lock);
/* Act: synchronous — a validation failure must never block */
t0 = cbrp_mono_ms();
(void)fn(&g_hc, &g_req_ep, req);
elapsed_ms = cbrp_mono_ms() - t0;
cbrp_scan_replies(&g_req_ep, &obs);
pthread_mutex_lock(&g_hc.persist.lock);
lsn_after = g_hc.persist.current_lsn;
pthread_mutex_unlock(&g_hc.persist.lock);
/* Assert: immediate error reply with the existing code, no append */
TEST_ASSERT_TRUE_MESSAGE(elapsed_ms < 1000,
"local validation failures must reply without a barrier wait");
TEST_ASSERT_TRUE_MESSAGE(obs.n_replies > 0,
"validation failure must still produce a requester reply");
if (check_exact_status)
TEST_ASSERT_EQUAL_INT_MESSAGE(expect_status, obs.last_status,
"validation failure must keep its existing error code");
else
TEST_ASSERT_TRUE_MESSAGE(obs.last_status < 0,
"validation failure must reply with an error status");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_before, lsn_after,
"a rejected operation must append no WAL entry");
}
/* ── do_jobcontrol_op (not-started branch → WAL_JOB_FAIL) ────────────── */
static void test_jobctrl_withheld_acks_replies_error(void)
{
cbrp_setup(300);
cbrp_arrange_jobctrl(42, 1);
cbrp_drive_withheld(do_jobcontrol_op, &g_jobctrl_req, 300);
cbrp_teardown();
printf("test_jobctrl_withheld_acks_replies_error: PASS\n");
}
static void test_jobctrl_success_reply_only_after_commit(void)
{
cbrp_setup(1500);
cbrp_arrange_jobctrl(43, 1);
cbrp_drive_acked(do_jobcontrol_op, &g_jobctrl_req);
cbrp_teardown();
printf("test_jobctrl_success_reply_only_after_commit: PASS\n");
}
static void test_jobctrl_unknown_job_error_reply_immediate(void)
{
cbrp_setup(3000);
cbrp_arrange_jobctrl(9999, 0); /* job does not exist */
cbrp_drive_validation_error(do_jobcontrol_op, &g_jobctrl_req,
-ENOENT, 1);
cbrp_teardown();
printf("test_jobctrl_unknown_job_error_reply_immediate: PASS\n");
}
/* ── do_queuecontrol_op (QUEUE_CLOSE → WAL_QUEUE_CTRL) ───────────────── */
static void test_queuectrl_withheld_acks_replies_error(void)
{
cbrp_setup(300);
cbrp_arrange_queuectrl("q1", 1);
cbrp_drive_withheld(do_queuecontrol_op, &g_queuectrl_req, 300);
cbrp_teardown();
printf("test_queuectrl_withheld_acks_replies_error: PASS\n");
}
static void test_queuectrl_success_reply_only_after_commit(void)
{
cbrp_setup(1500);
cbrp_arrange_queuectrl("q2", 1);
cbrp_drive_acked(do_queuecontrol_op, &g_queuectrl_req);
cbrp_teardown();
printf("test_queuectrl_success_reply_only_after_commit: PASS\n");
}
static void test_queuectrl_unknown_queue_error_reply_immediate(void)
{
cbrp_setup(3000);
cbrp_arrange_queuectrl("no-such-queue", 0);
cbrp_drive_validation_error(do_queuecontrol_op, &g_queuectrl_req,
-ENOENT, 1);
cbrp_teardown();
printf("test_queuectrl_unknown_queue_error_reply_immediate: PASS\n");
}
/* ── do_hostcontrol_op (HOST_CLOSE → WAL_WORKER_CTRL/UPDATE) ─────────── */
static void test_hostctrl_withheld_acks_replies_error(void)
{
cbrp_setup(300);
cbrp_arrange_hostctrl("hostA", 1);
cbrp_drive_withheld(do_hostcontrol_op, &g_hostctrl_req, 300);
cbrp_teardown();
printf("test_hostctrl_withheld_acks_replies_error: PASS\n");
}
static void test_hostctrl_success_reply_only_after_commit(void)
{
cbrp_setup(1500);
cbrp_arrange_hostctrl("hostB", 1);
cbrp_drive_acked(do_hostcontrol_op, &g_hostctrl_req);
cbrp_teardown();
printf("test_hostctrl_success_reply_only_after_commit: PASS\n");
}
static void test_hostctrl_unknown_host_error_reply_immediate(void)
{
cbrp_setup(3000);
cbrp_arrange_hostctrl("no-such-host", 0);
cbrp_drive_validation_error(do_hostcontrol_op, &g_hostctrl_req,
-ENOENT, 1);
cbrp_teardown();
printf("test_hostctrl_unknown_host_error_reply_immediate: PASS\n");
}
/* ── handle_worker_join via handle_join (→ WAL_WORKER_CTRL/JOIN) ─────── */
static void test_worker_join_withheld_acks_replies_error(void)
{
cbrp_setup(300);
cbrp_arrange_worker_join("workerA");
cbrp_drive_withheld(handle_join, &g_join_req, 300);
cbrp_teardown();
printf("test_worker_join_withheld_acks_replies_error: PASS\n");
}
static void test_worker_join_success_reply_only_after_commit(void)
{
cbrp_setup(1500);
cbrp_arrange_worker_join("workerB");
cbrp_drive_acked(handle_join, &g_join_req);
cbrp_teardown();
printf("test_worker_join_success_reply_only_after_commit: PASS\n");
}
static void test_worker_join_no_hostconfig_error_reply_immediate(void)
{
cbrp_setup(3000);
cbrp_arrange_worker_join("workerC");
/* Remove every hostconfig: find_matched_hostconfig returns NULL and
* handle_worker_join must reply its existing error immediately —
* the (future) barrier sits between the AFTER_EVENT hook and the
* SUCCESS reply only; error replies stay immediate. */
g_hc_clusterconf.nhconfigs = 0;
g_hc_clusterconf.hconfigs = NULL;
cbrp_drive_validation_error(handle_join, &g_join_req, 0, 0);
cbrp_teardown();
printf("test_worker_join_no_hostconfig_error_reply_immediate: PASS\n");
}
/* ── Test group entry point ──────────────────────────────────────────── */
void test_commit_barrier_reply(void)
{
/* makeString needs the global xstring table. In the full suite,
* test_worker_join already initialised it (unguarded, its line 520);
* standalone (-t test_commit_barrier_reply) it is uninitialised and
* makeString returns NULL. Probe instead of double-initialising. */
if (makeString((char *)"cbrp-probe") == NULL)
xstring_table_init(139);
UNITY_BEGIN();
/* do_jobcontrol_op — Req 1.1, 1.2, 1.4 */
RUN_TEST(test_jobctrl_withheld_acks_replies_error);
RUN_TEST(test_jobctrl_success_reply_only_after_commit);
RUN_TEST(test_jobctrl_unknown_job_error_reply_immediate);
/* do_queuecontrol_op */
RUN_TEST(test_queuectrl_withheld_acks_replies_error);
RUN_TEST(test_queuectrl_success_reply_only_after_commit);
RUN_TEST(test_queuectrl_unknown_queue_error_reply_immediate);
/* do_hostcontrol_op */
RUN_TEST(test_hostctrl_withheld_acks_replies_error);
RUN_TEST(test_hostctrl_success_reply_only_after_commit);
RUN_TEST(test_hostctrl_unknown_host_error_reply_immediate);
/* handle_worker_join (via handle_join) */
RUN_TEST(test_worker_join_withheld_acks_replies_error);
RUN_TEST(test_worker_join_success_reply_only_after_commit);
RUN_TEST(test_worker_join_no_hostconfig_error_reply_immediate);
UNITY_END();
}