Merge branch 'working' into job_peek_lchen

job_peek_lchen: Add fetchfile feature (job output/error peek proxy)
working: HA/Raft infrastructure (consensus, replication, config mgmt)

Conflict resolution summary:
- controller.capnp: fetchfile @41, fetchfilereply @18, fetchfilemsg @30
- cluster.c: bifurcated OPCode_reply for fetchfile + config-push ACK
- job.c: kept full fetchfile pending-request subsystem (~183 lines)
- request.c: kept both xce_fetchfile_req() and XCE_RETRY_DEFAULTS
- helper.c: added missing preVoteRequest/preVoteReply mappings
- All 15 new test registrations (test_cbuf.c + 14 HA tests)
This commit is contained in:
Lei Chen 2026-07-22 13:57:25 +08:00
commit 3a2094a461
529 changed files with 345174 additions and 5048 deletions

View file

@ -1,31 +1,21 @@
#include <errno.h>
#include <pthread.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include "client.h"
#include "client_internal.h"
#include "controller.capnp.h"
#include "mylock.h"
#include "request.h"
#include "xce/errnox.h" /* E_XCE_NOT_READY */
#include "xce/types.h"
#include "xceapi.h"
#include "xcectx.h"
#include "xlist.h"
typedef enum {
API_CTX_INIT = 0,
API_CTX_RUN,
API_CTX_DOWN,
API_CTX_ERR
} api_ctx_state_t;
typedef struct {
xce_client_config_t* conf;
endpoint_t ep;
xce_ctx_t ctx;
} api_ctx_t;
int xcecli_ctx_init(xcecli_ctx_t* c, xce_client_config_t* cfg)
{
api_ctx_t* ctx;
@ -88,6 +78,27 @@ int xcecli_set_handler(xcecli_ctx_t c, xce_request_handler_t handler,
return xce_ctx_set_handler(&(ctx->ctx), handler, args);
}
/*
* xcecli_set_reconnect_handler (task 4.1) — register the consumer rejoin hook
* the failover loop invokes after a transparent reconnect, before resuming the
* interrupted op. A NULL cb clears any previously registered hook (back to the
* pure-client no-op path). Stores cb+arg on the per-ctx failover state.
*/
int xcecli_set_reconnect_handler(xcecli_ctx_t c, xcecli_reconnect_cb cb,
void* arg)
{
api_ctx_t* ctx = (api_ctx_t*)c;
if (c == NULL) {
return -EINVAL;
}
ctx->failover.rejoin_cb = cb;
ctx->failover.rejoin_arg = arg;
return 0;
}
int xcecli_wakeup(xcecli_ctx_t c, int status)
{
api_ctx_t* ctx = (api_ctx_t*)c;
@ -112,19 +123,62 @@ int xcecli_try_wakeup(xcecli_ctx_t c, int status)
int xcecli_start(xcecli_ctx_t c)
{
int rc;
int rc = -ECONNREFUSED;
api_ctx_t* ctx = (api_ctx_t*)c;
int i;
int connected_idx = -1;
if (c == NULL) {
return -EINVAL;
}
rc = endpoint_connect(&(ctx->ep), ctx->conf->masters[0],
ctx->conf->master_ports[0]);
/* Initial connect rotates through ALL configured masters, not just
* masters[0] (bug-854 dead-first-master): if the first master is down
* (connection-refused after a failover that killed it), a hardcoded
* masters[0] connect fails the whole `client` session before the failover
* loop — which only runs on an ALREADY-connected ctx — ever gets a chance
* to rotate. Try each master in order; the first that connects becomes the
* sticky cursor start. The subsequent ctrlstatus/op failover loop then
* rotates onward to the authoritative primary as needed.
*
* ctx->ep was endpoint_init'd + bound to the iomgr in xcecli_init. A failed
* endpoint_connect leaves the socket fd unusable, so before retrying the
* NEXT master we must rebuild the endpoint exactly as xce_ctx_reconnect
* does: endpoint_fini → endpoint_init (fresh fd) → re-set peer + re-bind the
* iomgr (endpoint_connect needs ep->iomgr->ssl_ctx). */
for (i = 0; i < ctx->conf->nmasters; i++) {
if (i > 0) {
endpoint_fini(&(ctx->ep));
if (endpoint_init(&(ctx->ep), AF_INET, SOCK_STREAM, 0, 0) < 0)
continue;
endpoint_set_peer(&(ctx->ep), &(ctx->ctx.myinfo));
ctx->ep.iomgr = &(ctx->ctx.iomgr);
}
rc = endpoint_connect(&(ctx->ep), ctx->conf->masters[i],
ctx->conf->master_ports[i]);
if (rc >= 0) {
connected_idx = i;
break;
}
}
if (connected_idx < 0) {
/* No master reachable at all — surface the last transport error. */
endpoint_fini(&(ctx->ep));
return rc;
}
/* Initialize per-context failover state: sticky primary starts at the master
* we actually connected to (not always index 0), retry tunables default to
* XCE_RETRY_DEFAULTS, and no rejoin hook is installed for pure clients. */
rc = mylock_init(&(ctx->failover.lock));
if (rc < 0) {
endpoint_fini(&(ctx->ep));
return rc;
}
ctx->failover.cur_master = connected_idx;
ctx->failover.retry = XCE_RETRY_DEFAULTS;
ctx->failover.rejoin_cb = NULL;
ctx->failover.rejoin_arg = NULL;
rc = xce_ctx_start(&(ctx->ctx));
@ -141,6 +195,7 @@ int xcecli_stop(xcecli_ctx_t c)
xce_ctx_stop(&(ctx->ctx));
endpoint_fini(&(ctx->ep));
mylock_fini(&(ctx->failover.lock));
return 0;
}
@ -175,7 +230,8 @@ int xcecli_hello(xcecli_ctx_t c, char* msg)
return -EINVAL;
}
return xce_hello(&(ctx->ctx), msg);
return XCECLI_FAILOVER_CALL(ctx, /*is_write=*/false,
xce_hello(&(ctx->ctx), msg));
}
int xcecli_nop(xcecli_ctx_t c)
@ -186,7 +242,10 @@ int xcecli_nop(xcecli_ctx_t c)
return -EINVAL;
}
return xce_nop(&(ctx->ctx));
/* Fire-and-forget: best-effort transport reconnect on a known-dead
* connection, then a single-shot send. No status classification, no
* no-primary backoff, no retry budget (task 5.2). */
return XCECLI_FF_CALL(ctx, xce_nop(&(ctx->ctx)));
}
int xcecli_heartbeat(xcecli_ctx_t c, int hb_interval)
@ -197,7 +256,10 @@ int xcecli_heartbeat(xcecli_ctx_t c, int hb_interval)
return -EINVAL;
}
return xce_heartbeat(&(ctx->ctx), hb_interval);
/* Fire-and-forget: best-effort transport reconnect on a known-dead
* connection, then a single-shot send (task 5.2). A heartbeat must NEVER go
* through the no-primary backoff retry budget. */
return XCECLI_FF_CALL(ctx, xce_heartbeat(&(ctx->ctx), hb_interval));
}
int xcecli_shutdown(xcecli_ctx_t c)
@ -219,7 +281,8 @@ int xcecli_reslist(xcecli_ctx_t c, xce_resource_t*** res)
return -EINVAL;
}
return xce_reslist(&(ctx->ctx), res);
return XCECLI_FAILOVER_CALL(ctx, /*is_write=*/false,
xce_reslist(&(ctx->ctx), res));
}
int xcecli_join(xcecli_ctx_t c, int role, hostattrs_t* attrs)
@ -229,7 +292,8 @@ int xcecli_join(xcecli_ctx_t c, int role, hostattrs_t* attrs)
return -EINVAL;
}
return xce_join(&(ctx->ctx), role, attrs);
return XCECLI_FAILOVER_CALL(ctx, /*is_write=*/true,
xce_join(&(ctx->ctx), role, attrs));
}
int xcecli_workerlist(xcecli_ctx_t c, xce_worker_t*** workers)
@ -239,7 +303,8 @@ int xcecli_workerlist(xcecli_ctx_t c, xce_worker_t*** workers)
return -EINVAL;
}
return xce_workerlist(&(ctx->ctx), workers);
return XCECLI_FAILOVER_CALL(ctx, /*is_write=*/false,
xce_workerlist(&(ctx->ctx), workers));
}
int xcecli_jobdispatch(xcecli_ctx_t c, xce_job_dispatch_t* info)
@ -250,7 +315,8 @@ int xcecli_jobdispatch(xcecli_ctx_t c, xce_job_dispatch_t* info)
return -EINVAL;
}
return xce_jobdispatch(&(ctx->ctx), info);
return XCECLI_FAILOVER_CALL(ctx, /*is_write=*/true,
xce_jobdispatch(&(ctx->ctx), info));
}
int xcecli_jobsubmit(xcecli_ctx_t c, xce_job_submit_t* submit, uint64_t* jobid)
@ -261,7 +327,8 @@ int xcecli_jobsubmit(xcecli_ctx_t c, xce_job_submit_t* submit, uint64_t* jobid)
return -EINVAL;
}
return xce_jobsubmit(&(ctx->ctx), submit, jobid);
return XCECLI_FAILOVER_CALL(ctx, /*is_write=*/true,
xce_jobsubmit(&(ctx->ctx), submit, jobid));
}
int xcecli_joblist(xcecli_ctx_t c, xce_job_t*** jobs)
@ -272,7 +339,8 @@ int xcecli_joblist(xcecli_ctx_t c, xce_job_t*** jobs)
return -EINVAL;
}
return xce_joblist(&(ctx->ctx), jobs);
return XCECLI_FAILOVER_CALL(ctx, /*is_write=*/false,
xce_joblist(&(ctx->ctx), jobs));
}
int xcecli_clusterconf(xcecli_ctx_t c, xce_clusterconf_t** conf)
@ -283,7 +351,165 @@ int xcecli_clusterconf(xcecli_ctx_t c, xce_clusterconf_t** conf)
return -EINVAL;
}
return xce_clusterconf(&(ctx->ctx), conf);
return XCECLI_FAILOVER_CALL(ctx, /*is_write=*/false,
xce_clusterconf(&(ctx->ctx), conf));
}
int xcecli_ctrlstatus(xcecli_ctx_t c, xce_ctrlstatus_t** status)
{
api_ctx_t* ctx = (api_ctx_t*)c;
if (c == NULL) {
return -EINVAL;
}
return XCECLI_FAILOVER_CALL(ctx, /*is_write=*/false,
xce_ctrlstatus(&(ctx->ctx), status));
}
/*
* The primary's responder role on the wire is ctrl_role_t PRIMARY. That enum
* lives in the controller-private ha.h (CTRL_PRIMARY == 0) and is intentionally
* not exposed to libxce; the wire ordinal is part of the ctrlstatus reply
* contract (controller/ctrlstatus.c stamps responderRole = ha->role). We pin it
* here as a named constant so the comparison reads intentionally rather than as
* a bare 0. Any reorder of ctrl_role_t is a Revalidation Trigger for this spec.
*/
#define XCECLI_CTRLSTATUS_ROLE_PRIMARY ((uint8_t)0)
bool xcecli_ctrlstatus_is_authoritative(const xce_ctrlstatus_t* status)
{
if (status == NULL) {
return false;
}
return status->responderRole == XCECLI_CTRLSTATUS_ROLE_PRIMARY;
}
int xcecli_ctrlstatus_primary_addr(const xce_ctrlstatus_t* status,
char* host_out, size_t host_sz,
uint16_t* port_out)
{
int i;
if ((status == NULL) || (host_out == NULL) || (host_sz == 0)
|| (port_out == NULL)) {
return -EINVAL;
}
/* No known primary (empty/NULL primaryName) ⇒ nothing to target; the caller
* falls back to plain master rotation (Req 2.3 unknown-primary). */
if ((status->primaryName == NULL) || (status->primaryName[0] == '\0')) {
return -ENOENT;
}
/* Map the reported primaryName to its configured address via the reply's
* peers list. The list is a small bounded set (configured controllers). */
for (i = 0; i < status->peers.count; i++) {
xce_ctrlpeer_t* p = status->peers.list[i];
if ((p == NULL) || (p->name == NULL) || (p->host == NULL)) {
continue;
}
if (strcmp(p->name, status->primaryName) != 0) {
continue;
}
if ((p->host[0] == '\0') || (p->port == 0)) {
/* named primary has no usable address — treat as absent */
break;
}
strncpy(host_out, p->host, host_sz - 1);
host_out[host_sz - 1] = '\0';
*port_out = (uint16_t)p->port;
return 0;
}
/* primaryName not present in peers (or no usable address): fall back to
* rotation rather than targeting a bogus address. */
return -ENOENT;
}
int xcecli_ctrlstatus_primary(xcecli_ctx_t c, xce_ctrlstatus_t** status)
{
api_ctx_t* ctx = (api_ctx_t*)c;
xce_ctrlstatus_t* hint;
int rc;
int attempt;
int max_attempts;
if ((c == NULL) || (status == NULL)) {
return -EINVAL;
}
/* Bound the redirect loop by the configured failover retry budget so we
* never spin unbounded chasing a never-appearing primary (each inner
* xcecli_ctrlstatus already runs its own full reach-some-node failover).
* Floor it at the master-ring size so a standby-first / dead-first master
* list is always traversable in full before giving up (bug-854): one extra
* attempt lets the rotation that lands on the primary actually query it. */
max_attempts = ctx->failover.retry.max_retries;
if (max_attempts < 1) {
max_attempts = 1;
}
if (ctx->conf != NULL && ctx->conf->nmasters + 1 > max_attempts) {
max_attempts = ctx->conf->nmasters + 1;
}
for (attempt = 0; attempt < max_attempts; attempt++) {
hint = NULL;
rc = xcecli_ctrlstatus(c, &hint);
if (rc < 0) {
/* no reachable controller within the inner budget ⇒ typed failure;
* *status untouched (never a hint reply as the result, Req 9.2). */
return rc;
}
if (xcecli_ctrlstatus_is_authoritative(hint)) {
/* the primary itself answered — authoritative result */
*status = hint;
return 0;
}
/* Non-authoritative reply (standby/electing/init). The inner
* xcecli_ctrlstatus SUCCEEDS against ANY reachable node — ctrlstatus is
* answerable in every role — so it classifies the standby's status-0
* reply as success and NEVER rotates off it on its own. We must advance
* the endpoint here, or the loop re-queries the same non-primary every
* attempt and exhausts the budget without ever reaching the primary
* (bug-854). Capture a redirect hint when the responder named the primary
* (1-hop direct redirect), then FORCE rotation: xce_failover_rotate
* consumes the hint when set, else advances to the next configured
* master; recover repairs the ctx so the next attempt's send proceeds.
* Mirror xcecli_retry_next: only recover when the rotate succeeded — a
* failed rotate leaves the ctx for the next inner call's own transport-
* error failover to heal. All under failover.lock (the inner call takes
* and releases it itself, so it is not held here). */
{
char phost[XCE_MAXHOSTNAME_LEN];
uint16_t pport = 0;
mylock_lock(&ctx->failover.lock);
if (xcecli_ctrlstatus_primary_addr(hint, phost, sizeof(phost),
&pport)
== 0) {
xce_ctx_capture_redirect_hint(&ctx->ctx, phost, pport);
}
if (xce_failover_rotate(ctx) == 0) {
xcecli_ctx_recover(ctx);
}
mylock_unlock(&ctx->failover.lock);
}
/* Free the non-authoritative interim reply before the next attempt — no
* leak, and we never return a hint as the authoritative result. */
xcecli_free_ctrlstatus(hint);
}
/* Budget exhausted without the primary answering: typed failure, *status
* untouched (Req 9.2 — never return a non-primary reply as authoritative).
*/
return -E_XCE_NOT_READY;
}
int xcecli_notify_jobfinish(xcecli_ctx_t c, xce_jobfinish_t* jfinish)
@ -318,7 +544,8 @@ int xcecli_update_resvals(xcecli_ctx_t c, int nres, xce_resval_t** vals)
return -EINVAL;
}
rc = xce_update_resvals(&(ctx->ctx), nres, vals);
rc = XCECLI_FAILOVER_CALL(ctx, /*is_write=*/true,
xce_update_resvals(&(ctx->ctx), nres, vals));
return rc;
}
@ -334,8 +561,10 @@ int xcecli_update_jusage(xcecli_ctx_t c, uint64_t jobid, int npids, int* pids,
return -EINVAL;
}
rc = xce_update_jusage(&(ctx->ctx), jobid, npids, pids, npgids, pgids,
nusage, usages);
rc = XCECLI_FAILOVER_CALL(
ctx, /*is_write=*/true,
xce_update_jusage(&(ctx->ctx), jobid, npids, pids, npgids, pgids, nusage,
usages));
return rc;
}
@ -352,7 +581,8 @@ int xcecli_resval(xcecli_ctx_t c, char* host, char* res, xce_resval_t*** val)
return -EINVAL;
}
return xce_resval(&(ctx->ctx), host, res, val);
return XCECLI_FAILOVER_CALL(ctx, /*is_write=*/false,
xce_resval(&(ctx->ctx), host, res, val));
}
int xcecli_notify(xcecli_ctx_t c, xce_notify_data_t* data)
@ -367,7 +597,12 @@ int xcecli_notify(xcecli_ctx_t c, xce_notify_data_t* data)
return -EINVAL;
}
return xce_notify(&(ctx->ctx), data);
/* Fire-and-forget chokepoint: every thin notify_* wrapper delegates
* here, so applying the best-effort reconnect at this single point covers
* them all without double-wrapping/double-locking. Best-effort transport
* reconnect on a known-dead connection, then a single-shot send (task
* 5.2). */
return XCECLI_FF_CALL(ctx, xce_notify(&(ctx->ctx), data));
}
int xcecli_notify_jobstart(xcecli_ctx_t c, job_pidinfo_t* pidinfo)
@ -417,7 +652,8 @@ int xcecli_jobctrl(xcecli_ctx_t c, xce_jobctrl_t* ctrlinfo)
return -EINVAL;
}
rc = xce_jobctrl(&(ctx->ctx), ctrlinfo);
rc = XCECLI_FAILOVER_CALL(ctx, /*is_write=*/true,
xce_jobctrl(&(ctx->ctx), ctrlinfo));
return rc;
}
@ -511,7 +747,8 @@ int xcecli_queuectrl(xcecli_ctx_t c, xce_queuectrl_t* ctrlinfo)
return -EINVAL;
}
rc = xce_queuectrl(&(ctx->ctx), ctrlinfo);
rc = XCECLI_FAILOVER_CALL(ctx, /*is_write=*/true,
xce_queuectrl(&(ctx->ctx), ctrlinfo));
return rc;
}
@ -525,7 +762,8 @@ int xcecli_hostctrl(xcecli_ctx_t c, xce_hostctrl_t* ctrlinfo)
return -EINVAL;
}
rc = xce_hostctrl(&(ctx->ctx), ctrlinfo);
rc = XCECLI_FAILOVER_CALL(ctx, /*is_write=*/true,
xce_hostctrl(&(ctx->ctx), ctrlinfo));
return rc;
}
@ -542,7 +780,8 @@ int xcecli_runtask(xcecli_ctx_t c, xce_taskrun_t* info)
* controller to trace back
* client
*/
rc = xce_runtask(&(ctx->ctx), info);
rc = XCECLI_FAILOVER_CALL(ctx, /*is_write=*/true,
xce_runtask(&(ctx->ctx), info));
return rc;
}
@ -634,7 +873,8 @@ int xcecli_set_pend_reason(xcecli_ctx_t c, xce_preason_set_t* reasons)
return -EINVAL;
}
rc = xce_set_pend_reason(&(ctx->ctx), reasons);
rc = XCECLI_FAILOVER_CALL(ctx, /*is_write=*/true,
xce_set_pend_reason(&(ctx->ctx), reasons));
return rc;
}
@ -657,7 +897,8 @@ int xcecli_clear_reason(xcecli_ctx_t c, int njobs, uint64_t* jobs)
return 0;
}
rc = xce_clear_reason(&(ctx->ctx), &reason);
rc = XCECLI_FAILOVER_CALL(ctx, /*is_write=*/true,
xce_clear_reason(&(ctx->ctx), &reason));
return rc;
}
@ -670,7 +911,8 @@ int xcecli_jobinfo(xcecli_ctx_t c, uint64_t jobid, xce_job_t** job)
return -EINVAL;
}
rc = xce_jobinfo(&(ctx->ctx), jobid, job);
rc = XCECLI_FAILOVER_CALL(ctx, /*is_write=*/false,
xce_jobinfo(&(ctx->ctx), jobid, job));
return rc;
}