diff --git a/.kiro/specs/job-array-single-wal-record/tasks.md b/.kiro/specs/job-array-single-wal-record/tasks.md index b211dcaf..7dcc6550 100644 --- a/.kiro/specs/job-array-single-wal-record/tasks.md +++ b/.kiro/specs/job-array-single-wal-record/tasks.md @@ -41,7 +41,7 @@ - _Boundary: WAL Array Submission_ - _Depends: 1.1_ -- [ ] 2.3 Unit tests for the array append/commit path +- [x] 2.3 Unit tests for the array append/commit path - Verify exactly one WAL entry and one commit continuation are produced for an array submission, for several array sizes - Verify the encoded payload's size does not grow with array size (stays constant regardless of element count) - Verify a single-job (array size 1) submission produces a byte-identical WAL record, commit-barrier count, and reply compared to the pre-feature baseline diff --git a/src/controller/testsuite/test_commit_barrier_rollout.c b/src/controller/testsuite/test_commit_barrier_rollout.c index 8d1281ff..01b7aa3a 100644 --- a/src/controller/testsuite/test_commit_barrier_rollout.c +++ b/src/controller/testsuite/test_commit_barrier_rollout.c @@ -110,6 +110,7 @@ #include "xce/types.h" /* job_t */ #include "commit.h" +#include "config.h" /* MAX_ARRAY_IDX (task 2.3 bounded-record test) */ #include "config_repl.h" /* push_config_to_standbys, handle_upgrade_config * (+ netio.h, request.h, myconf.h, clusterconfig.h) */ #include "controller.capnp.h" /* OPCode_* (task 4.1 gate cases) */ @@ -2082,6 +2083,236 @@ static void test_job_array_submit_rejects_array_size_one_or_less(void) printf("test_job_array_submit_rejects_array_size_one_or_less: PASS\n"); } +/* ── One entry / one continuation holds across several array sizes + * (task 2.3, Req 1.1-1.3) ──────────────────────────────────────────────── + * test_job_array_submit_appends_exactly_one_wal_entry_and_one_continuation + * above pins the full one-append/one-continuation observable at exactly + * one size (5). This sweep repeats that SAME observable (append succeeds, + * current_lsn advances by exactly one, the continuation does not resolve + * before a covering ACK, and resolves exactly once after one) at several + * materially different sizes in a single test, proving the property is not + * size-specific. */ +static void test_job_array_submit_one_entry_one_continuation_across_sizes(void) +{ + static const uint64_t sizes[] = { 2, 3, 100, 1000 }; + size_t i; + + jap_setup(2000); + jap_arrange_base_spec(); + + for (i = 0; i < sizeof(sizes) / sizeof(sizes[0]); i++) { + const uint64_t array_size = sizes[i]; + const uint64_t base_jobid = 30000 + (uint64_t)i * 100000; + uint64_t lsn_before, out_lsn = 0; + int rc; + char msg[128]; + + jap_notify_reset(); + lsn_before = g_jap.current_lsn; + + rc = persist_submit_job_array_async(&g_jap, base_jobid, array_size, + &g_jap_spec, &out_lsn, + jap_notify_fn, &g_jap_obs); + + snprintf(msg, sizeof(msg), + "array_size=%llu: append must succeed", + (unsigned long long)array_size); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, msg); + + snprintf(msg, sizeof(msg), + "array_size=%llu: exactly ONE WAL entry must be appended, " + "not one per element", + (unsigned long long)array_size); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_before + 1, g_jap.current_lsn, + msg); + + snprintf(msg, sizeof(msg), + "array_size=%llu: continuation must not resolve before a " + "covering ACK", (unsigned long long)array_size); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_jap_obs.call_count, msg); + + TEST_ASSERT_EQUAL_INT(0, persist_flush(&g_jap)); + repl_on_ack(&g_jap.repl, &g_jaep_a, out_lsn); + + snprintf(msg, sizeof(msg), + "array_size=%llu: exactly ONE commit continuation must " + "resolve for the whole array, not one per element", + (unsigned long long)array_size); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_jap_obs.call_count, msg); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_jap_obs.last_rc, msg); + } + + jap_teardown(); + printf("test_job_array_submit_one_entry_one_continuation_across_sizes: " + "PASS\n"); +} + +/* ── Payload size stays bounded at the actual upstream array-size ceiling + * (task 2.3, Req 7.1, 7.2) ─────────────────────────────────────────────── + * test_job_array_submit_payload_size_independent_of_array_size above + * compares an arbitrary "large" sample (10000) against a small one. This + * test additionally exercises the REAL structural ceiling design.md relies + * on to satisfy Requirement 7 without a chunking policy: idgen() rejects + * arraySize > MAX_ARRAY_IDX + 1, so MAX_ARRAY_IDX + 1 (32768, config.h) is + * the largest array_size that can ever legally reach this function in + * production. Confirms the encoded payload at that exact ceiling is still + * identical in size to a tiny array's payload AND stays well under a sane + * absolute byte bound -- Requirement 7.1's "no single record exceeds a + * bounded size" is an absolute claim, not only a relative one. */ +static void test_job_array_submit_payload_size_bounded_at_max_array_idx(void) +{ + const uint64_t ceiling_array_size = MAX_ARRAY_IDX + 1; /* 32768 */ + wal_entry_hdr_t hdr_small, hdr_ceiling; + uint64_t out_lsn = 0; + int rc; + off_t off_small, off_ceiling; + + jap_setup(2000); + jap_arrange_base_spec(); + jap_notify_reset(); + + off_small = (off_t)g_jap.seg_offset; + rc = persist_submit_job_array_async(&g_jap, 40000, 2, &g_jap_spec, + &out_lsn, jap_notify_fn, &g_jap_obs); + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_EQUAL_INT(0, + jap_read_wal_hdr_at(&g_jap, off_small, &hdr_small)); + + jap_notify_reset(); + off_ceiling = (off_t)g_jap.seg_offset; + rc = persist_submit_job_array_async(&g_jap, 50000, ceiling_array_size, + &g_jap_spec, &out_lsn, + jap_notify_fn, &g_jap_obs); + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_EQUAL_INT(0, + jap_read_wal_hdr_at(&g_jap, off_ceiling, &hdr_ceiling)); + + TEST_ASSERT_EQUAL_UINT32_MESSAGE(hdr_small.payload_len, + hdr_ceiling.payload_len, + "the payload size at the actual upstream array_size ceiling " + "(MAX_ARRAY_IDX + 1 = 32768) must be identical to a tiny array's " + "payload size -- proving Requirement 7's bound holds at the real " + "cap, not just at an arbitrary large sample"); + TEST_ASSERT_TRUE_MESSAGE(hdr_ceiling.payload_len < 4096, + "the array record's encoded payload must stay well under a sane " + "absolute bound even at the maximum legal array_size (Req 7.1)"); + + /* Cleanup: resolve both pending continuations before teardown. */ + TEST_ASSERT_EQUAL_INT(0, persist_flush(&g_jap)); + repl_on_ack(&g_jap.repl, &g_jaep_a, hdr_ceiling.lsn); + + jap_teardown(); + printf("test_job_array_submit_payload_size_bounded_at_max_array_idx: " + "PASS\n"); +} + +/* ── Non-array (array_size == 1) baseline: unmodified WAL_JOB_SUBMIT shape + * (task 2.3, Req 8.1) ──────────────────────────────────────────────────── + * Drives persist_submit_job_payloads_async() -- the pre-existing, + * completely UNMODIFIED single-job append function job.c's array_size == 1 + * path still calls -- directly with exactly the one-payload xlist_t shape + * build_job_submit_payloads() constructs for a non-array submission. + * Confirms the concrete regression sentinel this feature must never + * disturb: the on-disk WAL entry type stays WAL_JOB_SUBMIT (never + * WAL_JOB_ARRAY_SUBMIT), the payload decodes via struct Job/xce_job_t + * (never struct JobSubmit/xce_job_submit_t), exactly one entry is + * appended, and exactly one commit continuation resolves -- the same four + * properties a pre-feature build already guaranteed, now pinned so a + * future change cannot silently route array_size == 1 into the + * array-shaped path or otherwise perturb the single-job wire format. */ +static void test_job_submit_single_payload_wal_type_matches_pre_feature_baseline( + void) +{ + const uint64_t jobid = 60000; + time_t now = time(NULL); + xce_job_t *jp; + xlist_t plist; + uint64_t lsns[1] = { 0 }; + uint64_t lsn_before; + off_t entry_off; + wal_entry_hdr_t hdr; + uint8_t *payload = NULL; + int rc; + + jap_setup(2000); + jap_arrange_base_spec(); + jap_notify_reset(); + + /* One payload -- the exact shape build_job_submit_payloads() (job.c) + * produces for a single-element (non-array) submission. */ + TEST_ASSERT_EQUAL_INT(0, xlist_init(&plist, 1)); + jp = calloc(1, sizeof(*jp)); + TEST_ASSERT_NOT_NULL(jp); + jp->submit = &g_jap_submit; + jp->jobid = jobid; + jp->sub_time = now; + jp->queue_time = now; + jp->state = JOB_STAT_PEND; + TEST_ASSERT_TRUE(xlist_store(&plist, jp) >= 0); + + lsn_before = g_jap.current_lsn; + entry_off = (off_t)g_jap.seg_offset; + + rc = persist_submit_job_payloads_async(&g_jap, &plist, lsns, + jap_notify_fn, &g_jap_obs); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, + "the unmodified single-job append path must still succeed"); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn_before + 1, g_jap.current_lsn, + "a single (non-array) submission must append exactly one WAL " + "entry, same as before this feature"); + + TEST_ASSERT_EQUAL_INT(0, + jap_read_wal_entry_at(&g_jap, entry_off, &hdr, &payload)); + TEST_ASSERT_EQUAL_UINT32_MESSAGE((uint32_t)WAL_JOB_SUBMIT, hdr.type, + "a non-array submission's WAL entry must still be typed " + "WAL_JOB_SUBMIT, never WAL_JOB_ARRAY_SUBMIT (Req 8.1)"); + + { + struct capn cctx; + TEST_ASSERT_EQUAL_INT(0, + capn_init_mem(&cctx, payload, hdr.payload_len, 0)); + + Job_ptr jobp; + jobp.p = capn_getp(capn_root(&cctx), 0, 1); + TEST_ASSERT_TRUE(jobp.p.type != CAPN_NULL); + + struct Job wire; + memset(&wire, 0, sizeof(wire)); + read_Job(&wire, jobp); + + xce_job_t decoded; + memset(&decoded, 0, sizeof(decoded)); + decode_Job(&decoded, &wire); + + TEST_ASSERT_EQUAL_UINT64_MESSAGE(jobid, decoded.jobid, + "the decoded jobid must round-trip through the unmodified " + "struct Job/xce_job_t wire shape, never struct JobSubmit"); + TEST_ASSERT_NOT_NULL(decoded.submit); + TEST_ASSERT_EQUAL_STRING("normal", decoded.submit->queue); + TEST_ASSERT_EQUAL_STRING("tester", decoded.submit->user); + + free_Job(&decoded); + capn_free(&cctx); + } + free(payload); + + TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_jap_obs.call_count, + "the continuation must not resolve before a covering ACK"); + TEST_ASSERT_EQUAL_INT(0, persist_flush(&g_jap)); + repl_on_ack(&g_jap.repl, &g_jaep_a, lsns[0]); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_jap_obs.call_count, + "exactly ONE commit continuation must resolve for the single " + "payload, same contract as before this feature"); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_jap_obs.last_rc, 0); + + xlist_fini(&plist); + free(jp); + + jap_teardown(); + printf("test_job_submit_single_payload_wal_type_matches_pre_feature_" + "baseline: PASS\n"); +} + /* ── Test group entry point ──────────────────────────────────────────────── */ void test_commit_barrier_rollout(void) @@ -2130,5 +2361,10 @@ void test_commit_barrier_rollout(void) RUN_TEST(test_job_array_submit_appends_exactly_one_wal_entry_and_one_continuation); RUN_TEST(test_job_array_submit_payload_size_independent_of_array_size); RUN_TEST(test_job_array_submit_rejects_array_size_one_or_less); + /* job-array-single-wal-record task 2.3 — unit tests for the array + * append/commit path (Req 1.1-1.3, 7.1, 7.2, 8.1) */ + RUN_TEST(test_job_array_submit_one_entry_one_continuation_across_sizes); + RUN_TEST(test_job_array_submit_payload_size_bounded_at_max_array_idx); + RUN_TEST(test_job_submit_single_payload_wal_type_matches_pre_feature_baseline); UNITY_END(); }