worker: re-report job state on transparent failover reconnect

Covers the case where the worker process stays alive and only the TCP link
to the controller drops (or the controller fails over): libxce's failover
loop reconnects transparently and xcecli_hello() returns 0, so
checking_connection() sees ret==0 and never takes its
xcecd_join_cluster()/resend_job_status() path — job state changes that
occurred while disconnected are never re-reported, leaving stale RUN/UNKWN
jobs on the controller.

- worker/controller_connector.c: new xcecd_reconnect_cb registered via
  xcecli_set_reconnect_handler after a successful join, resending the actual
  job state (bug-1509 semantics) on every transparent reconnect.
- lib/client.c (bug-1522): xcecli_start no longer resets rejoin_cb/rejoin_arg
  to NULL — it is initialized by xcecli_ctx_init's calloc, and a pre-start
  registration (scheduler sched_reconnect_cb, worker xcecd_reconnect_cb) must
  survive first connect; the failover loop invokes it after a transparent
  reconnect (xce_ctx_reconnect), which never re-runs xcecli_start. Previously
  every pre-start registration was dead code.
- scheduler.c: move xcecli_set_reconnect_handler to AFTER a successful join —
  xcecli_join also runs the failover loop, and a candidate-iteration rotate
  (standby redirect) must not fire the rejoin hook before the session exists
  (was firing sched_reconnect_cb during initial join → rejoin on
  uninitialized state → invalid-opcode crash).
- test_libxce_retry.c: regression test for bug-1522
  (test_reconnect_handler_survives_start_state_reset). Suite 296/296.

