diff --git a/src/controller/cluster.c b/src/controller/cluster.c index 902aedc5..9ef321e1 100644 --- a/src/controller/cluster.c +++ b/src/controller/cluster.c @@ -124,6 +124,46 @@ static void handle_endpoint_error(endpoint_t* ep) return; } +static int do_notify(cluster_t* c, endpoint_t* ep, xce_request_t* req); + +/* bug-1741: OPCode_notifyAcked -- the same job-lifecycle facts as + * OPCode_notify, but answered: the reply is sent only after the fact's WAL + * entry is majority-committed, so a worker that got status 0 knows the fact + * survives a failover, and a worker that got an error (or no reply) resends + * it. A resend is harmless: START/FINISH are guarded against duplicates and a + * repeated FAIL yields the same state. A barrier failure (e.g. transient + * quorum loss -- the fact itself is retained) is answered -E_XCE_NOT_READY + * so the worker retries in place rather than rotating away from a healthy + * primary (cf. bug-1850). */ +static int do_notify_acked(cluster_t* c, endpoint_t* ep, xce_request_t* req) +{ + xce_reply_t reply; + int rc; + + init_reply_by_req(&reply, req); + + do_notify(c, ep, req); + rc = persist_commit_barrier(&c->persist, 0); + if (rc != 0) { + WARN("%s: commit barrier failed (rc %d) -- worker will resend", + __func__, rc); + rc = -E_XCE_NOT_READY; + } + + reply.status = rc; + reply.data_which = RequestReply_data_nodata; + xce_reply(ep, c->myhost, &reply); + + return 0; +} + +#ifdef XCE_TESTING +int do_notify_acked_for_test(cluster_t* c, endpoint_t* ep, xce_request_t* req) +{ + return do_notify_acked(c, ep, req); +} +#endif + static int do_notify(cluster_t* c, endpoint_t* ep, xce_request_t* req) { xce_notify_data_t* data = req->body.notifymsg; @@ -612,6 +652,9 @@ static int myreq_handler(vtask_t* task) case OPCode_notify: rc = do_notify(c, ep, &request); break; + case OPCode_notifyAcked: + rc = do_notify_acked(c, ep, &request); + break; case OPCode_clusterconf: rc = handle_clusterconf(c, ep, &request); break; diff --git a/src/controller/ha.c b/src/controller/ha.c index ccf3bbca..f32fac4d 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -4745,6 +4745,7 @@ int ha_dispatch_gate(ha_state_t *ha, int opcode) * (bug-1741) — found live while auditing this gate for task * 6.2's own standby-redirect investigation, not assumed. */ case OPCode_notify: + case OPCode_notifyAcked: /* bug-1741: same facts, answered */ case OPCode_update: case OPCode_setpreason: case OPCode_clrpreason: diff --git a/src/controller/testsuite/test_commit_barrier_reply.c b/src/controller/testsuite/test_commit_barrier_reply.c index 55b5664a..144b6d56 100644 --- a/src/controller/testsuite/test_commit_barrier_reply.c +++ b/src/controller/testsuite/test_commit_barrier_reply.c @@ -779,6 +779,81 @@ static void cbrp_drive_validation_error(cbrp_op_fn_t fn, xce_request_t *req, "a rejected operation must append no WAL entry"); } +/* ── bug-1741: OPCode_notifyAcked (job-lifecycle facts, answered) ─────── */ + +extern int do_notify_acked_for_test(cluster_t *, endpoint_t *, xce_request_t *); + +static job_failinfo_t g_nack_fail; +static xce_notify_data_t g_nack_data; +static xce_request_t g_nack_req; + +static void cbrp_arrange_notify_acked_fail(uint64_t jobid) +{ + memset(&g_job_submit, 0, sizeof(g_job_submit)); + cbrp_make_job(jobid, &g_job_submit); + memset(&g_nack_fail, 0, sizeof(g_nack_fail)); + g_nack_fail.jobid = jobid; + g_nack_fail.stage = JOB_EXEC_STAGE_MAIN; + g_nack_fail.code = 3; + g_nack_fail.detail = (char *)"exit"; + memset(&g_nack_data, 0, sizeof(g_nack_data)); + g_nack_data.event = NotifyData_data_jobFailInfo; + g_nack_data.data.failinfo = &g_nack_fail; + memset(&g_nack_req, 0, sizeof(g_nack_req)); + g_nack_req.opcode = OPCode_notifyAcked; + g_nack_req.seq = 19; + g_nack_req.from = &g_req_from; + g_nack_req.body_which = Request_body_notifymsg; + g_nack_req.body.notifymsg = &g_nack_data; +} + +/* The ACK is sent only once the fact is committed -- a worker that got 0 + * knows the fact survives a failover. */ +static void test_notify_acked_reply_only_after_commit(void) +{ + cbrp_setup(1500); + cbrp_arrange_notify_acked_fail(81); + cbrp_drive_acked(do_notify_acked_for_test, &g_nack_req); + cbrp_teardown(); +} + +/* A barrier failure keeps the fact (appended, retained) and answers + * -E_XCE_NOT_READY, so the worker resends in place instead of rotating + * away from a healthy primary. */ +static void test_notify_acked_withheld_replies_not_ready(void) +{ + cbrp_reply_obs_t obs = { 0 }; + uint64_t lsn_before, lsn_after; + + cbrp_setup(300); + cbrp_arrange_notify_acked_fail(82); + + 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 = do_notify_acked_for_test; + g_op_arg.req = &g_nack_req; + g_op_arg.rc = -9999; + TEST_ASSERT_EQUAL_INT(0, pthread_create(&g_op_tid, NULL, + cbrp_op_thread_fn, &g_op_arg)); + cbrp_poll_reply(&g_req_ep, 300 + 1500, &obs); + pthread_join(g_op_tid, NULL); + + pthread_mutex_lock(&g_hc.persist.lock); + lsn_after = g_hc.persist.current_lsn; + pthread_mutex_unlock(&g_hc.persist.lock); + + TEST_ASSERT_TRUE_MESSAGE(lsn_after > lsn_before, + "the fact must still be appended (retained) without quorum"); + TEST_ASSERT_TRUE(obs.n_replies > 0); + TEST_ASSERT_EQUAL_INT_MESSAGE(-E_XCE_NOT_READY, obs.last_status, + "bug-1741: an uncommitted fact must be answered NOT_READY so the " + "worker resends it (without rotating controllers)"); + cbrp_teardown(); +} + + /* ── do_jobcontrol_op (not-started branch → WAL_JOB_FAIL) ────────────── */ static void test_jobctrl_withheld_acks_replies_error(void) @@ -3654,6 +3729,8 @@ void test_commit_barrier_reply(void) UnitySetTestFile(__FILE__); /* do_jobcontrol_op — Req 1.1, 1.2, 1.4 */ RUN_TEST(test_jobctrl_withheld_acks_replies_error); + RUN_TEST(test_notify_acked_reply_only_after_commit); + RUN_TEST(test_notify_acked_withheld_replies_not_ready); RUN_TEST(test_jobctrl_success_reply_only_after_commit); RUN_TEST(test_jobctrl_unknown_job_error_reply_immediate); /* do_queuecontrol_op */ diff --git a/src/include/client.h b/src/include/client.h index 39330fee..82b3b3f5 100644 --- a/src/include/client.h +++ b/src/include/client.h @@ -108,6 +108,10 @@ extern int xcecli_resval(xcecli_ctx_t, char*, char*, xce_resval_t***); extern int xcecli_update_jusage(xcecli_ctx_t, uint64_t, int, int*, int, int*, int, xce_rusage_t**); extern int xcecli_notify(xcecli_ctx_t, xce_notify_data_t*); +/* bug-1741: acknowledged notify -- retried (failover) until the controller + * confirms the fact committed; falls back to fire-and-forget against a + * controller that predates OPCode_notifyAcked. */ +extern int xcecli_notify_acked(xcecli_ctx_t, xce_notify_data_t*); extern int xcecli_jobctrl(xcecli_ctx_t, xce_jobctrl_t*); extern int xcecli_notify_jobchkpt(xcecli_ctx_t, uint64_t, char*); extern int xcecli_queuectrl(xcecli_ctx_t, xce_queuectrl_t*); diff --git a/src/include/request.h b/src/include/request.h index 01aacb52..ee6def8b 100644 --- a/src/include/request.h +++ b/src/include/request.h @@ -89,6 +89,8 @@ extern int xce_job_state_report_req(endpoint_t*, xce_peerinfo_t*, xce_check_fn_t, void*, int); extern int xce_notify_req(endpoint_t*, xce_notify_data_t*); +/* bug-1741: OPCode_notifyAcked -- answered once the fact commits. */ +extern int xce_notify_acked_req(endpoint_t*, xce_notify_data_t*); extern int extract_data(dataset_t*, xce_data_t*); extern int xce_update_req(endpoint_t*, xce_update_data_t*); extern int xce_jobctrl_req(endpoint_t*, xce_jobctrl_t*); diff --git a/src/include/xceapi.h b/src/include/xceapi.h index e4f93aed..61761691 100644 --- a/src/include/xceapi.h +++ b/src/include/xceapi.h @@ -24,6 +24,8 @@ extern int xce_update_rows(xce_ctx_t*, int, int, xlist_t*); extern int xce_resval(xce_ctx_t*, char*, char*, xce_resval_t***); extern int xce_notify(xce_ctx_t*, xce_notify_data_t*); +/* bug-1741: send a notify and wait for the controller's commit ACK. */ +extern int xce_notify_acked(xce_ctx_t*, xce_notify_data_t*); extern int xce_update_resvals(xce_ctx_t*, int, xce_resval_t**); extern int xce_jobctrl(xce_ctx_t*, xce_jobctrl_t*); extern int xce_queuectrl(xce_ctx_t*, xce_queuectrl_t*); diff --git a/src/lib/client.c b/src/lib/client.c index b7ba33cf..9b465f17 100644 --- a/src/lib/client.c +++ b/src/lib/client.c @@ -562,7 +562,7 @@ int xcecli_notify_jobfinish(xcecli_ctx_t c, xce_jobfinish_t* jfinish) return -EINVAL; } - return xcecli_notify(ctx, &data); + return xcecli_notify_acked(ctx, &data); /* bug-1741 */ } int xcecli_update_resvals(xcecli_ctx_t c, int nres, xce_resval_t** vals) @@ -636,6 +636,30 @@ int xcecli_notify(xcecli_ctx_t c, xce_notify_data_t* data) return XCECLI_FF_CALL(ctx, xce_notify(&(ctx->ctx), data)); } +/* bug-1741: see client.h. A lost job start/finish/fail report used to be + * gone for good (fire-and-forget, no reply), leaving the job RUN forever + * unless the worker happened to rejoin. This resends until the controller + * confirms the fact committed (duplicates are harmless controller-side). An + * old controller answers the unknown opcode with -E_XCE_BAD_OPCODE: send + * the plain fire-and-forget notify instead, so worker and controller can be + * upgraded in either order. */ +int xcecli_notify_acked(xcecli_ctx_t c, xce_notify_data_t* data) +{ + api_ctx_t* ctx = (api_ctx_t*)c; + int rc; + + if ((c == NULL) || (data == NULL)) { + return -EINVAL; + } + + rc = XCECLI_FAILOVER_CALL(ctx, /*is_write=*/true, + xce_notify_acked(&(ctx->ctx), data)); + if (rc == -E_XCE_BAD_OPCODE) { + rc = xcecli_notify(ctx, data); + } + return rc; +} + int xcecli_notify_jobstart(xcecli_ctx_t c, job_pidinfo_t* pidinfo) { api_ctx_t* ctx = (api_ctx_t*)c; @@ -652,7 +676,7 @@ int xcecli_notify_jobstart(xcecli_ctx_t c, job_pidinfo_t* pidinfo) return -EINVAL; } - return xcecli_notify(ctx, &data); + return xcecli_notify_acked(ctx, &data); /* bug-1741 */ } int xcecli_notify_jobfail(xcecli_ctx_t c, job_failinfo_t* failinfo) @@ -671,7 +695,7 @@ int xcecli_notify_jobfail(xcecli_ctx_t c, job_failinfo_t* failinfo) return -EINVAL; } - return xcecli_notify(ctx, &data); + return xcecli_notify_acked(ctx, &data); /* bug-1741 */ } int xcecli_jobctrl(xcecli_ctx_t c, xce_jobctrl_t* ctrlinfo) diff --git a/src/lib/request.c b/src/lib/request.c index 27a252a6..92902955 100644 --- a/src/lib/request.c +++ b/src/lib/request.c @@ -2099,6 +2099,25 @@ int xce_notify_req(endpoint_t* ep, xce_notify_data_t* data) return xce_request(ep, &request); } +int xce_notify_acked_req(endpoint_t* ep, xce_notify_data_t* data) +{ + /* clang-format off */ + xce_request_t request = { + .opcode = OPCode_notifyAcked, + .body_which = Request_body_notifymsg, + .body = { + .notifymsg = data + } + }; + /* clang-format on */ + + if ((ep == NULL) || (data == NULL)) { + return -EINVAL; + } + + return xce_request(ep, &request); +} + int xce_jobsubmit_req(endpoint_t* ep, xce_job_submit_t* submit) { /* clang-format off */ diff --git a/src/lib/xceapi.c b/src/lib/xceapi.c index 02933c54..dd71ee79 100644 --- a/src/lib/xceapi.c +++ b/src/lib/xceapi.c @@ -307,6 +307,32 @@ int xce_notify(xce_ctx_t* ctx, xce_notify_data_t* data) return rc; } +int xce_notify_acked(xce_ctx_t* ctx, xce_notify_data_t* data) +{ + int rc, status; + + if (ctx == NULL) { + return -EINVAL; + } + + if (ctx->state != XCE_CTX_RUN) { + return -EINVAL; + } + + rc = xce_notify_acked_req(ctx->ep, data); + if (rc < 0) { + return rc; + } + + struct timespec _dl; cbuf_deadline_10s(&_dl); + rc = cbuf_timedget(&(ctx->wbuf), &status, &_dl); + if (rc < 0) { + return rc; + } + + return status; +} + int xce_jobsubmit(xce_ctx_t* ctx, xce_job_submit_t* submit, uint64_t* jobid) { int rc, status; diff --git a/src/schema/controller.capnp b/src/schema/controller.capnp index 3f678f14..6b3f3264 100644 --- a/src/schema/controller.capnp +++ b/src/schema/controller.capnp @@ -53,6 +53,7 @@ enum OPCode { schedAdmin @42; jobStateQuery @43; jobStateReport @44; + notifyAcked @45; # bug-1741: OPCode_notify, answered after the fact commits } struct PeerInfo $C.mapname("xce_peerinfo_t") { diff --git a/src/worker/src/controller_connector.c b/src/worker/src/controller_connector.c index 82ffec47..48fe6131 100755 --- a/src/worker/src/controller_connector.c +++ b/src/worker/src/controller_connector.c @@ -1834,7 +1834,16 @@ bool xcecd_reply_startjob(xcecd_job_t* job) pidinfo.exdata = strdup(job->exdata); } - xcecli_notify_jobstart(xcecli_ctx, &pidinfo); + /* bug-1741: acknowledged -- resent until the controller confirms. */ + if (xcecli_notify_jobstart(xcecli_ctx, &pidinfo) < 0) { + char jobidStr[64]; + + fmt_jobid(jobidStr, sizeof(jobidStr), job->job_id); + XCECD_LOG(XCE_LOG_WARN, + "job %s start report not acknowledged by the controller " + "after retries -- the rejoin resync will reconcile it", + jobidStr); + } FREEUP(pidinfo.exdata); @@ -1892,7 +1901,18 @@ int xcecd_notify_job_fail(uint64_t jobid, }; XCECD_LOG(XCE_LOG_DEBUG, "Job failed and calling xcecli_notify_jobfail..."); - return xcecli_notify_jobfail(xcecli_ctx, &failinfo); + /* bug-1741: acknowledged -- resent until the controller confirms. */ + int rc = xcecli_notify_jobfail(xcecli_ctx, &failinfo); + if (rc < 0) { + char jobidStr[64]; + + fmt_jobid(jobidStr, sizeof(jobidStr), jobid); + XCECD_LOG(XCE_LOG_WARN, + "job %s failure report not acknowledged by the controller " + "(rc %d) after retries -- the rejoin resync will reconcile " + "it", jobidStr, rc); + } + return rc; } /** @@ -1915,7 +1935,18 @@ int xcecd_job_finished(xcecd_job_t* job) }; XCECD_LOG(XCE_LOG_DEBUG, "calling xcecli_jobfinish..."); - return xcecli_notify_jobfinish(xcecli_ctx, &myjob); + /* bug-1741: acknowledged -- resent until the controller confirms. */ + int rc = xcecli_notify_jobfinish(xcecli_ctx, &myjob); + if (rc < 0) { + char jobidStr[64]; + + fmt_jobid(jobidStr, sizeof(jobidStr), job->job_id); + XCECD_LOG(XCE_LOG_WARN, + "job %s finish report not acknowledged by the controller " + "(rc %d) after retries -- the rejoin resync will reconcile " + "it", jobidStr, rc); + } + return rc; } /**