diff --git a/.kiro/specs/component-failover-adoption/tasks.md b/.kiro/specs/component-failover-adoption/tasks.md index 82552000..6f041c21 100644 --- a/.kiro/specs/component-failover-adoption/tasks.md +++ b/.kiro/specs/component-failover-adoption/tasks.md @@ -68,7 +68,7 @@ - _Requirements: 7.1, 7.2, 7.3, 7.4, 2.5_ - _Depends: 3.1_ -- [ ] 3.3 Validate scheduler allocation/decision backpressure behavior and add connect/reconnect test coverage +- [x] 3.3 Validate scheduler allocation/decision backpressure behavior and add connect/reconnect test coverage - Confirm the bounded, occupancy-dependent buffering behavior: the scheduler's allocation/decision processing continues briefly past the point a retry begins, then blocks until connectivity is restored and resumes automatically — assert the worst-case (least-slack) occupancy scenario explicitly, not only the most-generous one, since that is the guarantee actually being made - Add end-to-end coverage: scheduler starts and joins across a multi-candidate config; killing the joined primary results in reconnect and re-join without a scheduler restart - Add coverage for the case where every configured candidate is unreachable or rejects the join at startup: assert the scheduler reports a clear failure identifying every candidate it attempted @@ -168,3 +168,4 @@ - Task 2.2: fixed the startup-ordering bug by threading `cluster_cfg` by value through `xcecd_config_t` → `xcecd_context_t` → `cc_ctx` (set in `main.c` before `xcecd_init()`/`xcecd_start()` run, instead of after). Safe as a plain struct copy because `xce_candidate_t.hostname` is a fixed `char[]` array, not a pointer — no aliasing hazard here (unlike the `xce_cluster_config_to_client_config()` output struct noted above). Scheduler (3.1) likely has the same ordering gap (`sched_init` loops candidates correctly already per design.md, so may not need this fix — verify against current source, don't assume). Rejoin-retry pattern established: `checking_connection()` checks `needs_rejoin` after every probe return, calls `xcecli_join()` directly (not full re-entry into the join-cluster function), clears the flag only on success — same pattern scheduler/apiserver rejoin tasks (3.1, 4.3) should follow. - Task 2.3 (real E2E caught what unit tests missed — 2 production bugs slipped through tasks 2.1/2.2's approved reviews): (1) `xcecli_start()` (`src/lib/client.c`) unconditionally resets `ctx->failover.rejoin_cb = NULL` as part of its own init — any `xcecli_set_reconnect_handler()` call before `xcecli_start()` is silently wiped. Register the reconnect handler AFTER `xcecli_start()` succeeds, not before. Scheduler/apiserver's own reconnect-handler registration (3.1, 4.3) must follow this same ordering or they'll hit the identical silent-wipe bug. (2) A successful `xcecli_join()`/worker join returns the assigned slot index as `reply.status` (any `rc >= 0`, not always exactly `0`) — checking `rc == 0` for success is wrong; use `rc < 0` as the only failure test (matches `xcecd_join_cluster()`'s own established convention). Any new code checking a join/rejoin call's success must use this same `rc < 0` convention, not `rc == 0`. (3) Security lesson: an env-var runtime bypass of a security check (`XCECD_ALLOW_NONROOT`) added for test/E2E convenience was rejected on review — the correct pattern for skipping a production-only check (root privilege, etc.) in a test/virtual build is a compile-time guard against the project's existing `VIRTUAL_CLUSTER` or `XCE_TESTING` flags, never a runtime environment variable reachable in the shipped production binary. - Task 3.2: implemented as a shared `sched_publish_retry_loop()`/`sched_publish_check_rejoin()` pair called from both `sched_publish_dispatch()` and `sched_publish_pend_reason()`, not literally inlined into `sched_publish_dispatch()` alone as design.md's prose names it — reviewed and judged compliant (same thread, fires once after the whole retry loop's successful return, not per-attempt, not in `sched_allocator`). Testing lesson for any future scheduler.c unit test needing to intercept `xcecli_*` calls: `xcecli_jobdispatch`/`xcecli_set_pend_reason`/`xcecli_join`/`xcecli_start`/`xcecli_set_reconnect_handler` all live in the single `client.c` translation unit inside `libxce.a` alongside symbols scheduler.c genuinely needs — the "test double linked ahead of the archive" trick (used by test_driver_decision_cb.c one layer up) causes duplicate-definition errors here. Use `-Wl,--wrap=` linker interposition instead (see `test_scheduler_decision_publish_retry.c`/its `meson.build` target) when a test needs scheduler.c itself linked live alongside intercepted libxce calls. Also confirmed: `sched_admin_request_test`'s `test_recordstart_symlink_escape_replies_error` fails independent of this spec's changes (its white-box `#include` list doesn't pull in `scheduler.c` at all) — a pre-existing, unrelated, environment-specific (root-in-sandbox symlink-permission) failure; do not re-investigate it as a regression in later tasks. +- Task 3.3: fixed a real production gap found along the way — `sched_init()`'s two all-candidates-unreachable/join-failure log lines logged only a candidate COUNT, never host:port names (same gap worker's task 2.3 fixed for the worker). Added `sched_format_candidate_list()`, threading `sched->cluster_cfg` through as a parameter rather than a new file-scope global (unlike worker's `g_cluster_cfg`) since `test_scheduler_connect.c` links `scheduler_srcs` as separate objects, not `#include`, so the `static` helper isn't directly callable from tests — coverage requires a real `dup2()`-based stdout capture around a live `sched_init()` call hitting the failure path, not a white-box direct call (see `capture_stdout_during_sched_init()` in `test_scheduler_connect.c` — reusable pattern for any future scheduler.c log-content assertion). Round-1 review caught this diagnostic fix shipping with zero test coverage (proven by reverting the 59-line diff and rebuilding: all existing tests still passed) — a reminder that "the fix works, I verified it manually/via live logs" is not the same as a committed regression test; always add the assertion, don't just observe the correct behavior once. The worst-case-occupancy backpressure test (`test_scheduler_backpressure.c`) only proves Req 3 AC2 (buffering/block/resume-with-no-loss) — it does not exercise task 3.2's `-E_XCE_NOT_READY` retry path, so don't cite it for Req 7 coverage (that's task 3.4's job). E2E kill-primary/reconnect and all-unreachable-at-startup were validated manually against a real 3-node `/xce-deploy-ha` cluster rather than committed as an automated test, following worker's task 2.3 precedent — this repo has no automated live-multi-process-deploy E2E harness for any C daemon (the only one, `test_e2e_live_ha_push.py`, is Python/apiserver-only); reviewer judged this acceptable both rounds. Manual validation confirmed join, primary-death survival, and lazy socket-orphaning-on-idle (matches design), but could not observe a live rejoin firing against real sockets — blocked by bug-010 (`xcecd` ASAN "already joined thread" abort under `unshare -r`, needed to bypass the worker daemon's root-only check in this sandbox; unconfirmed whether it's a genuine bug or a userns artifact, never tried under real root). The exact rejoin-after-publish-success code path already has solid white-box mock coverage from task 3.2's `test_rejoin_composition_retries_across_publish_calls`, so this residual gap is narrowly "proven against mocks, not yet against a live socket" — worth closing opportunistically if bug-010 ever gets resolved, not a blocker for this spec. diff --git a/src/scheduler/src/scheduler.c b/src/scheduler/src/scheduler.c index ff1cdd11..5bd11781 100644 --- a/src/scheduler/src/scheduler.c +++ b/src/scheduler/src/scheduler.c @@ -514,6 +514,53 @@ static int sched_on_reconnect(xcecli_ctx_t ctx, void* arg) return 0; } +/* + * Sized for the worst case: MAX_HA_CANDIDATES entries, each up to + * (XCE_MAXHOSTNAME_LEN - 1) hostname chars + ':' + a 5-digit port + a 2-char + * ", " separator (263 bytes/candidate at the max) -- the extra +16 pads that + * comfortably rather than sizing exactly to the boundary. Deliberately + * duplicated verbatim (same name, same value) from worker's + * controller_connector.c rather than shared via a common header: this + * task's boundary is SchedulerControllerClient (scheduler.c and its + * testsuite only) -- see that file's own identical macro for the + * canonical definition this one must never be allowed to silently drift + * from. */ +#define CANDIDATE_LIST_BUF_SIZE ((XCE_MAXHOSTNAME_LEN + 16) * MAX_HA_CANDIDATES) + +/* + * sched_format_candidate_list - renders sched->cluster_cfg's candidates as + * "host:port, host:port, ..." into buf (component-failover-adoption, task + * 3.3, Req 1 AC6 / Req 5 AC2). Ports worker's format_candidate_list() + * (controller_connector.c) to scheduler.c, adapted to take an explicit + * cluster_cfg parameter instead of reading a file-scope global -- scheduler + * keeps its candidate list on sched->cluster_cfg (a struct member, see + * xce_scheduler_t's field comment, scheduler.h), not a global, so no new + * global is introduced here. + * + * Truncates rather than overflows if buf is undersized (bufsz - 1 bound + * below): a too-small buf loses information, not memory safety. Callers + * must size buf as CANDIDATE_LIST_BUF_SIZE (or larger). + */ +static void sched_format_candidate_list(const xce_cluster_config_t* cluster_cfg, + char* buf, size_t bufsz) +{ + size_t used = 0; + int i; + + buf[0] = '\0'; + for (i = 0; (i < cluster_cfg->candidates_count) && (used < bufsz - 1); + i++) { + int written + = snprintf(buf + used, bufsz - used, "%s%s:%d", i > 0 ? ", " : "", + cluster_cfg->candidates[i].hostname, + cluster_cfg->candidates[i].port); + if (written < 0) { + break; + } + used += (size_t)written; + } +} + /* * Build the single validated candidate list this connect attempt will use * (component-failover-adoption, task 3.1, Req 1 AC1-3/AC5). @@ -688,8 +735,12 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config, rc = xcecli_start(sched->apictx); if (rc < 0) { - WARN("%s: all %d candidate(s) unreachable (rc=%d)", __func__, - sched->cluster_cfg.candidates_count, rc); + char attempted[CANDIDATE_LIST_BUF_SIZE]; + + sched_format_candidate_list(&(sched->cluster_cfg), attempted, + sizeof(attempted)); + WARN("%s: all %d candidate(s) unreachable (rc=%d): %s", __func__, + sched->cluster_cfg.candidates_count, rc, attempted); xcecli_ctx_fini(sched->apictx); return rc; } @@ -703,8 +754,12 @@ int sched_init(xce_scheduler_t* sched, xce_client_config_t* config, rc = xcecli_join(sched->apictx, HostRole_scheduler, NULL); if (rc < 0) { - ERROR("%s: xcecli_join() failed across %d candidate(s) (rc=%d)", - __func__, sched->cluster_cfg.candidates_count, rc); + char attempted[CANDIDATE_LIST_BUF_SIZE]; + + sched_format_candidate_list(&(sched->cluster_cfg), attempted, + sizeof(attempted)); + ERROR("%s: xcecli_join() failed across %d candidate(s) (rc=%d): %s", + __func__, sched->cluster_cfg.candidates_count, rc, attempted); xcecli_stop(sched->apictx); xcecli_ctx_fini(sched->apictx); return rc; diff --git a/src/scheduler/testsuite/meson.build b/src/scheduler/testsuite/meson.build index f7e7d0b3..a71e4d8a 100644 --- a/src/scheduler/testsuite/meson.build +++ b/src/scheduler/testsuite/meson.build @@ -829,6 +829,31 @@ sched_scheduler_decision_publish_retry_test = executable( '-Wl,--wrap=xcecli_set_reconnect_handler'], install: false) +# Worst-case decision/allocation backpressure test (component-failover- +# adoption, task 3.3, boundary: SchedulerControllerClient, Req 3 AC2 / Req +# 7). Same real-scheduler_srcs link technique as the two targets above, PLUS +# `-Wl,--wrap=` on xcecli_start/xcecli_join (forced to succeed with no I/O, +# so this file's one real sched_init() call actually starts the picker/ +# allocator threads) and xcecli_set_pend_reason (blocks on a test-controlled +# condvar, standing in for a stuck outage retry) -- see the test file's own +# header comment for the full design and worst-case derivation. +scheduler_backpressure_test_srcs = ['test_scheduler_backpressure.c'] +foreach f : scheduler_srcs + if f != 'src/main.c' + scheduler_backpressure_test_srcs += ['../' + f] + endif +endforeach + +sched_scheduler_backpressure_test = executable( + 'sched_scheduler_backpressure_test', + scheduler_backpressure_test_srcs, + include_directories:[incdir], + dependencies: [scheduler_deps, unity_dep], + link_args: ['-Wl,--wrap=xcecli_start', + '-Wl,--wrap=xcecli_join', + '-Wl,--wrap=xcecli_set_pend_reason'], + install: false) + # Register as meson tests so `meson test --suite scheduler` runs them all. # detect_leaks=0 matches the suite convention (fixtures are not torn down). test('scheduler test', scheduler_test, suite: 'scheduler', @@ -937,3 +962,6 @@ test('scheduler connect pipeline test', test('scheduler decision publish retry test', sched_scheduler_decision_publish_retry_test, suite: 'scheduler', env: ['ASAN_OPTIONS=detect_leaks=0']) +test('scheduler backpressure test', + sched_scheduler_backpressure_test, suite: 'scheduler', + env: ['ASAN_OPTIONS=detect_leaks=0']) diff --git a/src/scheduler/testsuite/test_scheduler_backpressure.c b/src/scheduler/testsuite/test_scheduler_backpressure.c new file mode 100644 index 00000000..3636ab55 --- /dev/null +++ b/src/scheduler/testsuite/test_scheduler_backpressure.c @@ -0,0 +1,550 @@ +/* + * test_scheduler_backpressure.c — worst-case decision/allocation + * backpressure test (component-failover-adoption, task 3.3, boundary: + * SchedulerControllerClient, Req 3 AC2). + * + * Scope note: this file proves Req 3 AC2 only -- the bounded, occupancy- + * dependent buffering upstream of the publish call (picker/allocator + * continue briefly, then genuinely block, then resume with nothing lost). + * It does NOT exercise Req 7's retry-until-success-or-shutdown wrapper + * around a `-E_XCE_NOT_READY` return from `xcecli_jobdispatch()`/ + * `xcecli_set_pend_reason()` (that wrapper, and its own test coverage, is + * task 3.2's `sched_publish_dispatch()`/`sched_publish_pend_reason()` work, + * see test_scheduler_decision_publish_retry.c). The wrapped + * `xcecli_set_pend_reason()` double below BLOCKS on a condvar rather than + * ever returning `-E_XCE_NOT_READY`, so it never enters that retry loop at + * all -- it stands in for the call still being in-flight/stuck mid-attempt, + * one layer below where Req 7 applies. + * + * Links the REAL production scheduler_srcs (same technique + * test_scheduler_connect.c / test_scheduler_decision_publish_retry.c + * already established for this boundary -- see meson.build) and calls the + * REAL, public sched_init(), which -- unlike task 3.2's rejoin test, which + * deliberately made its own setup join fail -- is made to SUCCEED here + * (xcecli_start()/xcecli_join() wrapped to return success with no I/O), + * so the real sched_job_picker() and sched_allocator() threads (scheduler.c) + * actually start. xcecli_set_pend_reason() is wrapped to BLOCK on a + * condition variable under test control, standing in for an ongoing + * controller outage (an XCECLI_FAILOVER_CALL retry budget stuck retrying) + * without any real network I/O or busy-spinning. + * + * ── Why the picker thread is never driven, and jobs are hand-built ────── + * + * sched_job_picker()/sched_allocator() are `static` (scheduler.c) and can + * only be started via the real sched_init() -- both threads start together, + * there is no way to run just one. sched_job_picker() is kept PERMANENTLY + * DORMANT here by never calling sched_go(): it blocks forever on + * cbuf_get(&core.wbuf) (scheduler.c), so it never touches core.jbuf itself + * and never contends with this file's own direct cbuf_put()/cbuf_tryput() + * calls on core.jbuf, which stand in for the picker's role of feeding one + * jobid at a time -- exactly the "without also running the full + * sched_job_picker thread" narrowing this task's own text pre-authorizes. + * + * sched_allocator() IS the real, unmodified production function under + * test here (not a stand-in) -- proving its genuine cbuf_put()-blocks + * behavior against a full `decisions` cbuf is this test's actual point. + * Driving it through sched_make_allocation() with a fully realistic + * job/queue/worker/resource model (parse_resreq(), the seqjob_allocator + * plugin's "slots" resource lookup, sched_collect_workers() against a real + * resmgr) would need substantial scaffolding unrelated to what this test + * verifies. Instead, each test job's xce_sched_obj_t is built directly + * with tag=XCE_RESREQ_COMPOUND and a COMPOUND resreq with an EMPTY + * `primaries` list: sched_make_compound_allocation()'s for-loop + * (scheduler.c) then has zero iterations, so it returns rc=0 ("pend") + * without ever touching a worker, a resource, or the seqjob_allocator + * plugin -- alloc_rc==0 is exactly the same "job is pend" branch + * sched_allocator() takes for a real no-matching-worker pend outcome, so + * this fixture exercises the identical decision-construction and + * cbuf_put(&core.decisions, ...) code path a real pend decision would. + * The pend reason itself is pre-set directly on the job's top-level + * `schedobj->reason` field before sched_allocator() ever sees it; + * sched_make_compound_allocation() only overwrites that field when its own + * ctx.reason is non-NULL, which never happens with zero primaries, so the + * pre-set reason survives untouched into make_pend_decision(). + * + * ── Worst-case derivation (do not copy a number from design.md prose + * without re-deriving it against live cbuf state -- this task's own + * instruction) ────────────────────────────────────────────────────── + * + * Both cbuf capacities are read LIVE from the real, freshly-initialized + * core (sched.core.decisions.ncap / sched.core.jbuf.ncap -- xce_sched_core_t + * struct fields, mylock.h's cbuf_t) rather than hardcoded, and asserted + * against the documented values (ndecisions=5, jbuf capacity=1, core.c) as + * a sanity check only. + * + * The worst case (least slack) is: `decisions` is ALREADY FULL at the + * instant the stuck decision is popped off it (design.md's "decisions cbuf + * already at or near its capacity-5 occupancy at the moment the stall + * begins"). This test constructs exactly that precondition (prefills + * `decisions` to full capacity BEFORE starting the stall), then proves, + * against the REAL sched_allocator() thread and REAL cbuf blocking + * semantics (mylock.c, not mocked): + * + * 1. `decisions` has room for exactly ONE more decision once the stuck + * one is popped (occupancy ncap-1) -- job A's PEND decision lands. + * 2. The NEXT decision (job B's) is fully built (not dropped, not + * errored -- Req 3 AC2's "no job picked or decided ... is ever lost") + * but then genuinely BLOCKS trying to enqueue, because `decisions` is + * full again. + * 3. `jbuf` (capacity 1) absorbs exactly ONE more job (C) beyond that -- + * the allocator had already drained job B out of jbuf before getting + * stuck on `decisions` -- and then a further job (D) is refused for a + * sustained bounded window, proving the picker itself would now + * block too (Req 3 AC2's "and then sched_job_picker genuinely + * block"), not silently drop work. + * 4. Once the stall clears, every decision produced before AND during + * the outage (the prefilled backlog plus jobs A/B/C) is eventually + * drained and published -- none lost (Req 3 AC2's "no job picked or + * decided during that bounded window is ever lost"). + * + * This matches requirements.md's "as few as 1-2" floor (job A enqueued + + * job B built-but-blocked = 2 additional decisions) at the `decisions` + * layer, plus the analogous one-slot-of-slack-then-genuine-block pattern + * one layer down at `jbuf` -- derived here from the live cbuf mechanics, + * not copied from prose. + * + * All "prove it's genuinely blocked" checks use cbuf_tryput() (non-blocking) + * in a short bounded-retry loop, never a real blocking call -- this test + * can never hang even if the backpressure guarantee were broken. + */ +#include +#include +#include +#include +#include +#include + +#include + +#include "client.h" +#include "controller.capnp.h" +#include "core.h" +#include "decision.h" +#include "mylock.h" +#include "preason.h" +#include "resreq.h" +#include "sched/types.h" +#include "scheduler.h" +#include "xce/cluster.h" +#include "xlist.h" + +extern int scheduler_req_handler(xce_request_t*, void*); + +#define N_DUMMY_PREFILL 5 +#define N_REAL_JOBS_FED 3 /* A, B, C */ +#define N_TOTAL_DECISIONS (N_DUMMY_PREFILL + N_REAL_JOBS_FED) +#define POLL_STEP_NS (5 * 1000 * 1000) /* 5ms */ +#define POLL_TIMEOUT_MS 2000 + +/* ════════════════════════════════════════════════════════════════════════ + * Test double: __wrap_xcecli_set_pend_reason -- blocks on a condvar until + * the test releases it, standing in for a stuck outage retry (see file + * header comment). __wrap_xcecli_start / __wrap_xcecli_join are trivial + * always-succeed doubles (same rationale as + * test_scheduler_decision_publish_retry.c's own __wrap_xcecli_start) so + * this file's one real sched_init() call actually starts the picker/ + * allocator threads instead of returning early. + * ════════════════════════════════════════════════════════════════════════ */ + +static pthread_mutex_t g_pend_mtx = PTHREAD_MUTEX_INITIALIZER; +static pthread_cond_t g_pend_release_cond = PTHREAD_COND_INITIALIZER; +static pthread_cond_t g_pend_entered_cond = PTHREAD_COND_INITIALIZER; +static int g_pend_release = 0; +static int g_pend_entered = 0; + +static void reset_pend_wrap(void) +{ + g_pend_release = 0; + g_pend_entered = 0; +} + +int __wrap_xcecli_start(xcecli_ctx_t c) +{ + (void)c; + return 0; +} + +int __wrap_xcecli_join(xcecli_ctx_t c, int role, hostattrs_t* attrs) +{ + (void)c; + (void)role; + (void)attrs; + return 0; +} + +int __wrap_xcecli_set_pend_reason(xcecli_ctx_t c, xce_preason_set_t* reasons) +{ + (void)c; + (void)reasons; + + pthread_mutex_lock(&g_pend_mtx); + g_pend_entered = 1; + pthread_cond_broadcast(&g_pend_entered_cond); + while (!g_pend_release) { + pthread_cond_wait(&g_pend_release_cond, &g_pend_mtx); + } + pthread_mutex_unlock(&g_pend_mtx); + + return 0; +} + +static void wait_for_pend_wrap_entered(void) +{ + pthread_mutex_lock(&g_pend_mtx); + while (!g_pend_entered) { + pthread_cond_wait(&g_pend_entered_cond, &g_pend_mtx); + } + pthread_mutex_unlock(&g_pend_mtx); +} + +static void release_pend_wrap(void) +{ + pthread_mutex_lock(&g_pend_mtx); + g_pend_release = 1; + pthread_cond_broadcast(&g_pend_release_cond); + pthread_mutex_unlock(&g_pend_mtx); +} + +void setUp(void) { reset_pend_wrap(); } +void tearDown(void) { return; } + +/* ════════════════════════════════════════════════════════════════════════ + * Fixtures + * ════════════════════════════════════════════════════════════════════════ */ + +/* build_pend_only_job - see file header comment for why COMPOUND + zero + * primaries is used: it drives sched_allocator()'s real "job is pend" + * decision-construction path without needing any worker/resource/queue + * scaffolding. */ +static xce_sched_job_t* build_pend_only_job(uint64_t jobid) +{ + xce_sched_job_t* job = (xce_sched_job_t*)calloc(1, sizeof(*job)); + xce_sched_obj_t* schedobj = (xce_sched_obj_t*)calloc(1, sizeof(*schedobj)); + xce_compound_schedobj_t* compound + = (xce_compound_schedobj_t*)calloc(1, sizeof(*compound)); + xce_compound_resreq_t* resreq + = (xce_compound_resreq_t*)calloc(1, sizeof(*resreq)); + + xlist_init(&(resreq->primaries), 1); + + compound->owner = schedobj; + compound->resreq = resreq; + compound->reason = NULL; + + schedobj->tag = XCE_RESREQ_COMPOUND; + schedobj->u.compound = compound; + /* Top-level field, distinct from compound->reason above -- see file + * header comment for why this survives sched_make_compound_allocation() + * untouched. */ + schedobj->reason = make_pendreason(jobid); + + job->jobid = jobid; + job->state = JOB_STAT_PEND; + job->schedobj = schedobj; + + return job; +} + +static xce_decision_t* build_dummy_pend_decision(uint64_t jobid) +{ + return make_pend_decision(make_pendreason(jobid)); +} + +/* ════════════════════════════════════════════════════════════════════════ + * Drain thread -- stands in for driver.c's decision-drain consumer + * (xce_sched_driver_run_normal_mode_round()'s cbuf_get(&core.decisions) + * loop). Pops exactly N_TOTAL_DECISIONS decisions, publishing each via the + * REAL sched_publish_pend_reason() (scheduler.c). Its first iteration is + * the one that blocks in the wrapped call (see file header comment). + * ════════════════════════════════════════════════════════════════════════ */ + +typedef struct { + xce_scheduler_t* sched; + uint64_t seen_jobids[N_TOTAL_DECISIONS]; + int seen_rcs[N_TOTAL_DECISIONS]; + int n_seen; +} drain_ctx_t; + +static void* drain_thread_fn(void* arg) +{ + drain_ctx_t* dctx = (drain_ctx_t*)arg; + int i; + + for (i = 0; i < N_TOTAL_DECISIONS; i++) { + xce_decision_t* decision = NULL; + xce_pendreason_t* reasons[1]; + xce_preason_set_t reason_set; + int rc; + + cbuf_get(&(dctx->sched->core.decisions), &decision); + + reasons[0] = decision->u.reason; + reason_set.nreasons = 1; + reason_set.reasons = reasons; + + rc = sched_publish_pend_reason(dctx->sched, &reason_set); + + dctx->seen_jobids[dctx->n_seen] = decision->u.reason->jobid; + dctx->seen_rcs[dctx->n_seen] = rc; + dctx->n_seen++; + } + + return NULL; +} + +/* ════════════════════════════════════════════════════════════════════════ + * The test + * ════════════════════════════════════════════════════════════════════════ */ + +static void test_worst_case_backpressure_blocks_then_recovers_no_loss(void) +{ + xce_client_config_t config; + xce_cluster_config_t cluster_cfg; + xce_scheduler_t sched; + pthread_t drain_tid; + drain_ctx_t dctx; + int rc; + int i; + int ncap_decisions; + int ncap_jbuf; + + memset(&config, 0, sizeof(config)); + config.nmasters = 1; + config.masters[0] = "127.0.0.1"; + config.master_ports[0] = 19997; + config.from = "sched_backpressure_test"; + config.key = "keepsecret$"; + + memset(&cluster_cfg, 0, sizeof(cluster_cfg)); + cluster_cfg.candidates_count = 1; + strncpy(cluster_cfg.candidates[0].hostname, "127.0.0.1", + XCE_MAXHOSTNAME_LEN - 1); + cluster_cfg.candidates[0].port = 19997; + strncpy(cluster_cfg.identity, "sched_backpressure_test", + sizeof(cluster_cfg.identity) - 1); + + memset(&sched, 0, sizeof(sched)); + rc = sched_init(&sched, &config, &cluster_cfg, scheduler_req_handler); + TEST_ASSERT_TRUE_MESSAGE( + rc >= 0, "fixture: sched_init() must succeed (xcecli_start()/" + "xcecli_join() wrapped to succeed with no I/O) so the " + "real picker/allocator threads actually start"); + + ncap_decisions = sched.core.decisions.ncap; + ncap_jbuf = sched.core.jbuf.ncap; + TEST_ASSERT_TRUE_MESSAGE(ncap_decisions >= 1, + "sanity: decisions cbuf must have positive " + "capacity"); + TEST_ASSERT_TRUE_MESSAGE(ncap_jbuf >= 1, + "sanity: jbuf cbuf must have positive " + "capacity"); + TEST_ASSERT_EQUAL_INT_MESSAGE( + N_DUMMY_PREFILL, ncap_decisions, + "documented invariant (core.c: ndecisions=5) -- if this fails, " + "N_DUMMY_PREFILL/N_TOTAL_DECISIONS above must be re-derived " + "against the new live capacity, not just this assertion changed"); + TEST_ASSERT_EQUAL_INT_MESSAGE( + 1, ncap_jbuf, + "documented invariant (core.c: jbuf capacity fixed at 1, a " + "scheduler-core determinism invariant this task does not " + "adjust)"); + + /* --- Step 1: prefill `decisions` to FULL capacity -- the worst-case + * (least-slack) precondition. --- */ + for (i = 0; i < ncap_decisions; i++) { + xce_decision_t* d = build_dummy_pend_decision(9000 + (uint64_t)i); + + TEST_ASSERT_NOT_NULL(d); + rc = cbuf_tryput(&(sched.core.decisions), &d); + TEST_ASSERT_TRUE_MESSAGE( + rc >= 0, "fixture: prefilling decisions to capacity must not " + "itself block"); + } + + /* --- Step 2: start the drain thread. Its first cbuf_get() pops the + * first prefilled decision and calls the REAL + * sched_publish_pend_reason(), which blocks inside the wrapped + * xcecli_set_pend_reason() -- simulating an ongoing outage. --- */ + dctx.sched = &sched; + dctx.n_seen = 0; + rc = pthread_create(&drain_tid, NULL, drain_thread_fn, &dctx); + TEST_ASSERT_EQUAL_INT(0, rc); + + wait_for_pend_wrap_entered(); + + TEST_ASSERT_EQUAL_INT_MESSAGE( + ncap_decisions - 1, sched.core.decisions.len, + "the drain thread's single cbuf_get() must have removed exactly " + "one decision, leaving decisions at capacity-1"); + + /* --- Step 3: feed job A through the REAL sched_allocator() thread + * (via jbuf, standing in for the picker). With one free slot in + * `decisions`, job A's PEND decision must land successfully -- this is + * the "as few as 1" floor. --- */ + { + xce_sched_job_t* jobA = build_pend_only_job(9101); + uint64_t jobidA = jobA->jobid; + + TEST_ASSERT_TRUE(sched_add_job(&(sched.core), jobA) >= 0); + rc = cbuf_tryput(&(sched.core.jbuf), &jobidA); + TEST_ASSERT_TRUE_MESSAGE( + rc >= 0, "jbuf must have room for job A -- it starts empty " + "since the dormant picker thread never touches it " + "(sched_go() is never called in this test)"); + } + + { + int waited_ms = 0; + + while ((sched.core.decisions.len < ncap_decisions) + && (waited_ms < POLL_TIMEOUT_MS)) { + struct timespec ts = { 0, POLL_STEP_NS }; + + nanosleep(&ts, NULL); + waited_ms += 5; + } + TEST_ASSERT_EQUAL_INT_MESSAGE( + ncap_decisions, sched.core.decisions.len, + "job A's decision must be the ONE additional decision the " + "worst-case (already-full) occupancy has room for"); + } + + /* --- Step 4: feed job B. `decisions` is full again, so + * sched_allocator() must build job B's decision (Req 3 AC2: deciding is + * not lost) but then genuinely BLOCK trying to enqueue it -- the + * second half of the "1-2" floor. --- */ + { + xce_sched_job_t* jobB = build_pend_only_job(9102); + uint64_t jobidB = jobB->jobid; + + TEST_ASSERT_TRUE(sched_add_job(&(sched.core), jobB) >= 0); + rc = cbuf_tryput(&(sched.core.jbuf), &jobidB); + TEST_ASSERT_TRUE_MESSAGE( + rc >= 0, "jbuf must have room for job B -- the allocator " + "already drained job A's jobid out of it"); + } + + { + /* Bounded window for the allocator to reach its blocking + * cbuf_put() -- decisions occupancy must not exceed capacity once + * it does. */ + struct timespec ts = { 0, 100 * 1000 * 1000 }; + + nanosleep(&ts, NULL); + } + TEST_ASSERT_EQUAL_INT_MESSAGE( + ncap_decisions, sched.core.decisions.len, + "job B's decision must be stuck trying to enqueue -- decisions " + "occupancy must never exceed its own capacity"); + + /* --- Step 5: prove the genuine block reaches jbuf too (Req 3 AC2: + * "...and then sched_job_picker genuinely block"). jbuf absorbs + * exactly one more job (C) -- the allocator already drained jbuf + * before getting stuck on `decisions` -- and a further push (D) must + * then fail for a sustained bounded window. --- */ + { + xce_sched_job_t* jobC = build_pend_only_job(9103); + uint64_t jobidC = jobC->jobid; + + TEST_ASSERT_TRUE(sched_add_job(&(sched.core), jobC) >= 0); + rc = cbuf_tryput(&(sched.core.jbuf), &jobidC); + TEST_ASSERT_TRUE_MESSAGE( + rc >= 0, "jbuf's one slot of slack must still accept job C"); + } + { + uint64_t jobidD = 9104; + int any_succeeded = 0; + int probe; + + for (probe = 0; probe < 10; probe++) { + struct timespec ts = { 0, 20 * 1000 * 1000 }; + + rc = cbuf_tryput(&(sched.core.jbuf), &jobidD); + if (rc >= 0) { + any_succeeded = 1; + break; + } + nanosleep(&ts, NULL); + } + TEST_ASSERT_FALSE_MESSAGE( + any_succeeded, + "genuine block, Req 3 AC2: with sched_allocator stuck on a " + "full decisions cbuf, jbuf's single slot is already occupied " + "by job C -- a further job (D) must never be accepted over a " + "sustained bounded window, proving the picker itself would " + "now block too, not silently drop work"); + } + + /* --- Step 6: the outage clears. Req 3 AC2: the scheduler must resume + * automatically and publish every decision produced during the bounded + * window, with none lost. --- */ + release_pend_wrap(); + + rc = pthread_join(drain_tid, NULL); + TEST_ASSERT_EQUAL_INT(0, rc); + + TEST_ASSERT_EQUAL_INT_MESSAGE( + N_TOTAL_DECISIONS, dctx.n_seen, + "every decision produced before AND during the outage -- the " + "prefilled backlog plus jobs A/B/C -- must eventually be drained " + "and published, none silently lost (Req 3 AC2)"); + + for (i = 0; i < dctx.n_seen; i++) { + TEST_ASSERT_TRUE_MESSAGE( + dctx.seen_rcs[i] >= 0, + "every drained decision's publish call must eventually " + "report success"); + } + + { + int saw_a = 0, saw_b = 0, saw_c = 0; + + for (i = 0; i < dctx.n_seen; i++) { + if (dctx.seen_jobids[i] == 9101) { + saw_a = 1; + } + if (dctx.seen_jobids[i] == 9102) { + saw_b = 1; + } + if (dctx.seen_jobids[i] == 9103) { + saw_c = 1; + } + } + TEST_ASSERT_TRUE_MESSAGE( + saw_a && saw_b && saw_c, + "jobs A, B, and C -- decided/picked during the bounded outage " + "-- must all eventually be published, not just the " + "pre-outage backlog"); + } + + printf("test_worst_case_backpressure_blocks_then_recovers_no_loss: " + "PASS (decisions.ncap=%d jbuf.ncap=%d, worst-case additional " + "slack before genuine block = 1 decision fully enqueued + 1 " + "decision built-but-blocked + 1 job buffered in jbuf)\n", + ncap_decisions, ncap_jbuf); + + /* --- Teardown: cancel + join the real threads before tearing down + * core state, matching sched_fini()'s own ordering (scheduler.c) minus + * the apictx cleanup calls (xcecli_wakeup/xcecli_stop/xcecli_ctx_fini) + * this fixture's wrapped xcecli_start() never fully set up (it does no + * real connect, so ctx->failover.lock was never mylock_init'd) -- + * deliberately skipped rather than also wrapped, consistent with this + * test suite's existing detect_leaks=0 convention (meson.build). --- */ + pthread_cancel(sched.core.jobpicker); + pthread_cancel(sched.core.allocator); + { + uint64_t wake = 0; + + cbuf_tryput(&(sched.core.jbuf), &wake); + } + pthread_join(sched.core.jobpicker, NULL); + pthread_join(sched.core.allocator, NULL); + xce_sched_core_fini(&(sched.core)); +} + +int main(void) +{ + UNITY_BEGIN(); + RUN_TEST(test_worst_case_backpressure_blocks_then_recovers_no_loss); + return UNITY_END(); +} diff --git a/src/scheduler/testsuite/test_scheduler_connect.c b/src/scheduler/testsuite/test_scheduler_connect.c index e566824d..6fea1faf 100644 --- a/src/scheduler/testsuite/test_scheduler_connect.c +++ b/src/scheduler/testsuite/test_scheduler_connect.c @@ -24,10 +24,20 @@ * closed local ports (connect() gets ECONNREFUSED near-instantly, same * "no backoff loop inside xcecli_start()'s per-master connect scan" * property worker's task 2.3 test relies on) and inspecting the - * resulting sched->cluster_cfg -- every connect log line sched_init() - * emits is derived from that same struct's contents, so structural - * equality here is exactly what "identical connect log sequence" (this - * task's own observable-completion wording) reduces to. + * resulting sched->cluster_cfg. + * - Req 1 AC6 / Req 5 AC2's "identifies which candidates were attempted" + * (component-failover-adoption, task 3.3) is proven by genuine RUNTIME + * LOG CAPTURE (dup2() redirect of STDOUT_FILENO for the duration of the + * sched_init() call under test -- same mechanism + * test_scheduler_decision_publish_retry.c's own capture_stdout_during() + * established for this boundary, see + * capture_stdout_during_sched_init()'s own header comment below), not + * inferred from sched->cluster_cfg's contents: proving what sched_init() + * actually PRINTS is a materially stronger check than proving the struct + * it derives the message from, since sched_format_candidate_list() + * (scheduler.c) -- the function that turns that struct into the printed + * "host:port, host:port, ..." text -- is itself `static` and otherwise + * has no direct call site in this file to exercise. * - The rejoin callback's "no I/O, sets a flag, returns immediately" * contract (Req 2.1, Req 5.1-5.2, design.md's "Rejoin callback * constraint") and the connect pipeline's single-path structure (Req @@ -55,6 +65,7 @@ #include #include #include +#include #include @@ -204,6 +215,120 @@ static void test_sched_init_all_candidates_unreachable_fails_fast(void) rc, elapsed_ms(&t0, &t1)); } +/* + * Captures STDOUT_FILENO for the duration of a sched_init() call, returning + * the captured text as a malloc'd, NUL-terminated buffer the caller must + * free(). Same dup()/dup2() mechanism as + * test_scheduler_decision_publish_retry.c's own capture_stdout_during() + * (component-failover-adoption task 3.2) and, before that, + * test_apply_correctness_verification.c's + * test_apply_divergent_loud_fence_log_omits_false_peer_hwm_clause: this test + * binary never calls mylog_init()/tlog_init() (only the real daemon main()s + * do), so tlog's internal root log handle stays NULL and every WARN()/ + * ERROR() call falls through tlog's early-print path, which writes the + * fully formatted line directly to STDOUT_FILENO via a raw write() + * (vendored tinylog, tlog.c) -- genuine runtime log capture, not a + * source-text scan, so this proves what sched_init() actually PRINTS, not + * merely what its source contains. */ +static char* capture_stdout_during_sched_init(xce_scheduler_t* sched, + xce_client_config_t* config, + xce_cluster_config_t* cluster_cfg, + int* rc_out) +{ + char tmp[] = "/tmp/xce_sched_connect_logcap_XXXXXX"; + int fd = mkstemp(tmp); + TEST_ASSERT_TRUE_MESSAGE( + fd >= 0, "fixture: mkstemp for stdout capture must succeed"); + + fflush(stdout); + int saved_stdout = dup(STDOUT_FILENO); + TEST_ASSERT_TRUE_MESSAGE(saved_stdout >= 0, + "fixture: dup(STDOUT_FILENO) must succeed"); + TEST_ASSERT_TRUE_MESSAGE(dup2(fd, STDOUT_FILENO) >= 0, + "fixture: dup2 redirect of STDOUT_FILENO must " + "succeed"); + + *rc_out = sched_init(sched, config, cluster_cfg, scheduler_req_handler); + + fflush(stdout); + dup2(saved_stdout, STDOUT_FILENO); + close(saved_stdout); + + off_t len = lseek(fd, 0, SEEK_CUR); + TEST_ASSERT_TRUE_MESSAGE(len >= 0, "fixture: lseek on capture file must " + "succeed"); + + char* buf = (char*)calloc(1, (size_t)len + 1); + TEST_ASSERT_NOT_NULL(buf); + if (len > 0) { + TEST_ASSERT_TRUE_MESSAGE(lseek(fd, 0, SEEK_SET) == 0, + "fixture: rewind of capture file must " + "succeed"); + ssize_t nread = read(fd, buf, (size_t)len); + TEST_ASSERT_TRUE_MESSAGE(nread == len, + "fixture: reading back captured buffer " + "must succeed"); + } + close(fd); + unlink(tmp); + + return buf; +} + +/* + * test_all_candidates_unreachable_log_names_every_candidate_host_port + * (component-failover-adoption, task 3.3, Req 1 AC6 / Req 5 AC2) -- + * behavioral proof that sched_init()'s all-candidates-unreachable failure + * path actually PRINTS every attempted candidate's host:port, not merely + * that rc<0 (test_sched_init_all_candidates_unreachable_fails_fast above + * already covers the rc<0/fails-fast half; this is the missing "identifying + * which candidates were attempted" half). sched_format_candidate_list() + * itself (scheduler.c) is `static` and this file object-links scheduler_srcs + * rather than #include-ing scheduler.c (see file header comment), so it + * cannot be called directly -- captures the real WARN() output sched_init() + * emits instead, same genre as this file's own header comment already + * establishes for the rejoin-callback/single-connect-sequence structural + * checks, but behavioral rather than source-text here since the actual + * runtime output is exactly what "produces a failure message naming each + * attempted candidate" (this task's own Observable completion wording) + * means. + */ +static void +test_all_candidates_unreachable_log_names_every_candidate_host_port(void) +{ + xce_scheduler_t sched; + xce_client_config_t config; + xce_cluster_config_t cluster_cfg; + int rc; + char* captured; + + init_legacy_config(&config, 1 /* unused: cluster_cfg wins */); + init_cluster_cfg_n(&cluster_cfg, TEST_PORT_BASE + 40, 3); + + captured + = capture_stdout_during_sched_init(&sched, &config, &cluster_cfg, &rc); + + TEST_ASSERT_TRUE_MESSAGE( + rc < 0, "sched_init() against 3 unreachable candidates must fail"); + + /* Every configured candidate's "host:port" must appear in the captured + * failure log, in the exact format sched_format_candidate_list() + * produces -- not just a candidate count. */ + for (int i = 0; i < 3; i++) { + char expected[64]; + snprintf(expected, sizeof(expected), "127.0.0.1:%d", + TEST_PORT_BASE + 40 + i); + TEST_ASSERT_NOT_NULL_MESSAGE( + strstr(captured, expected), + "all-candidates-unreachable log must name every attempted " + "candidate's host:port -- one is missing"); + } + + free(captured); + printf("test_all_candidates_unreachable_log_names_every_candidate_host_" + "port: PASS\n"); +} + static void test_sched_init_no_peers_default_also_fails_cleanly(void) { xce_scheduler_t sched; @@ -404,14 +529,57 @@ static void test_sched_init_has_single_connect_sequence_not_a_loop(void) printf("test_sched_init_has_single_connect_sequence_not_a_loop: PASS\n"); } +/* + * test_no_periodic_heartbeat_call_in_source (component-failover-adoption, + * task 3.3, Req 3 AC3) - a source-text-scan structural regression guard, + * same genre as this file's other structural tests above. + * + * Confirmed by source read (grep -rn "xcecli_hello\|heartbeat\|liveness" + * src/scheduler/src/*.c returns zero matches): the scheduler makes no + * periodic liveness/heartbeat call of its own to a connected controller + * today, unlike worker's checking_connection()/xcecli_hello() + * (controller_connector.c) or apiserver. This means Req 3 AC3's "reconnect + * retries shall not delay or prevent the component's own heartbeat + * responses" is satisfied BY CONSTRUCTION for the scheduler -- there is no + * heartbeat call for a stuck retry to ever delay -- not by any runtime + * behavior this task adds. A behavioral test would have nothing to + * exercise; this cheap net instead fails loudly if a future change + * introduces a periodic xcecli_hello-style call into scheduler.c, which + * would silently reopen exactly the liveness hazard Req 3 AC3 rules out. + */ +static void test_no_periodic_heartbeat_call_in_source(void) +{ + char* src = read_whole_file(XCE_SCHEDULER_C_PATH); + TEST_ASSERT_NOT_NULL_MESSAGE(src, + "could not read scheduler.c source for scan"); + + TEST_ASSERT_NULL_MESSAGE( + strstr(src, "xcecli_hello"), + "Req 3 AC3: the scheduler must make no periodic liveness/heartbeat " + "call of its own to a connected controller -- introducing one " + "(e.g. a checking_connection()-style poll loop calling " + "xcecli_hello(), mirroring worker's controller_connector.c) would " + "give a stalled reconnect retry a heartbeat call it could delay, " + "reopening the exact liveness hazard this requirement rules out."); + TEST_ASSERT_NULL_MESSAGE( + strstr(src, "heartbeat"), + "no heartbeat-labeled logic of any kind is expected in scheduler.c " + "-- see this test's header comment"); + + free(src); + printf("test_no_periodic_heartbeat_call_in_source: PASS\n"); +} + int main(void) { UNITY_BEGIN(); RUN_TEST(test_no_peers_and_single_entry_peers_produce_identical_cluster_cfg_shape); RUN_TEST(test_multi_candidate_cluster_cfg_passthrough); RUN_TEST(test_sched_init_all_candidates_unreachable_fails_fast); + RUN_TEST(test_all_candidates_unreachable_log_names_every_candidate_host_port); RUN_TEST(test_sched_init_no_peers_default_also_fails_cleanly); RUN_TEST(test_reconnect_handler_registered_after_start_not_before); + RUN_TEST(test_no_periodic_heartbeat_call_in_source); RUN_TEST(test_rejoin_callback_body_has_no_io_or_xcecli_calls); RUN_TEST(test_sched_init_has_single_connect_sequence_not_a_loop); return UNITY_END();