feat: acknowledged job start/finish/fail reports (bug-1741)

A worker reported job start, finish and failure with a fire-and-forget
OPCode_notify the controller never answers, so a report lost while the
worker stayed connected (the live incident: 6 of 7 finish reports in a
burst) left the job RUN forever; the rejoin resync (165fc158) only
recovers reports lost around a reconnect.

- controller.capnp: OPCode_notifyAcked -- the same NotifyData body,
  answered. A separate opcode because replies are matched through the
  client's single wbuf, not by sequence number: answering plain notifies
  would feed stale statuses to the next synchronous request.
- controller (cluster.c): do_notify_acked() applies the fact like
  do_notify, then waits on the commit barrier and replies 0 only once it
  is majority-committed. A barrier failure (the fact is still appended and
  retained) replies -E_XCE_NOT_READY, so the worker retries in place
  instead of rotating away from a healthy primary (cf. bug-1850). Admitted
  by the no-quorum fact-bearing gate like OPCode_notify.
- libxce: xce_notify_acked_req/xce_notify_acked (send, wait for the
  status) and xcecli_notify_acked (XCECLI_FAILOVER_CALL retry/failover).
  xcecli_notify_jobstart/jobfinish/jobfail use it; on -E_XCE_BAD_OPCODE
  (a controller predating the opcode) they fall back to the plain
  fire-and-forget notify, so workers and controllers can be upgraded in
  either order.
- Resends are safe: START/FINISH are duplicate-guarded and a repeated FAIL
  yields the same state.
- worker: logs a WARN when a report is still unacknowledged after the
  retry budget (the rejoin resync remains the backstop). These calls run
  on the worker daemon's IPC callback, so while the controller lacks
  quorum other jobs' IPC messages wait (buffered, not lost) up to the
  retry budget.

Tests: test_notify_acked_reply_only_after_commit and
test_notify_acked_withheld_replies_not_ready (no reply before the covering
ACK; -E_XCE_NOT_READY with the fact retained when ACKs are withheld).
Written with the opcode, so no separate RED run. Full /xce-build all:
111 suites pass; controller suite 1245/0.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

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;

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:

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 */

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*);

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*);

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*);

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)

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 */

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;

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") {

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;
 }
 
 /**