diff --git a/src/worker/include/xcecd_worker_utils.h b/src/worker/include/xcecd_worker_utils.h index d437eb1d..23c397e3 100644 --- a/src/worker/include/xcecd_worker_utils.h +++ b/src/worker/include/xcecd_worker_utils.h @@ -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 diff --git a/src/worker/src/controller_connector.c b/src/worker/src/controller_connector.c index 48784109..bbeeeb6b 100755 --- a/src/worker/src/controller_connector.c +++ b/src/worker/src/controller_connector.c @@ -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); " diff --git a/src/worker/src/xcecd_worker_utils.c b/src/worker/src/xcecd_worker_utils.c index 900a608b..9a5c0b3f 100644 --- a/src/worker/src/xcecd_worker_utils.c +++ b/src/worker/src/xcecd_worker_utils.c @@ -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) diff --git a/src/worker/testsuite/test_controller_connector.c b/src/worker/testsuite/test_controller_connector.c index bb1513b9..5d93b62d 100644 --- a/src/worker/testsuite/test_controller_connector.c +++ b/src/worker/testsuite/test_controller_connector.c @@ -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) diff --git a/src/worker/testsuite/test_finished_job_retention.c b/src/worker/testsuite/test_finished_job_retention.c index 884ab44a..ff60612d 100644 --- a/src/worker/testsuite/test_finished_job_retention.c +++ b/src/worker/testsuite/test_finished_job_retention.c @@ -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(); } diff --git a/src/worker/testsuite/test_worker_launch_type_dispatch.c b/src/worker/testsuite/test_worker_launch_type_dispatch.c index 1e1e287a..ab5bdc88 100644 --- a/src/worker/testsuite/test_worker_launch_type_dispatch.c +++ b/src/worker/testsuite/test_worker_launch_type_dispatch.c @@ -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) { }