Functional verification on the test cluster: worker joins (host OK), runs a
job, and after a controller-link break re-joins ("Rejoined cluster
successfully after retry N"). Scheduler stays up (no invalid-opcode crash).

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
lei.chen 2026-08-15 08:40:20 +00:00
parent ee791791bd
commit 9d7628ef3f
6 changed files with 160 additions and 5 deletions

View file

@ -123,6 +123,33 @@ Covered in `.kiro/steering/ha-async-io.md` (connection ownership, async I/O, hea
## Key Files
## Current Functional Status (2026-08-15)
### Completed (HA / reliability layer is mature)
- **Raft/HA family fully landed**: term-safety, commit-durability (ack ⇒ durable on majority), commit-barrier-rollout, commit-engine, wal-replication, apply-decoupling (canonical pre-hooks → commit → apply → notify), single-leader-role, convergence-recovery, config-bootstrap, libxce-failover, controller-status (18/18 tasks).
- **Job-array single WAL record** (job-array-single-wal-record).
- **Scheduler message-gap** G-4/G-5/G-6/G-8 largely implemented (`sched_update_existing_job`, rusage storage, distinct UPDATE_JOB_FIELDS op).
- **job_ctrl** (suspend/resume/hold/release/kill) and **queue_ctrl** (open/close/inactivate) exist in `job.c`/`queue.c`.
### Gaps / missing functionality
These are the main functional shortfalls (see `.kiro/steering/roadmap.md` for full backlog):
| Area | Gap | Location / note |
|---|---|---|
| **Job control residual** | `job_ctrl` direct-mutate path (admin suspend/resume) still mutates state then fires hook — not commit-then-apply (same family as bug-795/828/851/1060, narrowly scoped to admin suspend/resume) | roadmap Hardening: "investigate: job_ctrl direct-mutate path" |
| **Worker rejoin resync** | On rejoin/failover, worker does NOT re-report running/finished job states (`xce_join_req` carries only `hostattrs_t`, no job state) → facts sent during disconnect/failover window are lost | roadmap backlog: worker-rejoin-resync |
| **Component failover adoption** | Daemons still pass single endpoint (`conf->masters[0]`) to libxce; bounded-retry liveness not yet verified per component | roadmap backlog: component-failover-adoption |
| **Config-upgrade snapshot** | `do_upgrade_config_with_bytes` does config-push+restart+per-node replay — can leave divergent materialized state (bug-857); needs entity-bearing snapshots (HARD dep, still 16-byte LSN sentinel) | roadmap backlog: config-upgrade-snapshot, entity-bearing-snapshots |
| **Job-array bulk query/cancel** | Single client op to query/cancel whole array not implemented (per-element only) | roadmap backlog: bulk job-array query/cancel |
| **Preemption** | No mechanism to preempt lower-priority running jobs (high-priority waits for free resources) | roadmap backlog / scheduler-preemption spec |
| **Group building** | `cluster_build_groups()` is an empty TODO stub | `cluster.c:997` |
| **API-server fact delivery** | FIXMEs: job pid / exit code / fail reason not sent to apiserver | `apiserver.c:248,283,318` |
| **Peer auth** | `validate_host()` is a stub returning success for all peers; `OP_upgradeConfig`/`OP_replConfig` unauthenticated (trusted-LAN model) | roadmap Hardening: validate_host, peer-authentication-hardening (security-critical) |
| **Admin login** | xceadmin uses zero-config TLS, no login/token/session/audit | roadmap Hardening: peer-authentication-hardening (d) |
### Note
Reliability (HA/durability/consistency) is the mature layer; the gaps above are mostly **business/operational features** (bulk ops, preemption, auth, rejoin resync) plus a few correctness residuals (job_ctrl mutate-before-commit). See also `src/scheduler/CLAUDE.md` for scheduler-side status.
## See Also
- Shared library: `src/lib/CLAUDE.md`
- Root guidelines: `CLAUDE.md` (repo root)

View file

@ -179,8 +179,14 @@ int xcecli_start(xcecli_ctx_t c)
}
ctx->failover.cur_master = connected_idx;
ctx->failover.retry = XCE_RETRY_DEFAULTS;
ctx->failover.rejoin_cb = NULL;
ctx->failover.rejoin_arg = NULL;
/* Do NOT reset rejoin_cb/rejoin_arg here: xcecli_ctx_init() calloc's the
* failover state (both NULL), and a consumer that registered a reconnect
* handler BEFORE start (scheduler sched_reconnect_cb, worker
* xcecd_reconnect_cb) must survive first connect — the failover loop
* invokes it after a TRANSPARENT reconnect (xce_ctx_reconnect), which
* never re-runs xcecli_start. Clearing it here silently disabled every
* pre-start registration (the callback was dead code). */
rc = xce_ctx_start(&(ctx->ctx));

View file

@ -1423,6 +1423,44 @@ static void test_invoke_rejoin_null_ctx_safe(void)
printf("test_invoke_rejoin_null_ctx_safe: PASS\n");
}
/* Regression (bug-1522): a reconnect handler registered BEFORE xcecli_start
* must survive the failover-state reset xcecli_start performs on first
* connect. xcecli_start sets cur_master/retry but must NOT clear
* rejoin_cb/rejoin_arg — the failover loop invokes the hook after a
* TRANSPARENT reconnect (xce_ctx_reconnect), which never re-runs
* xcecli_start, so clearing it here would make every pre-start registration
* dead code (scheduler sched_reconnect_cb, worker xcecd_reconnect_cb). */
static void test_reconnect_handler_survives_start_state_reset(void)
{
xce_client_config_t conf;
api_ctx_t ac;
fo_make_client_masters(&conf);
fo_make_api_ctx(&ac, &conf);
fo_rejoin_probe_t probe = { 0 };
probe.ret = 0;
/* register BEFORE the start-equivalent failover-state update */
TEST_ASSERT_EQUAL_INT(
0, xcecli_set_reconnect_handler((xcecli_ctx_t)&ac, fo_rejoin_cb,
&probe));
TEST_ASSERT_TRUE(ac.failover.rejoin_cb == fo_rejoin_cb);
TEST_ASSERT_TRUE(ac.failover.rejoin_arg == &probe);
/* xcecli_start's state reset: cursor + retry tunables are (re)initialised,
* the rejoin hook is left untouched (bug-1522 fix). */
ac.failover.cur_master = 0;
ac.failover.retry = XCE_RETRY_DEFAULTS;
/* after the reset the hook is still registered and fires on rejoin */
TEST_ASSERT_TRUE(ac.failover.rejoin_cb == fo_rejoin_cb);
TEST_ASSERT_TRUE(ac.failover.rejoin_arg == &probe);
TEST_ASSERT_EQUAL_INT(0, xcecli_invoke_rejoin(&ac));
TEST_ASSERT_EQUAL_INT(1, probe.called);
printf("test_reconnect_handler_survives_start_state_reset: PASS\n");
}
/* ════════════════════════════════════════════════════════════════════════
* Task 5.1 — XcecliRetryLoop wrapping + per-ctx lock (CI2 concurrency).
*
@ -1709,6 +1747,7 @@ void test_libxce_retry(void)
RUN_TEST(test_invoke_rejoin_no_hook_is_noop);
RUN_TEST(test_set_reconnect_handler_null_safe_and_clears);
RUN_TEST(test_invoke_rejoin_null_ctx_safe);
RUN_TEST(test_reconnect_handler_survives_start_state_reset);
RUN_TEST(test_failover_call_serialises_shared_ctx);
RUN_TEST(test_failover_call_success_runs_once);
RUN_TEST(test_ff_needs_reconnect_on_err);

View file

@ -17,6 +17,29 @@ meson build && ninja -C build
Links against `libxce` from `src/lib/`.
## Current scheduling policies (2026-08-15)
There is exactly **one** job-selection policy today, plus one constraint hook and two
allocators. The plugin registry (`scheduler.c:356-359`) registers all four; the job
picker is selected by a **hardcoded** name in `sched_job_picker()` (`scheduler.c:44`,
`job_pick_plugin = "xce_sched_fcfs"`) — there is no runtime way to swap it.
| Plugin | Role | Notes |
|---|---|---|
| `fcfs.c` | **Job picker / policy** | Only election plugin. Orders queues by priority, jobs by (priority desc, queue_time asc). Runs in a dedicated `jobpicker` thread; emits picked jobid via `jbuf` cbuf. |
| `limit.c` | **Constraint hook, NOT a policy** | Hooks into `job_election_hook` + `prepare_allocation_hook`: per-queue/user limits/quota. Blocks over-limit jobs and attaches a PEND reason. Runs alongside the picker, does not choose jobs itself. |
| `seqjob_allocator.c` | **Allocator** | `maxTasks == 1` (serial) resource placement. |
| `parjob_allocator.c` | **Allocator (stub)** | `my_alloc()` is effectively empty — parallel/multi-slot jobs get no allocation. |
Election (pick) and allocation are two decoupled steps: fcfs picks *which* job, the
allocators decide *where* it runs. No fairshare / priority-weighted / backfill policy
exists yet.
**Future work** — see `.kiro/specs/scheduler-framework/brief.md` ("Current State" +
"What's missing"): configurable job picker (deferred), parallel allocation, iterator
generation, preemption, project-level scheduling, simulation mode. This doc is the
authoritative *current-state* note; the brief is the roadmap.
## Conventions
- C standard: gnu17
- Format with `clang-format`
@ -26,3 +49,4 @@ Links against `libxce` from `src/lib/`.
## See Also
- Shared library: `src/lib/CLAUDE.md`
- Root guidelines: `CLAUDE.md` (repo root)
- Framework roadmap: `.kiro/specs/scheduler-framework/brief.md`

View file

@ -506,7 +506,6 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config,
}
xcecli_set_handler(sched->apictx, handler, sched);
xcecli_set_reconnect_handler(sched->apictx, sched_reconnect_cb, sched);
/* HA mode: iterate candidates until one accepts as primary */
if (sched->cluster_cfg.candidates_count > 0) {
@ -536,8 +535,6 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config,
continue;
}
xcecli_set_handler(sched->apictx, handler, sched);
xcecli_set_reconnect_handler(sched->apictx, sched_reconnect_cb,
sched);
rc = xcecli_start(sched->apictx);
if (rc < 0) {
@ -548,6 +545,12 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config,
rc = xcecli_join(sched->apictx, HostRole_scheduler, NULL);
if (rc == 0) {
joined = 1;
/* Register AFTER a successful join: xcecli_join also runs the
* failover loop, and a candidate-iteration rotate (standby
* redirect) must NOT fire the rejoin hook before the session
* exists. Also after start(), which resets rejoin_cb. */
xcecli_set_reconnect_handler(sched->apictx,
sched_reconnect_cb, sched);
break;
}
@ -570,6 +573,9 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config,
if (rc < 0) {
return rc;
}
/* After successful join (and after start, which resets rejoin_cb) —
* same reasoning as the HA path above. */
xcecli_set_reconnect_handler(sched->apictx, sched_reconnect_cb, sched);
}
rc = pthread_create(&(sched->jobpicker), NULL, &sched_job_picker, sched);

