fix(worker): don't let a failover outage expire a finished job's retained outcome (bug-1859)
The finish notify is fire-and-forget, so the retained outcome is the only way a controller learns a finish lost in a failover. Its 60s window ran from the finish time, so an 84s outage expired it before the rejoin resync could report it; the controller got "unrecognized" and requeued a completed job (then stuck: scheduler kept RUN, see bug-1860). Expiry is now measured from max(finish_time, last reconnect) (xcecd_on_reconnect), and nothing expires while checking_connection's hello probe reports the connection unhealthy. The reported finish_time_ms is unchanged. Verified: worker suite 10/10 (two new retention tests, red first); chaotic HA soak 3/3 PASS (was 1 in 3 failing). Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
b5d59253e5
commit
7c98c056c9
6 changed files with 134 additions and 5 deletions
|
|
@ -123,6 +123,16 @@ void xcecd_finished_job_retention_set_window_ms(uint64_t window_ms);
|
|||
/* Returns the currently effective retention window in milliseconds. */
|
||||
uint64_t xcecd_finished_job_retention_get_window_ms(void);
|
||||
|
||||
/* bug-1859: the finish notify is fire-and-forget, so time the worker spends
|
||||
* unable to reach a controller must not count against the window. After a
|
||||
* reconnect every retained entry gets a full window again (expiry is
|
||||
* measured from max(finish_time_ms, last reconnect)), and while the
|
||||
* connection is known unhealthy nothing expires. finish_time_ms itself
|
||||
* (what gets reported) is unchanged. Both only write scalars, so they are
|
||||
* safe to call from the reconnect callback. */
|
||||
void xcecd_finished_job_retention_note_reconnect(uint64_t now_ms);
|
||||
void xcecd_finished_job_retention_set_connection_healthy(bool healthy);
|
||||
|
||||
/* worker-rejoin-resync task 4.2 (Requirement 3.6): copies every currently
|
||||
* retained, non-expired (as of now_ms) finished-job entry into a freshly
|
||||
* heap-allocated array of xcecd_finished_job_t VALUES (not linked -- each
|
||||
|
|
|
|||
|
|
@ -292,6 +292,8 @@ static void format_candidate_list(char* buf, size_t bufsz)
|
|||
* touches xcecd_resync.c's own statics (a bool and a small heap buffer), no
|
||||
* I/O and no xcecli_* call.
|
||||
*/
|
||||
static uint64_t resync_wallclock_now_ms(void);
|
||||
|
||||
static int xcecd_on_reconnect(xcecli_ctx_t ctx, void* arg)
|
||||
{
|
||||
(void)ctx;
|
||||
|
|
@ -299,6 +301,10 @@ static int xcecd_on_reconnect(xcecli_ctx_t ctx, void* arg)
|
|||
|
||||
needs_rejoin = true;
|
||||
xcecd_resync_reset();
|
||||
/* bug-1859: a finish notify lost in the outage is only recoverable via
|
||||
* the retained outcome -- restart its window so the rejoin resync can
|
||||
* still report it. Scalar writes only, same constraint as above. */
|
||||
xcecd_finished_job_retention_note_reconnect(resync_wallclock_now_ms());
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
|
@ -506,6 +512,10 @@ int checking_connection()
|
|||
{
|
||||
int ret = xcecli_hello(xcecli_ctx, "hello");
|
||||
|
||||
/* bug-1859: retained finished-job outcomes must not expire while no
|
||||
* controller can be told about them */
|
||||
xcecd_finished_job_retention_set_connection_healthy(ret == 0);
|
||||
|
||||
if (ret != 0) {
|
||||
XCECD_LOG(XCE_LOG_ERROR,
|
||||
"Connection to controller not currently healthy (rc=%d); "
|
||||
|
|
|
|||
|
|
@ -27,17 +27,37 @@ static xcecd_finished_job_t* g_finished_job_retention = NULL;
|
|||
static uint64_t g_finished_job_retention_window_ms
|
||||
= XCECD_FINISHED_JOB_RETENTION_DEFAULT_WINDOW_MS;
|
||||
|
||||
/* bug-1859: see xcecd_finished_job_retention_note_reconnect() in the
|
||||
* header. Scalars only, written from the reconnect callback and the
|
||||
* connection probe, read by the expiry check below. */
|
||||
static uint64_t g_finished_job_retention_last_reconnect_ms = 0;
|
||||
static bool g_finished_job_retention_conn_healthy = true;
|
||||
|
||||
static bool finished_job_entry_expired(const xcecd_finished_job_t* e,
|
||||
uint64_t now_ms, uint64_t window_ms)
|
||||
{
|
||||
if (now_ms < e->finish_time_ms) {
|
||||
/* now_ms predates the entry's own finish time (clock skew, or a
|
||||
* caller/test passing an earlier now_ms) -- treat as not yet
|
||||
* expired rather than underflow the unsigned subtraction below. */
|
||||
uint64_t start_ms = e->finish_time_ms;
|
||||
|
||||
/* nothing can reach a controller right now -- keep everything */
|
||||
if (!g_finished_job_retention_conn_healthy) {
|
||||
return false;
|
||||
}
|
||||
|
||||
return (now_ms - e->finish_time_ms) > window_ms;
|
||||
/* the window restarts at the last reconnect: an outcome whose finish
|
||||
* notify was lost in an outage gets a full window for the rejoin
|
||||
* resync to report it */
|
||||
if (g_finished_job_retention_last_reconnect_ms > start_ms) {
|
||||
start_ms = g_finished_job_retention_last_reconnect_ms;
|
||||
}
|
||||
|
||||
if (now_ms < start_ms) {
|
||||
/* now_ms predates the window start (clock skew, or a caller/test
|
||||
* passing an earlier now_ms) -- treat as not yet expired rather
|
||||
* than underflow the unsigned subtraction below. */
|
||||
return false;
|
||||
}
|
||||
|
||||
return (now_ms - start_ms) > window_ms;
|
||||
}
|
||||
|
||||
/* Drops every entry older than window_ms as of now_ms. This is the lazy
|
||||
|
|
@ -582,6 +602,17 @@ uint64_t xcecd_finished_job_retention_get_window_ms(void)
|
|||
return g_finished_job_retention_window_ms;
|
||||
}
|
||||
|
||||
void xcecd_finished_job_retention_note_reconnect(uint64_t now_ms)
|
||||
{
|
||||
g_finished_job_retention_last_reconnect_ms = now_ms;
|
||||
g_finished_job_retention_conn_healthy = true;
|
||||
}
|
||||
|
||||
void xcecd_finished_job_retention_set_connection_healthy(bool healthy)
|
||||
{
|
||||
g_finished_job_retention_conn_healthy = healthy;
|
||||
}
|
||||
|
||||
int xcecd_finished_job_retention_snapshot(uint64_t now_ms,
|
||||
xcecd_finished_job_t** out,
|
||||
int* out_n)
|
||||
|
|
@ -637,6 +668,8 @@ void xcecd_finished_job_retention_reset(void)
|
|||
|
||||
g_finished_job_retention = NULL;
|
||||
g_finished_job_retention_window_ms = XCECD_FINISHED_JOB_RETENTION_DEFAULT_WINDOW_MS;
|
||||
g_finished_job_retention_last_reconnect_ms = 0;
|
||||
g_finished_job_retention_conn_healthy = true;
|
||||
}
|
||||
|
||||
xcecd_worker_t* find_worker_by_pid(void* worker_list, int pid)
|
||||
|
|
|
|||
|
|
@ -173,6 +173,17 @@ void xcecd_finished_job_retention_set_window_ms(uint64_t window_ms)
|
|||
g_retention_window_ms_set = window_ms;
|
||||
}
|
||||
|
||||
/* bug-1859 hooks called from xcecd_on_reconnect()/checking_connection() */
|
||||
void xcecd_finished_job_retention_note_reconnect(uint64_t now_ms)
|
||||
{
|
||||
(void)now_ms;
|
||||
}
|
||||
|
||||
void xcecd_finished_job_retention_set_connection_healthy(bool healthy)
|
||||
{
|
||||
(void)healthy;
|
||||
}
|
||||
|
||||
bool xcecd_resync_is_pending(void) { return g_test_resync_pending; }
|
||||
|
||||
void xcecd_resync_reset(void)
|
||||
|
|
|
|||
|
|
@ -318,6 +318,59 @@ static void test_lookup_after_expiry_prunes_entry_permanently(void)
|
|||
xcecd_finished_job_lookup(70003, finish_time_ms, &out));
|
||||
}
|
||||
|
||||
/* bug-1859: the finish notify is fire-and-forget, so this store is the only
|
||||
* way a controller learns an outcome that was lost in a failover. A job that
|
||||
* finished before an outage longer than the window must still be reported
|
||||
* by the rejoin resync: the window restarts at reconnect. */
|
||||
static void test_reconnect_restarts_window_for_outcome_finished_before_outage(void)
|
||||
{
|
||||
const uint64_t finish_time_ms = 4000000ULL;
|
||||
const uint64_t window_ms = 60000ULL;
|
||||
const uint64_t reconnect_ms = finish_time_ms + 84000ULL; /* 84s outage */
|
||||
|
||||
xcecd_finished_job_retention_set_window_ms(window_ms);
|
||||
xcecd_retain_finished_job(70004, WORKER_DONE, 0, finish_time_ms);
|
||||
xcecd_finished_job_retention_note_reconnect(reconnect_ms);
|
||||
|
||||
xcecd_finished_job_t out;
|
||||
memset(&out, 0, sizeof(out));
|
||||
|
||||
TEST_ASSERT_TRUE_MESSAGE(
|
||||
xcecd_finished_job_lookup(70004, reconnect_ms + 10, &out),
|
||||
"an outcome that finished before the outage must survive until the "
|
||||
"rejoin resync right after reconnect");
|
||||
/* the reported finish time is the real one, not the reconnect time */
|
||||
TEST_ASSERT_EQUAL_UINT64(finish_time_ms, out.finish_time_ms);
|
||||
|
||||
/* ...and still expires a full window after reconnect */
|
||||
TEST_ASSERT_FALSE(xcecd_finished_job_lookup(
|
||||
70004, reconnect_ms + window_ms + 1, &out));
|
||||
}
|
||||
|
||||
/* bug-1859: while the connection is known unhealthy nothing can be reported
|
||||
* to a controller, so nothing may be pruned; expiry resumes once healthy. */
|
||||
static void test_unhealthy_connection_suspends_expiry(void)
|
||||
{
|
||||
const uint64_t finish_time_ms = 5000000ULL;
|
||||
const uint64_t window_ms = 1000ULL;
|
||||
|
||||
xcecd_finished_job_retention_set_window_ms(window_ms);
|
||||
xcecd_retain_finished_job(70005, WORKER_DONE, 0, finish_time_ms);
|
||||
xcecd_finished_job_retention_set_connection_healthy(false);
|
||||
|
||||
xcecd_finished_job_t out;
|
||||
memset(&out, 0, sizeof(out));
|
||||
|
||||
TEST_ASSERT_TRUE_MESSAGE(
|
||||
xcecd_finished_job_lookup(70005, finish_time_ms + window_ms + 100,
|
||||
&out),
|
||||
"no entry may expire while the connection is unhealthy");
|
||||
|
||||
xcecd_finished_job_retention_set_connection_healthy(true);
|
||||
TEST_ASSERT_FALSE(xcecd_finished_job_lookup(
|
||||
70005, finish_time_ms + window_ms + 100, &out));
|
||||
}
|
||||
|
||||
/* ════════════════════════════════════════════════════════════════════════
|
||||
* main
|
||||
* ════════════════════════════════════════════════════════════════════════ */
|
||||
|
|
@ -336,6 +389,8 @@ int main(void)
|
|||
RUN_TEST(test_lookup_within_window_returns_outcome);
|
||||
RUN_TEST(test_lookup_just_outside_window_returns_not_found);
|
||||
RUN_TEST(test_lookup_after_expiry_prunes_entry_permanently);
|
||||
RUN_TEST(test_reconnect_restarts_window_for_outcome_finished_before_outage);
|
||||
RUN_TEST(test_unhealthy_connection_suspends_expiry);
|
||||
|
||||
return UNITY_END();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -252,6 +252,16 @@ void xcecd_finished_job_retention_set_window_ms(uint64_t window_ms)
|
|||
(void)window_ms;
|
||||
}
|
||||
|
||||
void xcecd_finished_job_retention_note_reconnect(uint64_t now_ms)
|
||||
{
|
||||
(void)now_ms;
|
||||
}
|
||||
|
||||
void xcecd_finished_job_retention_set_connection_healthy(bool healthy)
|
||||
{
|
||||
(void)healthy;
|
||||
}
|
||||
|
||||
bool xcecd_resync_is_pending(void) { return false; }
|
||||
|
||||
void xcecd_resync_reset(void) { }
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue