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

feat(scheduler-sim-decision-parity): report reconciled decisions through the existing channel (task 2.4)

Browse source 
Feeds a reconciled job's decision (task 2.3) through the exact same
recording/print/REPL-check calls the existing decision-drain loop uses for
a freshly-drained decision, via make_dispatch_decision() -- the same
DISPATCH-construction idiom the live allocator path already uses.
Dedup (Requirement 3.3) needs no new mechanism: the job->accounts gate
from tasks 2.2/2.3 already prevents double-reporting, now proven at the
reporting layer too (zero recorded decisions / no console output for an
already-accounted job).

Required extending sim_reconcile_unaccounted_jobs()'s signature with a
third parameter, a nullable recording-context handle -- design.md's
Service Interface updated to match, with the confirmed reasoning: the
handle only exists as a stack-local inside the round loop, and task 3's
still-unwired call site sits in that same scope, so this is necessary
plumbing, not scope creep.

46/47 scheduler tests pass; the 1 failure is pre-existing bug-1560
(unrelated).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Rongsong Shen 2026-08-22 09:07:24 +00:00
parent eccba84112
commit 0c237f601a
No known key found for this signature in database
GPG key ID: 5BCDAAE0123E9FE2
4 changed files with 278 additions and 61 deletions
Download patch file Download diff file Expand all files Collapse all files

223
src/scheduler/testsuite/test_sim_reconcile_unaccounted.c
View file

@ -1,21 +1,21 @@
/*
* test_sim_reconcile_unaccounted.c - contract test for
* sim_reconcile_unaccounted_jobs() (sim_loop.c, scheduler-sim-decision-
* parity spec, task 2.2, "Recorded-Decision Reconciliation" component,
* design.md; Requirement 4.1, 4.2).
* parity spec, tasks 2.2-2.4, "Recorded-Decision Reconciliation" component,
* design.md; Requirement 1.1, 1.2, 3.1, 3.2, 3.3, 4.1, 4.2).
*
* Scope (task 2.2 only, per design.md's Implementation Notes and this
* task's own text): the once-per-round trigger (RUN-state job with
* job->accounts == NULL), the jobid-based search through this round's
* decoded_session->decisions, and the "not found" path -- leave the job
* Scope: the once-per-round trigger (RUN-state job with job->accounts ==
* NULL), the jobid-based search through this round's
* decoded_session->decisions, the "not found" path -- leave the job
* unaccounted, report usage unavailable (== job->accounts staying NULL,
* which sched_account_jobs() already tolerates safely), never treat this as
* an error, and stay safe to re-check next round. Allocation reconstruction
* for a FOUND match (ownership transfer, array compaction, decision
* reporting) is deliberately NOT exercised here -- that is tasks 2.3/2.4's
* own test coverage; this file's "found" test (see below) only asserts
* that a match does not corrupt an already-accounted job, not that a match
* gets reconstructed.
* an error, and stay safe to re-check next round -- AND (task 2.3/2.4) a
* FOUND match's allocation reconstruction (ownership transfer, array
* compaction) plus feeding the reconstructed decision through the same
* three-call reporting channel (sched_record_session_add_decision(),
* sim_print_decision(), xce_repl_check_decision()) a freshly-drained
* decision uses, and that an already-accounted job produces NO duplicate
* report through this mechanism (Requirement 3.3).
*
* sim_reconcile_unaccounted_jobs() is `static` and not yet called from
* anywhere in production code (task 3 wires it into
@ -48,6 +48,7 @@
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <unity/unity.h>
@ -329,7 +330,7 @@ test_reconcile_leaves_run_job_unaccounted_when_no_recorded_decision(void)
* jobid at all -- the replay's own window never captured it. */
decoded = make_decoded_session(1000ULL, NULL);
sim_reconcile_unaccounted_jobs(&sim, decoded);
sim_reconcile_unaccounted_jobs(&sim, decoded, NULL);
TEST_ASSERT_NULL_MESSAGE(
job->accounts,
@ -360,14 +361,14 @@ static void test_reconcile_is_safe_to_recheck_on_a_later_round(void)
job = add_job(&(sim.core), 7002ULL, JOB_STAT_RUN, NULL);
decoded_round1 = make_decoded_session(1000ULL, NULL);
sim_reconcile_unaccounted_jobs(&sim, decoded_round1);
sim_reconcile_unaccounted_jobs(&sim, decoded_round1, NULL);
TEST_ASSERT_NULL(job->accounts);
/* Round 2: still no matching recorded decision -- calling again must
* not error, crash, or otherwise behave differently just because this
* is a repeat check of the same job. */
decoded_round2 = make_decoded_session(2000ULL, NULL);
sim_reconcile_unaccounted_jobs(&sim, decoded_round2);
sim_reconcile_unaccounted_jobs(&sim, decoded_round2, NULL);
TEST_ASSERT_NULL_MESSAGE(
job->accounts,
@ -383,16 +384,24 @@ static void test_reconcile_is_safe_to_recheck_on_a_later_round(void)
/* earlier or freshly dispatched by the live path) is never re-touched -- */
/* the trigger condition (job->accounts == NULL) must gate correctly even */
/* when a matching recorded DISPATCH entry for the same jobid IS present */
/* this round (task 2.2 does not implement match-handling yet, but the */
/* trigger itself must still never fire for an already-accounted job). */
/* this round. Extended (task 2.4) to also prove Requirement 3.3's "report */
/* only the fresh one" guarantee at the decision-reporting layer, not just */
/* at the job->accounts layer: neither the recording channel nor the */
/* step/continue console output may see a duplicate report for this job. */
/* ---------------------------------------------------------------------- */
static void test_reconcile_skips_job_that_already_has_accounts(void)
{
xce_sched_sim_driver_t sim;
xce_sched_job_t* job;
xce_job_account_t* accounts_before;
replay_session_t* decoded;
xce_sched_sim_driver_t sim;
xce_sched_job_t* job;
xce_job_account_t* accounts_before;
replay_session_t* decoded;
sched_record_session_ctx_t* record_ctx;
char tmp[] = "/tmp/xce_simreconcile_dedup_XXXXXX";
int fd = mkstemp(tmp);
int saved_stdout;
off_t len;
char* captured;
fixture_init(&sim);
@ -403,8 +412,36 @@ static void test_reconcile_skips_job_that_already_has_accounts(void)
accounts_before = job->accounts;
decoded = make_decoded_session(1000ULL, make_dispatch_entry(7003ULL));
record_ctx = sched_record_session_ctx_new(1000ULL);
TEST_ASSERT_NOT_NULL(record_ctx);
sim_reconcile_unaccounted_jobs(&sim, decoded);
/* Same dup()/dup2() stdout-capture technique test_sim_verify.c's own
* capture_stdout_around()/run_sim_and_capture_stdout() establish,
* inlined here (this call takes more than the zero args that helper's
* function-pointer shape allows). */
TEST_ASSERT_TRUE_MESSAGE(fd >= 0, "mkstemp for stdout capture must "
"succeed");
fflush(stdout);
saved_stdout = dup(STDOUT_FILENO);
TEST_ASSERT_TRUE(saved_stdout >= 0);
TEST_ASSERT_TRUE(dup2(fd, STDOUT_FILENO) >= 0);
sim_reconcile_unaccounted_jobs(&sim, decoded, record_ctx);
fflush(stdout);
dup2(saved_stdout, STDOUT_FILENO);
close(saved_stdout);
len = lseek(fd, 0, SEEK_CUR);
TEST_ASSERT_TRUE(len >= 0);
captured = (char*)calloc(1, (size_t)len + 1);
TEST_ASSERT_NOT_NULL(captured);
TEST_ASSERT_TRUE(lseek(fd, 0, SEEK_SET) == 0);
if (len > 0) {
ssize_t nread = read(fd, captured, (size_t)len);
TEST_ASSERT_TRUE(nread == len);
}
close(fd);
TEST_ASSERT_EQUAL_PTR_MESSAGE(
accounts_before, job->accounts,
@ -413,9 +450,20 @@ static void test_reconcile_skips_job_that_already_has_accounts(void)
"double-application guard)");
TEST_ASSERT_EQUAL_INT_MESSAGE(
1, decoded->n_decisions,
"task 2.2 does not consume/compact recorded entries -- that is "
"task 2.3's job");
"an already-accounted job's matching entry must not be consumed/"
"compacted either -- the trigger never fires for it at all");
TEST_ASSERT_EQUAL_INT_MESSAGE(
0, record_ctx->n,
"Requirement 3.3: an already-accounted job (fresh live decision or "
"earlier reconciliation) must produce NO recorded-decision report "
"from this mechanism -- reporting only the fresh one, never a "
"duplicate");
TEST_ASSERT_NULL_MESSAGE(
strstr(captured, "[sim] DISPATCH"),
"Requirement 3.3: an already-accounted job must produce NO "
"step/continue console output from this mechanism either");
free(captured);
printf("test_reconcile_skips_job_that_already_has_accounts: PASS\n");
}
@ -438,7 +486,7 @@ static void test_reconcile_ignores_non_dispatch_entry_with_matching_jobid(
decoded = make_decoded_session(1000ULL, make_pend_entry(7004ULL));
sim_reconcile_unaccounted_jobs(&sim, decoded);
sim_reconcile_unaccounted_jobs(&sim, decoded, NULL);
TEST_ASSERT_NULL_MESSAGE(
job->accounts,
@ -472,7 +520,7 @@ static void test_reconcile_skips_non_run_state_job(void)
* job before any lookup happens. */
decoded = make_decoded_session(1000ULL, make_dispatch_entry(7005ULL));
sim_reconcile_unaccounted_jobs(&sim, decoded);
sim_reconcile_unaccounted_jobs(&sim, decoded, NULL);
TEST_ASSERT_NULL_MESSAGE(
job->accounts,
@ -507,7 +555,7 @@ test_reconcile_reconstructs_allocs_and_accounts_from_matching_dispatch(void)
decoded
= make_decoded_session(1000ULL, make_dispatch_entry_from_alloc(jalloc));
sim_reconcile_unaccounted_jobs(&sim, decoded);
sim_reconcile_unaccounted_jobs(&sim, decoded, NULL);
TEST_ASSERT_EQUAL_PTR_MESSAGE(
jalloc, job->allocs,
@ -563,7 +611,7 @@ static void test_reconcile_compacts_via_shift_down_preserving_order(void)
decoded->decisions[3] = make_pend_entry(9003ULL); /* C */
decoded->n_decisions = 4;
sim_reconcile_unaccounted_jobs(&sim, decoded);
sim_reconcile_unaccounted_jobs(&sim, decoded, NULL);
TEST_ASSERT_NOT_NULL(job->accounts);
TEST_ASSERT_EQUAL_INT_MESSAGE(
@ -616,7 +664,7 @@ static void test_reconcile_then_release_via_job_state_change_is_asan_clean(void)
1000ULL, make_dispatch_entry_from_alloc(make_test_job_alloc_with_host(
7012ULL, "worker1", "slots", 4.0f)));
sim_reconcile_unaccounted_jobs(&sim, decoded);
sim_reconcile_unaccounted_jobs(&sim, decoded, NULL);
TEST_ASSERT_NOT_NULL(job->accounts);
TEST_ASSERT_NOT_NULL(job->allocs);
@ -651,6 +699,115 @@ static void test_reconcile_then_release_via_job_state_change_is_asan_clean(void)
"PASS\n");
}
/* ---------------------------------------------------------------------- */
/* Test 10 (task 2.4, the task's own PRIMARY Observable): a matching */
/* recorded DISPATCH entry, once reconstructed (task 2.3), is fed through */
/* the exact same three-call decision-reporting channel a freshly-drained */
/* decision uses -- sched_record_session_add_decision() (proven via a real */
/* record_ctx's own accumulated decisions[]/n/n_dispatched fields, exactly */
/* the same struct sched_sim_driver_run()'s drain loop populates) and */
/* sim_print_decision() (proven via captured stdout -- the literal */
/* mechanism by which a decision "shows... in step/continue output", per */
/* this task's own Observable Completion text). xce_repl_check_decision() */
/* is also called (not independently observable here without a live REPL */
/* breakpoint fixture -- repl.c/repl_break.c/repl_info.c are unchanged and */
/* out of this task's boundary per design.md) but must not crash, which */
/* this test's mere completion already proves for both sim->interactive */
/* on and off. Preserves the original decision's type (DISPATCH) and */
/* resource-allocation detail (Requirement 3.2): the SAME job->allocs */
/* pointer the reconciliation step already reconstructed is what reaches */
/* both the recording channel and the console output. */
/* ---------------------------------------------------------------------- */
static void
test_reconcile_reports_reconstructed_decision_through_existing_channel(void)
{
xce_sched_sim_driver_t sim;
xce_sched_job_t* job;
xce_job_alloc_t* jalloc;
replay_session_t* decoded;
sched_record_session_ctx_t* record_ctx;
char tmp[] = "/tmp/xce_simreconcile_report_XXXXXX";
int fd = mkstemp(tmp);
int saved_stdout;
off_t len;
char* captured;
fixture_init(&sim);
job = add_job(&(sim.core), 7020ULL, JOB_STAT_RUN, NULL);
TEST_ASSERT_NULL(job->accounts);
jalloc = make_test_job_alloc_with_host(7020ULL, "worker9", "slots", 3.0f);
decoded
= make_decoded_session(1000ULL, make_dispatch_entry_from_alloc(jalloc));
record_ctx = sched_record_session_ctx_new(1000ULL);
TEST_ASSERT_NOT_NULL(record_ctx);
/* Exercise the REPL hook too (sim->interactive on) -- must remain a
* safe no-op with no repl_state allocated yet (repl.h's own "never
* load-bearing" precondition), proving xce_repl_check_decision() is
* actually reached and does not crash regardless of interactive mode. */
sim.interactive = true;
TEST_ASSERT_TRUE_MESSAGE(fd >= 0, "mkstemp for stdout capture must "
"succeed");
fflush(stdout);
saved_stdout = dup(STDOUT_FILENO);
TEST_ASSERT_TRUE(saved_stdout >= 0);
TEST_ASSERT_TRUE(dup2(fd, STDOUT_FILENO) >= 0);
sim_reconcile_unaccounted_jobs(&sim, decoded, record_ctx);
fflush(stdout);
dup2(saved_stdout, STDOUT_FILENO);
close(saved_stdout);
len = lseek(fd, 0, SEEK_CUR);
TEST_ASSERT_TRUE(len >= 0);
captured = (char*)calloc(1, (size_t)len + 1);
TEST_ASSERT_NOT_NULL(captured);
TEST_ASSERT_TRUE(lseek(fd, 0, SEEK_SET) == 0);
if (len > 0) {
ssize_t nread = read(fd, captured, (size_t)len);
TEST_ASSERT_TRUE(nread == len);
}
close(fd);
/* The reconciliation itself (task 2.3) still happened. */
TEST_ASSERT_EQUAL_PTR(jalloc, job->allocs);
TEST_ASSERT_NOT_NULL(job->accounts);
/* sched_record_session_add_decision() reporting (Requirement 3.1). */
TEST_ASSERT_EQUAL_INT_MESSAGE(
1, record_ctx->n,
"the reconstructed DISPATCH decision must be reported through the "
"recording channel exactly once");
TEST_ASSERT_EQUAL_INT_MESSAGE(1, record_ctx->n_dispatched,
"must be counted as a DISPATCH");
TEST_ASSERT_EQUAL_INT_MESSAGE(
ReplayDecision_body_dispatch, record_ctx->decisions[0]->body_which,
"must be recorded with the DISPATCH body shape");
TEST_ASSERT_EQUAL_PTR_MESSAGE(
jalloc, record_ctx->decisions[0]->body.dispatch->alloc,
"Requirement 3.2: the recorded entry must carry the SAME "
"resource-allocation detail (identical pointer, borrowed -- same "
"convention recording_session.c's own header comment documents "
"for a freshly-drained decision) the reconciliation step "
"reconstructed, not a copy or a different value");
/* sim_print_decision() console output (Requirement 3.1 -- this IS the
* "shows... in step/continue output" mechanism). */
TEST_ASSERT_NOT_NULL_MESSAGE(
strstr(captured, "[sim] DISPATCH jobid="),
"the reconstructed decision must be printed through the same "
"step/continue console output a freshly-computed decision uses");
free(captured);
printf("test_reconcile_reports_reconstructed_decision_through_existing_"
"channel: PASS\n");
}
/* ---------------------------------------------------------------------- */
/* Test 6: NULL `sim`/`decoded_session` is a safe no-op (design.md's */
/* Service Interface preconditions -- "never load-bearing" convention). */
@ -664,9 +821,9 @@ static void test_reconcile_null_args_are_a_safe_noop(void)
fixture_init(&sim);
decoded = make_decoded_session(1000ULL, NULL);
sim_reconcile_unaccounted_jobs(NULL, decoded);
sim_reconcile_unaccounted_jobs(&sim, NULL);
sim_reconcile_unaccounted_jobs(NULL, NULL);
sim_reconcile_unaccounted_jobs(NULL, decoded, NULL);
sim_reconcile_unaccounted_jobs(&sim, NULL, NULL);
sim_reconcile_unaccounted_jobs(NULL, NULL, NULL);
/* Reaching this line without a crash IS the assertion. */
TEST_ASSERT_TRUE(1);
@ -687,6 +844,8 @@ int main(void)
test_reconcile_reconstructs_allocs_and_accounts_from_matching_dispatch);
RUN_TEST(test_reconcile_compacts_via_shift_down_preserving_order);
RUN_TEST(test_reconcile_then_release_via_job_state_change_is_asan_clean);
RUN_TEST(
test_reconcile_reports_reconstructed_decision_through_existing_channel);
RUN_TEST(test_reconcile_null_args_are_a_safe_noop);
return UNITY_END();
}