wdgzyjx/myxce
1
0
Fork 0
forked from lei.chen/myxce
Code Pull requests Activity

test(job-array-single-wal-record): close remaining array append/commit gaps

Browse source 
Adds 3 targeted tests after confirming tasks 2.1/2.2 already cover
most of this task's observable criteria:
- one-entry/one-continuation swept across sizes {2,3,100,1000}
  (existing coverage only tested size 5)
- payload-size boundedness at the real idgen()-enforced ceiling
  (MAX_ARRAY_IDX+1 = 32768), plus an absolute byte-cap assertion
  (existing coverage only compared 2 vs 10000)
- array_size==1 baseline: explicit WAL-byte readback proving the
  entry type is WAL_JOB_SUBMIT (not WAL_JOB_ARRAY_SUBMIT) and the
  payload decodes via struct Job, not struct JobSubmit (existing
  coverage only proved "old tests still pass" implicitly)

No production code changed -- test-only.

Task 2.3 of job-array-single-wal-record.
This commit is contained in:
Rongsong Shen 2026-07-14 10:53:20 +00:00
parent d38ca72e76
commit 78eb983a74
2 changed files with 237 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

2
.kiro/specs/job-array-single-wal-record/tasks.md
View file

@ -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

236
src/controller/testsuite/test_commit_barrier_rollout.c
View file

@ -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();
}