fix(controller): persist a failed job's finish_time through WAL_JOB_FAIL (bug-1747)
The FAIL half of bug-1747: apply_job_fail_transition() stamps finish_time = time(NULL) at APPLY time, and WAL_JOB_FAIL carried no time, so a failed job's finish_time changed on every restart replay / standby catch-up, and a worker-reported failure found on rejoin could never keep the worker's own time. - WAL_JOB_FAIL's trailer grows to int32_t code + uint64_t finish_time (seconds). Decoded by length: 16 B (pre-bug-1754), 20 B (code only, bug-1754) and 28 B (code + time) all apply as before for what they carry. - record_job_fail_fact_at(c, job, failinfo, finish_time_ms), the fail-side sibling of record_job_finish_fact_at(): 0 = receipt time. on_job_fail() uses receipt time; resync's finishedFail branch passes the worker- reported finish_time_ms (Req 4.2 now holds for failures too). - Both apply seams adopt the wire time after the transition. 4 new tests in test_wal_job_fail_fact.c (3 RED before the fix) and the resync round-trip now asserts the worker-reported failure time (RED before). Controller suite 1232/0. Factstream HA soak (2 rounds, 60 jobs) PASS. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
21e72726f0
commit
1d6c04c28a
GPG key ID:
80533536B0EB651C
9 changed files with 349 additions and 68 deletions
|
|
@ -32,6 +32,7 @@
|
|||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include <time.h>
|
||||
|
||||
#include <unity/unity.h>
|
||||
|
||||
|
|
@ -395,6 +396,228 @@ static void test_wjfl_real_on_job_fail_survives_restart(void)
|
|||
fprintf(stderr, "cleanup warning: rm -rf %s failed\n", tmp);
|
||||
}
|
||||
|
||||
/* ── bug-1747: WAL_JOB_FAIL also carries the failure time ─────────────── */
|
||||
|
||||
/* persist_submit_job_fail_fact()'s full shape: header + int32_t code +
|
||||
* uint64_t finish_time (seconds). */
|
||||
typedef struct {
|
||||
wal_jobstate_t js;
|
||||
int32_t code;
|
||||
uint64_t finish_time;
|
||||
} __attribute__((packed)) wjfl_fail_payload_t2;
|
||||
|
||||
static wjfl_fail_payload_t2 wjfl_fail_payload2(uint64_t jobid, uint32_t stage,
|
||||
int32_t code, uint64_t ft)
|
||||
{
|
||||
wjfl_fail_payload_t2 pl;
|
||||
|
||||
memset(&pl, 0, sizeof(pl));
|
||||
pl.js.jobid = jobid;
|
||||
pl.js._pad = WAL_JOB_FAIL_STAGE_FLAG | stage;
|
||||
pl.code = code;
|
||||
pl.finish_time = ft;
|
||||
return pl;
|
||||
}
|
||||
|
||||
static void test_wjfl_apply_committed_entry_adopts_wire_finish_time(void)
|
||||
{
|
||||
wjfl_setup();
|
||||
|
||||
job_t* j = wjfl_seed_job(74101, JOB_STAT_RUN);
|
||||
TEST_ASSERT_NOT_NULL(j);
|
||||
|
||||
wjfl_fail_payload_t2 pl
|
||||
= wjfl_fail_payload2(74101, JOB_EXEC_STAGE_MAIN, 42, 1700000500ULL);
|
||||
TEST_ASSERT_EQUAL_INT(0, apply_committed_entry(&g_wjfl_p, WAL_JOB_FAIL,
|
||||
410, &pl, sizeof(pl)));
|
||||
|
||||
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_EXIT, j->job.state);
|
||||
wjfl_assert_exit_code(j->job.exit_info, 42);
|
||||
TEST_ASSERT_EQUAL_UINT64_MESSAGE(1700000500ULL,
|
||||
(uint64_t)j->job.finish_time,
|
||||
"bug-1747: a failed job's finish_time must be the wire-carried "
|
||||
"failure time, not apply-time time(NULL)");
|
||||
|
||||
wjfl_teardown();
|
||||
}
|
||||
|
||||
static void test_wjfl_apply_committed_entry_code_only_payload_compat(void)
|
||||
{
|
||||
wjfl_setup();
|
||||
|
||||
job_t* j = wjfl_seed_job(74102, JOB_STAT_RUN);
|
||||
TEST_ASSERT_NOT_NULL(j);
|
||||
|
||||
time_t before = time(NULL);
|
||||
wjfl_fail_payload_t pl = wjfl_fail_payload(74102, JOB_EXEC_STAGE_MAIN, 7);
|
||||
TEST_ASSERT_EQUAL_INT(0, apply_committed_entry(&g_wjfl_p, WAL_JOB_FAIL,
|
||||
411, &pl, sizeof(pl)));
|
||||
|
||||
wjfl_assert_exit_code(j->job.exit_info, 7);
|
||||
TEST_ASSERT_TRUE_MESSAGE(j->job.finish_time >= before,
|
||||
"a code-only (bug-1754-era) payload carries no time: finish_time "
|
||||
"stays apply-time, as before");
|
||||
|
||||
wjfl_teardown();
|
||||
}
|
||||
|
||||
static void test_wjfl_apply_wal_entry_adopts_wire_finish_time(void)
|
||||
{
|
||||
wjfl_setup();
|
||||
|
||||
job_t* j = wjfl_seed_job(74103, JOB_STAT_RUN);
|
||||
TEST_ASSERT_NOT_NULL(j);
|
||||
|
||||
wjfl_fail_payload_t2 pl
|
||||
= wjfl_fail_payload2(74103, JOB_EXEC_STAGE_MAIN, 5, 1700000600ULL);
|
||||
TEST_ASSERT_EQUAL_INT(0, apply_wal_entry(&g_wjfl_p, WAL_JOB_FAIL, 412,
|
||||
&pl, sizeof(pl)));
|
||||
|
||||
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_EXIT, j->job.state);
|
||||
wjfl_assert_exit_code(j->job.exit_info, 5);
|
||||
TEST_ASSERT_EQUAL_UINT64(1700000600ULL, (uint64_t)j->job.finish_time);
|
||||
|
||||
wjfl_teardown();
|
||||
}
|
||||
|
||||
/* Real record_job_fail_fact_at() + WAL write + persist_recover(): a failure
|
||||
* reported with an old time (as a rejoining worker reports one) keeps that
|
||||
* time across a restart; a 0 time falls back to receipt time. */
|
||||
static void test_wjfl_record_fail_fact_at_survives_restart(void)
|
||||
{
|
||||
char tmp[256];
|
||||
strcpy(tmp, "/tmp/xce_wjfl_failat_XXXXXX");
|
||||
TEST_ASSERT_NOT_NULL(mkdtemp(tmp));
|
||||
|
||||
const uint64_t jobid_old = 74104;
|
||||
const uint64_t jobid_zero = 74105;
|
||||
const uint64_t old_s = (uint64_t)time(NULL) - 86400;
|
||||
time_t before, after;
|
||||
|
||||
{
|
||||
persist_config_t cfg;
|
||||
cluster_t c;
|
||||
kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT,
|
||||
.nslots = 32, .delta = 16 };
|
||||
const uint64_t ids[2] = { jobid_old, jobid_zero };
|
||||
int k;
|
||||
|
||||
if (makeString((char*)"wjfl-failat-probe") == NULL)
|
||||
xstring_table_init(64);
|
||||
if (!g_wjfl_jobobj_ready) {
|
||||
TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32));
|
||||
g_wjfl_jobobj_ready = 1;
|
||||
}
|
||||
|
||||
memset(&c, 0, sizeof(c));
|
||||
TEST_ASSERT_EQUAL_INT(0,
|
||||
xce_hook_init(&c.hooksys, XCE_CONTROLLER_HOOK_MAX));
|
||||
c.jobmgr = makeInstance(&xkvmgr_type, &int_params);
|
||||
TEST_ASSERT_NOT_NULL(c.jobmgr);
|
||||
|
||||
cfg = persist_config_defaults();
|
||||
cfg.wal_sync_ms = 5000;
|
||||
persist_set_workdir(&c.persist, tmp, "ctrl-test");
|
||||
TEST_ASSERT_EQUAL_INT(0, persist_init(&c.persist, &cfg, NULL));
|
||||
TEST_ASSERT_EQUAL_INT(0, persist_start(&c.persist));
|
||||
|
||||
for (k = 0; k < 2; k++) {
|
||||
job_submit_data_t submit;
|
||||
xce_job_t jdef;
|
||||
uint8_t* buf = NULL;
|
||||
size_t sz = 0;
|
||||
job_t* j;
|
||||
|
||||
memset(&submit, 0, sizeof(submit));
|
||||
submit.queue = (char*)"oq";
|
||||
submit.user = (char*)"tester";
|
||||
memset(&jdef, 0, sizeof(jdef));
|
||||
jdef.jobid = ids[k];
|
||||
jdef.state = JOB_STAT_PEND;
|
||||
jdef.submit = &submit;
|
||||
|
||||
TEST_ASSERT_EQUAL_INT(0, wjfl_encode_job(&jdef, &buf, &sz));
|
||||
TEST_ASSERT_EQUAL_INT(0, persist_wal_async(&c.persist,
|
||||
WAL_JOB_SUBMIT, buf, (uint32_t)sz));
|
||||
free(buf);
|
||||
|
||||
jdef.state = JOB_STAT_RUN;
|
||||
j = makeJob(c.jobmgr, &jdef);
|
||||
TEST_ASSERT_NOT_NULL(j);
|
||||
|
||||
job_failinfo_t fail = { .jobid = ids[k],
|
||||
.stage = JOB_EXEC_STAGE_MAIN,
|
||||
.code = 3,
|
||||
.detail = (char*)"exit" };
|
||||
before = time(NULL);
|
||||
lock(j, RTTI_LOCK_WRITE);
|
||||
TEST_ASSERT_EQUAL_INT(0, record_job_fail_fact_at(&c, j, &fail,
|
||||
(ids[k] == jobid_old) ? old_s * 1000ULL : 0));
|
||||
unlock(j);
|
||||
after = time(NULL);
|
||||
}
|
||||
|
||||
TEST_ASSERT_EQUAL_INT(0, persist_flush(&c.persist));
|
||||
persist_stop(&c.persist);
|
||||
persist_fini(&c.persist);
|
||||
dereference(c.jobmgr);
|
||||
xce_hook_fini(&c.hooksys);
|
||||
}
|
||||
|
||||
{
|
||||
persist_t* p2 = calloc(1, sizeof(persist_t));
|
||||
cluster_t c2;
|
||||
job_t* j_old = NULL;
|
||||
job_t* j_zero = NULL;
|
||||
kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT,
|
||||
.nslots = 32, .delta = 16 };
|
||||
persist_config_t cfg = persist_config_defaults();
|
||||
|
||||
TEST_ASSERT_NOT_NULL(p2);
|
||||
cfg.wal_sync_ms = 5000;
|
||||
persist_set_workdir(p2, tmp, "ctrl-test");
|
||||
TEST_ASSERT_EQUAL_INT(0, persist_init(p2, &cfg, NULL));
|
||||
TEST_ASSERT_EQUAL_INT(0, persist_start(p2));
|
||||
|
||||
memset(&c2, 0, sizeof(c2));
|
||||
TEST_ASSERT_EQUAL_INT(0,
|
||||
xce_hook_init(&c2.hooksys, XCE_CONTROLLER_HOOK_MAX));
|
||||
c2.jobmgr = makeInstance(&xkvmgr_type, &int_params);
|
||||
TEST_ASSERT_NOT_NULL(c2.jobmgr);
|
||||
|
||||
p2->cluster = &c2;
|
||||
TEST_ASSERT_EQUAL_INT(0, persist_recover(p2));
|
||||
|
||||
lock(c2.jobmgr, RTTI_LOCK_READ);
|
||||
key_find(c2.jobmgr, (void*)jobid_old, (void**)&j_old);
|
||||
key_find(c2.jobmgr, (void*)jobid_zero, (void**)&j_zero);
|
||||
unlock(c2.jobmgr);
|
||||
TEST_ASSERT_NOT_NULL(j_old);
|
||||
TEST_ASSERT_NOT_NULL(j_zero);
|
||||
|
||||
TEST_ASSERT_EQUAL_UINT32(JOB_STAT_EXIT, j_old->job.state);
|
||||
wjfl_assert_exit_code(j_old->job.exit_info, 3);
|
||||
TEST_ASSERT_EQUAL_UINT64_MESSAGE(old_s, (uint64_t)j_old->job.finish_time,
|
||||
"bug-1747: the reported failure time must survive restart "
|
||||
"replay, not become replay-time time(NULL)");
|
||||
|
||||
TEST_ASSERT_TRUE_MESSAGE(j_zero->job.finish_time >= before
|
||||
&& j_zero->job.finish_time <= after,
|
||||
"a 0 (unknown) failure time falls back to receipt time");
|
||||
|
||||
xce_hook_fini(&c2.hooksys);
|
||||
dereference(c2.jobmgr);
|
||||
persist_stop(p2);
|
||||
persist_fini(p2);
|
||||
free(p2);
|
||||
}
|
||||
|
||||
char cmd[512];
|
||||
snprintf(cmd, sizeof(cmd), "rm -rf %s", tmp);
|
||||
if (system(cmd) != 0)
|
||||
fprintf(stderr, "cleanup warning: rm -rf %s failed\n", tmp);
|
||||
}
|
||||
|
||||
/* ── Test group entry point ───────────────────────────────────────────── */
|
||||
|
||||
void test_wal_job_fail_fact(void)
|
||||
|
|
@ -405,4 +628,8 @@ void test_wal_job_fail_fact(void)
|
|||
RUN_TEST(test_wjfl_apply_committed_entry_backward_compat_old_payload);
|
||||
RUN_TEST(test_wjfl_apply_wal_entry_sets_state_and_exit_info);
|
||||
RUN_TEST(test_wjfl_real_on_job_fail_survives_restart);
|
||||
RUN_TEST(test_wjfl_apply_committed_entry_adopts_wire_finish_time);
|
||||
RUN_TEST(test_wjfl_apply_committed_entry_code_only_payload_compat);
|
||||
RUN_TEST(test_wjfl_apply_wal_entry_adopts_wire_finish_time);
|
||||
RUN_TEST(test_wjfl_record_fail_fact_at_survives_restart);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -2838,7 +2838,9 @@ static void test_full_rejoin_reconciliation_round_trip(void)
|
|||
.jobid = TWR_JOBID_RT_FINFAIL,
|
||||
.status = JobStateStatus_finishedFail,
|
||||
.exit_code = worker_reported_exit_code,
|
||||
.finish_time_ms = 0, .queried = 1
|
||||
.finish_time_ms = (worker_reported_finish_time_s - 60)
|
||||
* 1000ULL,
|
||||
.queried = 1
|
||||
};
|
||||
e_unrec = (xce_job_state_entry_t){
|
||||
.jobid = TWR_JOBID_RT_UNREC,
|
||||
|
|
@ -3013,10 +3015,13 @@ static void test_full_rejoin_reconciliation_round_trip(void)
|
|||
TEST_ASSERT_EQUAL_UINT32_MESSAGE(JOB_STAT_EXIT, j2_finfail->job.state,
|
||||
"finished-failure: the DEFERRED on_job_fail/record_job_fail_"
|
||||
"fact fact must apply to EXIT on replay (Req 4.1)");
|
||||
/* Req 4.2 remains DOCUMENTED-unsatisfiable for the finished-
|
||||
* FAILURE finish_time specifically (bug-1747) -- not asserted here.
|
||||
* The worker-reported EXIT CODE, however, is durably recorded via
|
||||
* WAL_JOB_FAIL's code trailer since bug-1754. */
|
||||
/* Req 4.2 for the finished-FAILURE outcome: the worker-reported
|
||||
* exit code (bug-1754) and failure time (bug-1747) both ride
|
||||
* WAL_JOB_FAIL's trailer and survive restart replay. */
|
||||
TEST_ASSERT_EQUAL_UINT64_MESSAGE(worker_reported_finish_time_s - 60,
|
||||
(uint64_t)j2_finfail->job.finish_time,
|
||||
"bug-1747: the recovered finish_time of a finished-failure must "
|
||||
"be the WORKER-reported time, not replay-time time(NULL)");
|
||||
TEST_ASSERT_NOT_NULL_MESSAGE(j2_finfail->job.exit_info,
|
||||
"bug-1754: a finished-failure's worker-reported exit code must "
|
||||
"survive restart replay in job.exit_info");
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue