lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(component-failover-adoption): add scheduler decision-publish retry-until-success-or-shutdown

Browse source 
Wraps sched_publish_dispatch()/sched_publish_pend_reason() in a shared
retry loop that keeps retrying a -E_XCE_NOT_READY-exhausted publish
until it succeeds or the scheduler is shutting down (core.stop checked
before each attempt, not only after), composed with the pending-rejoin
follow-up established in task 3.1. Extended-retry success logs
distinctly from an ordinary first-attempt success (Req 7 AC4).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NNB4zGPCsHyDUCDJyMbiQN
This commit is contained in:
Rongsong Shen 2026-08-11 12:03:21 +00:00
parent 54d53df956
commit 7c260e0c97
Signed by: rshen
SSH key fingerprint: SHA256:8Vf2pZ3h6rJa7EhGK+WaAQ9Mta/apZvteW6g9U/WKKk
4 changed files with 747 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

131
src/scheduler/src/scheduler.c
View file

@ -1338,15 +1338,142 @@ int sched_commit_decision(xce_scheduler_t* sched, xce_decision_t* decision)
return 0;
}
/*
* sched_publish_check_rejoin - after a decision publish call finally
* succeeds, re-issue xcecli_join() as a separate follow-up if the client
* library's reconnect callback (sched_on_reconnect(), above) flagged a
* pending rejoin (component-failover-adoption, task 3.2, Req 2 AC5).
*
* Deliberately NOT its own retry loop: on failure, sched_needs_rejoin
* stays set, so the NEXT decision's publish call retries this same
* rejoin -- the per-decision retry cadence sched_publish_retry_loop()
* below already establishes IS the retry mechanism (design.md's
* "Rejoin-check interaction" note).
*
* Uses the same rc<0-is-the-only-failure-test convention sched_init()'s
* own xcecli_join() check uses above (component-failover-adoption task
* 2.3's worker lesson: a successful join returns the assigned slot index
* as reply.status, which is not always exactly 0).
*/
static void sched_publish_check_rejoin(xce_scheduler_t* sched)
{
int rc;
if (!sched_needs_rejoin) {
return;
}
rc = xcecli_join(sched->apictx, HostRole_scheduler, NULL);
if (rc < 0) {
WARN("%s: pending rejoin retry failed (rc=%d) -- will retry on the "
"next decision publish",
__func__, rc);
return;
}
sched_needs_rejoin = false;
INFO("%s: pending rejoin succeeded, cluster membership restored", __func__);
}
/*
* sched_publish_attempt_fn_t / sched_publish_retry_loop - shared
* retry-until-success-or-shutdown wrapper for the scheduler's two
* decision-publish call sites (component-failover-adoption, task 3.2, Req
* 7 AC1-AC4).
*
* xcecli_jobdispatch()/xcecli_set_pend_reason() are each already
* XCECLI_FAILOVER_CALL-wrapped, retrying a SINGLE call up to a bounded
* budget (max_retries=20, backoff capped at max_retry_ms=5000ms --
* roughly 1-2 minutes total, client_failover.c) before returning
* -E_XCE_NOT_READY. Left unhandled, that return was silently swallowed by
* the driver's decision-drain loop (xce_sched_driver_on_decision(),
* driver.c, never inspects its own rc) -- since sched_commit_decision()
* already mutates scheduler-local job state BEFORE the publish attempt
* (sched_allocator()), a budget-exhausted publish permanently diverged
* scheduler-local state from what the controller was told. This loop
* re-invokes the attempt on -E_XCE_NOT_READY -- each iteration
* re-invokes the whole XCECLI_FAILOVER_CALL-wrapped function, starting a
* FRESH internal retry budget rather than building a second retry
* mechanism around it -- until either a non--E_XCE_NOT_READY return or
* shutdown.
*
* core.stop is checked BEFORE each attempt, not only after: a decision
* already queued behind the one actively retrying must see core.stop
* already true and return immediately, rather than each burning one more
* full ~100s budget cycle just to discover shutdown was requested
* (design.md's third-pass finding -- this is why this is a `for(;;)` loop
* with a leading stop check, not a `do { ... } while (!stop)`).
*/
typedef int (*sched_publish_attempt_fn_t)(xce_scheduler_t* sched,
void* payload);
static int sched_publish_retry_loop(xce_scheduler_t* sched, void* payload,
sched_publish_attempt_fn_t attempt_fn,
const char* what)
{
int rc;
int attempts = 0;
for (;;) {
if (sched->core.stop) {
WARN("%s: scheduler shutting down, abandoning %s publish "
"retry after %d attempt(s)",
__func__, what, attempts);
rc = -E_XCE_NOT_READY;
break;
}
rc = attempt_fn(sched, payload);
attempts++;
if (rc != -E_XCE_NOT_READY) {
break;
}
WARN("%s: %s publish retry-budget exhausted (attempt=%d), "
"retrying",
__func__, what, attempts);
}
/* Req 7 AC4: an extended-retry success (more than one attempt) must
* log distinctly from an ordinary first-attempt success -- there is
* no separate "ordinary success" log line to collide with today, so
* this extra INFO() line alone satisfies "distinctly" by
* construction. */
if ((rc >= 0) && (attempts > 1)) {
INFO("%s: %s published after extended retry (%d attempts)", __func__,
what, attempts);
}
if (rc >= 0) {
sched_publish_check_rejoin(sched);
}
return rc;
}
static int sched_publish_pend_reason_attempt(xce_scheduler_t* sched,
void* payload)
{
return xcecli_set_pend_reason(sched->apictx, (xce_preason_set_t*)payload);
}
static int sched_publish_dispatch_attempt(xce_scheduler_t* sched, void* payload)
{
return xcecli_jobdispatch(sched->apictx, (xce_job_dispatch_t*)payload);
}
int sched_publish_pend_reason(xce_scheduler_t* sched,
xce_preason_set_t* reasons)
{
return xcecli_set_pend_reason(sched->apictx, reasons);
return sched_publish_retry_loop(
sched, reasons, sched_publish_pend_reason_attempt, "pend-reason");
}
int sched_publish_dispatch(xce_scheduler_t* sched, xce_job_dispatch_t* dispatch)
{
return xcecli_jobdispatch(sched->apictx, dispatch);
return sched_publish_retry_loop(sched, dispatch,
sched_publish_dispatch_attempt, "dispatch");
}
typedef struct {

41
src/scheduler/testsuite/meson.build
View file

@ -791,6 +791,44 @@ sched_scheduler_connect_test = executable(
+ '"'],
install: false)
# Tests for the scheduler's decision-publish retry-until-success-or-shutdown
# wrapper, composed with rejoin retry-on-failure (component-failover-adoption,
# task 3.2, boundary: SchedulerControllerClient). scheduler.c IS the file
# under test here (sched_publish_dispatch()/sched_publish_pend_reason()), so
# unlike sched_driver_decision_cb_test this target cannot spy on those two
# functions directly. Same real-scheduler_srcs link technique as
# sched_scheduler_connect_test above (reused scheduler_connect_test_srcs
# list), PLUS `-Wl,--wrap=` linker symbol interposition on the three libxce
# client calls this task's retry/rejoin logic makes (xcecli_jobdispatch,
# xcecli_set_pend_reason, xcecli_join) and two more needed only to let this
# file's one real sched_init() setup call reach xcecli_set_reconnect_handler()
# without a live controller (xcecli_start, xcecli_set_reconnect_handler) --
# see the test file's own header comment for why `--wrap` (which redirects
# call sites, not archive members) was required instead of the "test double
# ahead of the archive" trick: all of these symbols share client.c's single
# translation unit inside libxce.a with several OTHER xcecli_* entry points
# this test's real scheduler_srcs dependency graph also needs, so simply
# excluding client.c from the link was not viable.
scheduler_decision_publish_retry_test_srcs = [
'test_scheduler_decision_publish_retry.c']
foreach f : scheduler_srcs
if f != 'src/main.c'
scheduler_decision_publish_retry_test_srcs += ['../' + f]
endif
endforeach
sched_scheduler_decision_publish_retry_test = executable(
'sched_scheduler_decision_publish_retry_test',
scheduler_decision_publish_retry_test_srcs,
include_directories:[incdir],
dependencies: [scheduler_deps, unity_dep],
link_args: ['-Wl,--wrap=xcecli_jobdispatch',
'-Wl,--wrap=xcecli_set_pend_reason',
'-Wl,--wrap=xcecli_join',
'-Wl,--wrap=xcecli_start',
'-Wl,--wrap=xcecli_set_reconnect_handler'],
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',
@ -896,3 +934,6 @@ test('scheduler repl equivalence test',
test('scheduler connect pipeline test',
sched_scheduler_connect_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])
test('scheduler decision publish retry test',
sched_scheduler_decision_publish_retry_test, suite: 'scheduler',
env: ['ASAN_OPTIONS=detect_leaks=0'])

575
src/scheduler/testsuite/test_scheduler_decision_publish_retry.c Normal file
View file

@ -0,0 +1,575 @@
/*
* test_scheduler_decision_publish_retry.c — tests for the scheduler's
* decision-publish retry-until-success-or-shutdown wrapper, composed with
* rejoin retry-on-failure (component-failover-adoption, task 3.2, boundary:
* SchedulerControllerClient).
*
* scheduler.c IS the file under test here (sched_publish_dispatch()/
* sched_publish_pend_reason()), so unlike test_driver_decision_cb.c this
* file cannot spy on those two functions directly -- they ARE the
* production code under test. Instead this file links the REAL
* scheduler_srcs (same technique test_scheduler_connect.c already
* established for this boundary -- see meson.build) and intercepts the
* THREE libxce client-library calls scheduler.c's retry/rejoin logic
* makes -- xcecli_jobdispatch(), xcecli_set_pend_reason(), xcecli_join()
* -- via linker symbol interposition (`-Wl,--wrap=`, meson.build), not by
* excluding client.c from the link: client.c is a SINGLE translation unit
* inside libxce.a that also defines xcecli_ctx_init()/xcecli_start()/
* xcecli_stop()/xcecli_ctx_fini()/xcecli_set_reconnect_handler()/
* xcecli_wakeup()/xcecli_reply_schedadmin() -- every other xcecli_* symbol
* sched_init()/sched_fini()/driver.c/handler.c need -- so excluding
* client.c entirely (the "put the test double before the archive on the
* link line" trick) would silently fail to link (or worse, risk a
* duplicate-definition error the moment any other needed symbol pulls the
* same .o back in) unless this file reimplemented all of those too.
* `--wrap` instead redirects only the three named symbols' CALL SITES to
* this file's __wrap_* doubles, at every call site in the linked
* scheduler_srcs object code, regardless of which archive member
* originally defined them -- the real client.c is still linked (and still
* provides the other xcecli_* entry points sched_init() genuinely needs),
* so nothing else about the production dependency graph is stubbed.
*
* xcecli_start() is ALSO wrapped (forced to always succeed, no I/O)
* purely so the one setup call this file makes to the real sched_init()
* (see test_rejoin_composition_retries_across_publish_calls below) can
* reach xcecli_set_reconnect_handler() without a live controller to
* connect to -- xcecli_start() is the one call in sched_init()'s sequence
* that performs real connect() attempts (client.c); every other xcecli_*
* call sched_init() makes before it (xcecli_ctx_init, xcecli_set_handler)
* is pure local state setup with no I/O, so those run for real, unwrapped.
*
* xcecli_set_reconnect_handler() is ALSO wrapped -- not to change its
* behavior (the real implementation is a trivial field store on the ctx,
* see client.c) but to CAPTURE the callback pointer sched_init()
* registers (sched_on_reconnect(), scheduler.c) without needing to name
* that function directly: sched_on_reconnect() is `static` (deliberately
* kept untestable-directly, per test_scheduler_connect.c's own header
* comment), so the only way to invoke it from this file is through the
* function pointer scheduler.c itself hands to
* xcecli_set_reconnect_handler() -- capturing it here and calling it
* directly is how this file simulates "the library detected a
* transparent reconnect and invoked the registered rejoin hook" without a
* real reconnect ever happening, letting
* test_rejoin_composition_retries_across_publish_calls set the real,
* private sched_needs_rejoin flag (scheduler.c) to true exactly the way
* production code would, then observe its value indirectly through the
* wrapped xcecli_join()'s call count/return sequence across two separate
* sched_publish_dispatch() calls -- the task's own suggested verification
* technique.
*
* Req 7 AC4's "log an extended-retry success distinctly from an ordinary
* single-attempt success" is verified via genuine RUNTIME log capture
* (dup2() redirect of STDOUT_FILENO for the duration of the call under
* test), not a source-text scan -- same mechanism and same rationale
* test_apply_correctness_verification.c's
* test_apply_divergent_loud_fence_log_omits_false_peer_hwm_clause already
* established in the controller testsuite: 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()/INFO() 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).
*/
#include <errno.h>
#include <fcntl.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <unity/unity.h>
#include "client.h"
#include "controller.capnp.h"
#include "preason.h"
#include "scheduler.h"
#include "xce/cluster.h"
#include "xce/errnox.h"
extern int scheduler_req_handler(xce_request_t*, void*);
#define MAX_CANNED_RC 8
/* ════════════════════════════════════════════════════════════════════════
* Test doubles: __wrap_xcecli_jobdispatch / __wrap_xcecli_set_pend_reason
* ════════════════════════════════════════════════════════════════════════ */
static int g_dispatch_returns[MAX_CANNED_RC];
static int g_dispatch_nreturns;
static int g_dispatch_call_count;
static xce_scheduler_t* g_dispatch_stop_after_first_call;
static void reset_dispatch_double(void)
{
memset(g_dispatch_returns, 0, sizeof(g_dispatch_returns));
g_dispatch_nreturns = 0;
g_dispatch_call_count = 0;
g_dispatch_stop_after_first_call = NULL;
}
int __wrap_xcecli_jobdispatch(xcecli_ctx_t c, xce_job_dispatch_t* info)
{
int rc;
(void)c;
(void)info;
rc = (g_dispatch_call_count < g_dispatch_nreturns)
? g_dispatch_returns[g_dispatch_call_count]
: g_dispatch_returns[g_dispatch_nreturns > 0
? g_dispatch_nreturns - 1
: 0];
g_dispatch_call_count++;
if ((g_dispatch_stop_after_first_call != NULL)
&& (g_dispatch_call_count == 1)) {
g_dispatch_stop_after_first_call->core.stop = 1;
}
return rc;
}
static int g_pend_returns[MAX_CANNED_RC];
static int g_pend_nreturns;
static int g_pend_call_count;
static void reset_pend_double(void)
{
memset(g_pend_returns, 0, sizeof(g_pend_returns));
g_pend_nreturns = 0;
g_pend_call_count = 0;
}
int __wrap_xcecli_set_pend_reason(xcecli_ctx_t c, xce_preason_set_t* reasons)
{
int rc;
(void)c;
(void)reasons;
rc = (g_pend_call_count < g_pend_nreturns)
? g_pend_returns[g_pend_call_count]
: g_pend_returns[g_pend_nreturns > 0 ? g_pend_nreturns - 1 : 0];
g_pend_call_count++;
return rc;
}
/* ════════════════════════════════════════════════════════════════════════
* Test double: __wrap_xcecli_join
* ════════════════════════════════════════════════════════════════════════ */
static int g_join_returns[MAX_CANNED_RC];
static int g_join_nreturns;
static int g_join_call_count;
static void reset_join_double(void)
{
memset(g_join_returns, 0, sizeof(g_join_returns));
g_join_nreturns = 0;
g_join_call_count = 0;
}
int __wrap_xcecli_join(xcecli_ctx_t c, int role, hostattrs_t* attrs)
{
int rc;
(void)c;
(void)role;
(void)attrs;
rc = (g_join_call_count < g_join_nreturns)
? g_join_returns[g_join_call_count]
: -EINVAL;
g_join_call_count++;
return rc;
}
/* ════════════════════════════════════════════════════════════════════════
* Test doubles: __wrap_xcecli_start / __wrap_xcecli_set_reconnect_handler
* -- setup-only doubles for the real sched_init() call the rejoin-
* composition test below makes; see file header comment.
* ════════════════════════════════════════════════════════════════════════ */
int __wrap_xcecli_start(xcecli_ctx_t c)
{
(void)c;
return 0;
}
static xcecli_reconnect_cb g_captured_reconnect_cb;
static void* g_captured_reconnect_arg;
int __wrap_xcecli_set_reconnect_handler(xcecli_ctx_t c, xcecli_reconnect_cb cb,
void* arg)
{
(void)c;
g_captured_reconnect_cb = cb;
g_captured_reconnect_arg = arg;
return 0;
}
/* ════════════════════════════════════════════════════════════════════════
* setUp / tearDown
* ════════════════════════════════════════════════════════════════════════ */
void setUp(void)
{
reset_dispatch_double();
reset_pend_double();
reset_join_double();
}
void tearDown(void) { return; }
/* ════════════════════════════════════════════════════════════════════════
* Helpers
* ════════════════════════════════════════════════════════════════════════ */
static xce_job_dispatch_t make_dispatch_fixture(void)
{
xce_job_dispatch_t dispatch;
xce_job_alloc_t* alloc = (xce_job_alloc_t*)calloc(1, sizeof(*alloc));
xce_job_alloc_t** allocs = (xce_job_alloc_t**)calloc(1, sizeof(*allocs));
allocs[0] = alloc;
dispatch.njobs = 1;
dispatch.allocs = allocs;
return dispatch;
}
/* Captures STDOUT_FILENO for the duration of `body` (a single
* sched_publish_dispatch() call in every test below), returning the
* captured text as a malloc'd, NUL-terminated buffer the caller must
* free(). Same mechanism as
* test_apply_correctness_verification.c's
* test_apply_divergent_loud_fence_log_omits_false_peer_hwm_clause -- see
* file header comment. */
static char* capture_stdout_during(xce_scheduler_t* sched,
xce_job_dispatch_t* dispatch, int* rc_out)
{
char tmp[] = "/tmp/xce_sched_pubretry_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_publish_dispatch(sched, dispatch);
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;
}
/* ════════════════════════════════════════════════════════════════════════
* Req 7 AC1/AC2 — retry until success, same decision, not skipped
* ════════════════════════════════════════════════════════════════════════ */
static void test_dispatch_retries_past_not_ready_until_success(void)
{
xce_scheduler_t sched;
xce_job_dispatch_t dispatch = make_dispatch_fixture();
int rc;
memset(&sched, 0, sizeof(sched));
g_dispatch_returns[0] = -E_XCE_NOT_READY;
g_dispatch_returns[1] = -E_XCE_NOT_READY;
g_dispatch_returns[2] = 0;
g_dispatch_nreturns = 3;
rc = sched_publish_dispatch(&sched, &dispatch);
TEST_ASSERT_TRUE_MESSAGE(
rc >= 0,
"a decision publish that eventually succeeds must itself report "
"success to its caller");
TEST_ASSERT_EQUAL_INT_MESSAGE(
3, g_dispatch_call_count,
"the retry loop must re-invoke xcecli_jobdispatch() once per "
"-E_XCE_NOT_READY for the SAME decision -- never skipped, never "
"dropped to the next queued one");
TEST_ASSERT_EQUAL_INT_MESSAGE(
0, g_join_call_count,
"no rejoin was pending, so xcecli_join() must not be called");
printf("test_dispatch_retries_past_not_ready_until_success: PASS\n");
}
static void test_pend_reason_retries_past_not_ready_until_success(void)
{
xce_scheduler_t sched;
xce_pendreason_t* reason;
xce_pendreason_t* reasons[1];
xce_preason_set_t preason_set;
int rc;
memset(&sched, 0, sizeof(sched));
reason = make_pendreason(7);
reasons[0] = reason;
preason_set.nreasons = 1;
preason_set.reasons = reasons;
g_pend_returns[0] = -E_XCE_NOT_READY;
g_pend_returns[1] = 0;
g_pend_nreturns = 2;
rc = sched_publish_pend_reason(&sched, &preason_set);
TEST_ASSERT_TRUE_MESSAGE(
rc >= 0, "a pend-reason publish that eventually succeeds must "
"itself report success");
TEST_ASSERT_EQUAL_INT_MESSAGE(
2, g_pend_call_count,
"the retry loop must re-invoke xcecli_set_pend_reason() once per "
"-E_XCE_NOT_READY, same requirement as the dispatch path");
printf("test_pend_reason_retries_past_not_ready_until_success: PASS\n");
}
/* ════════════════════════════════════════════════════════════════════════
* Req 7 AC1/AC2 — shutdown mid-retry breaks the loop within one cycle
* ════════════════════════════════════════════════════════════════════════ */
static void test_dispatch_shutdown_breaks_retry_within_one_cycle(void)
{
xce_scheduler_t sched;
xce_job_dispatch_t dispatch = make_dispatch_fixture();
int rc;
memset(&sched, 0, sizeof(sched));
sched.core.stop = 0;
/* First attempt returns -E_XCE_NOT_READY AND (as a side effect,
* simulating an operator-triggered shutdown arriving concurrently)
* sets core.stop=1. A second canned return of 0 is present ONLY to
* prove it is never reached -- if the loop's stop-check were at the
* bottom instead of the top of the loop, this second call WOULD
* happen and WOULD succeed, hiding the bug. */
g_dispatch_returns[0] = -E_XCE_NOT_READY;
g_dispatch_returns[1] = 0;
g_dispatch_nreturns = 2;
g_dispatch_stop_after_first_call = &sched;
rc = sched_publish_dispatch(&sched, &dispatch);
TEST_ASSERT_TRUE_MESSAGE(
rc < 0,
"abandoning a publish because of shutdown must not be reported as "
"success");
TEST_ASSERT_EQUAL_INT_MESSAGE(
1, g_dispatch_call_count,
"core.stop must be checked BEFORE each attempt: setting it after "
"the first call must prevent any second call, not allow one more "
"full retry cycle to run first");
printf("test_dispatch_shutdown_breaks_retry_within_one_cycle: PASS\n");
}
/* ════════════════════════════════════════════════════════════════════════
* Req 7 AC4 — extended-retry success logs distinctly from an ordinary
* first-attempt success (genuine runtime log capture)
* ════════════════════════════════════════════════════════════════════════ */
static void test_extended_retry_success_logs_distinctly(void)
{
xce_scheduler_t sched;
xce_job_dispatch_t dispatch = make_dispatch_fixture();
int rc;
char* buf;
memset(&sched, 0, sizeof(sched));
g_dispatch_returns[0] = -E_XCE_NOT_READY;
g_dispatch_returns[1] = 0;
g_dispatch_nreturns = 2;
buf = capture_stdout_during(&sched, &dispatch, &rc);
TEST_ASSERT_TRUE(rc >= 0);
TEST_ASSERT_NOT_NULL_MESSAGE(
strstr(buf, "extended retry"),
"an extended-retry success (more than one attempt) must log "
"something distinguishing it from an ordinary first-attempt "
"success, so operators can tell 'delayed but delivered' from "
"'delivered on the first try'");
free(buf);
printf("test_extended_retry_success_logs_distinctly: PASS\n");
}
static void test_first_attempt_success_does_not_log_extended_retry(void)
{
xce_scheduler_t sched;
xce_job_dispatch_t dispatch = make_dispatch_fixture();
int rc;
char* buf;
memset(&sched, 0, sizeof(sched));
g_dispatch_returns[0] = 0;
g_dispatch_nreturns = 1;
buf = capture_stdout_during(&sched, &dispatch, &rc);
TEST_ASSERT_TRUE(rc >= 0);
TEST_ASSERT_NULL_MESSAGE(
strstr(buf, "extended retry"),
"an ordinary first-attempt success must NOT log the same "
"extended-retry-success message -- this is what makes the two "
"cases distinct, not merely that the extended case logs "
"something");
free(buf);
printf("test_first_attempt_success_does_not_log_extended_retry: PASS\n");
}
/* ════════════════════════════════════════════════════════════════════════
* Req 2 AC5 / Req 7 — rejoin composition: after a publish loop succeeds,
* a pending rejoin is issued as a separate follow-up, and stays pending
* (retried on the NEXT publish) until it itself succeeds.
* ════════════════════════════════════════════════════════════════════════ */
static void test_rejoin_composition_retries_across_publish_calls(void)
{
xce_scheduler_t sched;
xce_client_config_t config;
xce_cluster_config_t cluster_cfg;
xce_job_dispatch_t dispatch = make_dispatch_fixture();
int rc;
/* --- Setup: reach a real sched_init() far enough that it calls the
* real xcecli_set_reconnect_handler(sched->apictx, sched_on_reconnect,
* NULL) -- captured here via __wrap_xcecli_set_reconnect_handler --
* then force the join sched_init() itself performs to fail (this
* setup call, consumed from g_join_returns[0] below, is NOT part of
* the later assertions) so sched_init() returns before ever spinning
* up the picker/allocator threads. This file only needs a populated
* sched->apictx and the captured callback pointer, not a running
* scheduler -- see file header comment for the full rationale. */
memset(&config, 0, sizeof(config));
config.nmasters = 1;
config.masters[0] = "127.0.0.1";
config.master_ports[0] = 19999;
config.from = "sched_pubretry_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 = 19999;
strncpy(cluster_cfg.identity, "sched_pubretry_test",
sizeof(cluster_cfg.identity) - 1);
g_join_returns[0] = -EINVAL; /* sched_init()'s own join: fail on purpose */
g_join_nreturns = 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()'s own join was deliberately made to fail "
"so no picker/allocator thread is spun up by this setup step");
TEST_ASSERT_NOT_NULL_MESSAGE(
g_captured_reconnect_cb,
"fixture: sched_init() must register the reconnect handler (and "
"this file's wrap must capture it) before its xcecli_join() call");
/* Simulate "the library transparently reconnected" by invoking the
* captured real sched_on_reconnect() directly -- sets the real,
* private sched_needs_rejoin flag (scheduler.c) to true, exactly as a
* genuine reconnect would (component-failover-adoption, task 3.1). */
g_captured_reconnect_cb(sched.apictx, g_captured_reconnect_arg);
/* --- Call 1: publish succeeds; the follow-up rejoin is attempted and
* fails -- the publish itself must still report success. --- */
reset_dispatch_double();
g_dispatch_returns[0] = 0;
g_dispatch_nreturns = 1;
reset_join_double();
g_join_returns[0] = -E_XCE_NOT_READY; /* the pending rejoin itself fails */
g_join_nreturns = 1;
rc = sched_publish_dispatch(&sched, &dispatch);
TEST_ASSERT_TRUE_MESSAGE(
rc >= 0,
"the decision publish itself must report success even though the "
"follow-up rejoin call failed");
TEST_ASSERT_EQUAL_INT_MESSAGE(
1, g_join_call_count,
"a pending rejoin must be attempted as a separate follow-up call "
"once the publish loop returns success");
/* --- Call 2: publish succeeds again; the flag must still be set
* (call 1's rejoin failed), so THIS call must retry it -- let it
* succeed this time. Deliberately does NOT call reset_join_double()
* so g_join_call_count keeps accumulating across both calls. --- */
reset_dispatch_double();
g_dispatch_returns[0] = 0;
g_dispatch_nreturns = 1;
g_join_returns[1] = 0; /* second slot in the SAME sequence array */
g_join_nreturns = 2;
rc = sched_publish_dispatch(&sched, &dispatch);
TEST_ASSERT_TRUE(rc >= 0);
TEST_ASSERT_EQUAL_INT_MESSAGE(
2, g_join_call_count,
"the still-pending rejoin flag must cause the SECOND publish call "
"to retry xcecli_join() again -- on its own retry cadence, not a "
"separate loop");
/* --- Call 3: the rejoin succeeded on call 2, so the flag must now be
* clear -- no further xcecli_join() calls. --- */
reset_dispatch_double();
g_dispatch_returns[0] = 0;
g_dispatch_nreturns = 1;
rc = sched_publish_dispatch(&sched, &dispatch);
TEST_ASSERT_TRUE(rc >= 0);
TEST_ASSERT_EQUAL_INT_MESSAGE(
2, g_join_call_count,
"once the rejoin has succeeded, sched_needs_rejoin must be "
"cleared -- a later publish success must not attempt another "
"join");
printf("test_rejoin_composition_retries_across_publish_calls: PASS\n");
}
int main(void)
{
UNITY_BEGIN();
RUN_TEST(test_dispatch_retries_past_not_ready_until_success);
RUN_TEST(test_pend_reason_retries_past_not_ready_until_success);
RUN_TEST(test_dispatch_shutdown_breaks_retry_within_one_cycle);
RUN_TEST(test_extended_retry_success_logs_distinctly);
RUN_TEST(test_first_attempt_success_does_not_log_extended_retry);
RUN_TEST(test_rejoin_composition_retries_across_publish_calls);
return UNITY_END();
}