View file

@ -160,6 +160,44 @@ static void init_client_config(xce_client_config_t* config)
config->key);
}
/**
* @brief Reconnect callback — invoked by libxce's transparent failover after
* a connection is re-established (and the ctx recovered), BEFORE the
* interrupted op resumes.
*
* This covers the case where the worker process stays alive and only the
* TCP link to the controller dropped (or the controller failed over): the
* failover retry loop in xcecli_hello() reconnects transparently and returns
* success, so checking_connection() sees ret==0 and never takes its
* xcecd_join_cluster()/resend_job_status() path. Without this hook, job
* state changes that occurred while disconnected are never re-reported and
* the controller is left with stale RUN/UNKWN jobs.
*
* @return 0 to let the interrupted op resume; <0 would be treated by the
* failover loop as "new primary rejected rejoin" and it would rotate.
*/
static int xcecd_reconnect_cb(xcecli_ctx_t c, void* arg)
{
(void)c;
(void)arg;
if (cc_ctx == NULL || cc_ctx->worker_list == NULL
|| *cc_ctx->worker_list == NULL) {
return 0;
}
XCECD_LOG(XCE_LOG_INFO,
"Controller connection restored; resending job status");
/* Same semantics as checking_connection()'s rejoin resend (bug-1509
* B2/B3): report the ACTUAL job state so the controller reconciles its
* view (clears UNKWN on a live claim; converges to a terminal state for
* processes that are gone). */
(void)resend_job_status(*cc_ctx->worker_list);
return 0;
}
/**
* @brief Handles I/O events for the endpoint, focusing on error conditions
* and connection liveness checks using EPOLLIN/EPOLLOUT.
@ -267,6 +305,17 @@ int xcecd_join_cluster(hostattrs_t* attrs)
XCECD_LOG(XCE_LOG_INFO,
"xcecd_join_cluster: joined primary at %s:%d",
host, port);
/* Re-report running job state when libxce transparently
* reconnects (failover) — checking_connection() only covers
* the explicit xcecd_join_cluster path, not the auto-reconnect
* path. Registered AFTER a successful join: xcecli_join also
* runs the failover loop, and a candidate-iteration rotate
* (standby redirect) must NOT fire the rejoin hook before the
* session exists. */
xcecli_set_reconnect_handler(xcecli_ctx, xcecd_reconnect_cb,
NULL);
return 0;
}
@ -299,6 +348,10 @@ int xcecd_join_cluster(hostattrs_t* attrs)
return rc;
}
/* Same reconnect hook as the HA path above — after a successful join so a
* candidate-iteration rotate never fires it pre-session. */
xcecli_set_reconnect_handler(xcecli_ctx, xcecd_reconnect_cb, NULL);
return 0;
}