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

fix(job-array-single-wal-record): close bug-1443's bulk-insert partial-visibility gap

Browse source 
restore_job_array()'s bulk-insert pass could fail partway through
(key_add() on realloc growth exhaustion), leaving an already-inserted
prefix permanently visible in c->jobmgr but excluded from
out_new_jobs -- so those elements existed and were individually
addressable but never reached HOOK_DETAIL_JOB_NEW, never scheduled or
pushed, not on this attempt or any later successful retry. A literal
violation of Requirements 2.3 and 9.4, tracked as bug-1443.

Adds kvmgr_reserve()/xzarray_reserve(), new purely-additive primitives
in shared kvmgr.c/zarray.c (following the existing kvmgr_capture_keys()
template -- bare extern function, no RTTI vtable slot, caller-holds-
write-lock contract), that pre-grow c->jobmgr's backing store before
the bulk-insert pass begins, closing the dominant N-scaling failure
trigger structurally.

Found a genuine off-by-one in the existing, unmodified
xzarray_ensure_items(): its strict less-than sufficiency check means
reserving exactly array_size and then performing array_size real
stores still triggers one more growth cycle on the last store.
Reserving array_size + 1 is the exact minimal correct compensation,
independently re-derived and confirmed by review with concrete
arithmetic.

For the now-much-narrower residual case (kvtab_add()'s own small
per-entry allocation, or a content collision -- both non-N-scaling,
honestly documented as still theoretically possible): restore_job_array()
now populates out_new_jobs with whatever prefix was genuinely inserted
before a failure, and apply_committed_entry() fires HOOK_DETAIL_JOB_NEW
for that non-empty partial prefix even when the overall return is
negative, while still propagating the negative rc so the LSN correctly
halts and the Live Apply Resource Fence still engages for the
remainder. A genuinely-inserted element is never left un-pushed again.

Task 7.1 of job-array-single-wal-record (Requirements 2.3, 9.4, 9.6).
This commit is contained in:
Rongsong Shen 2026-07-16 15:08:02 +00:00
parent 575f928bbc
commit dcf9f39c22
13 changed files with 910 additions and 36 deletions
Download patch file Download diff file Expand all files Collapse all files

24
src/lib/kvmgr.c
View file

@ -410,3 +410,27 @@ int kvmgr_capture_keys(var kvmgr_var,
*n_keys_out = ctx.filled;
return 0;
}
/* ------------------------------------------------------------------ */
/* kvmgr_reserve — job-array-single-wal-record task 7.1 (bug-1443) */
/* ------------------------------------------------------------------ */
int kvmgr_reserve(var kvmgr_var, int n)
{
if ((kvmgr_var == NULL) || (n < 0)) {
return -EINVAL;
}
/* Same public-pointer -> meta_obj_t -> xkvmgr_t unwrap as
* kvmgr_capture_keys() above. */
meta_obj_t* obj = RTTI_DATA_TO_OBJ(kvmgr_var);
xkvmgr_t* kvm = RTTI_OBJ_TO_DATA(xkvmgr_t, obj);
/* Pre-grow the value store ONLY -- this is the piece xkvmgr_add()
* mutates first (vstore(kvm->data, ...), see xkvmgr_add() above) and
* the only piece with an N-scaling realloc() reachable from a bulk
* key_add() loop. kvm->kvtab's own per-entry hash-chain-node allocation
* (kv.c) is deliberately NOT covered here -- see this function's doc
* comment (kvmgr.h) for why. */
return xzarray_reserve(kvm->data, n);
}

2
src/lib/meson.build
View file

@ -169,6 +169,8 @@ libxce_test_src = ['testsuite/test.c',
'testsuite/test_request.c',
'testsuite/test_zarray_cow.c',
'testsuite/test_kvmgr_capture.c',
'testsuite/test_zarray_reserve.c',
'testsuite/test_kvmgr_reserve.c',
'testsuite/test_hooks.c',
'testsuite/test_cluster.c',
'testsuite/test_cluster_parse.c',

4
src/lib/testsuite/test.c
View file

@ -32,6 +32,8 @@ extern void test_wobj(void);
extern void test_request(void);
extern void test_zarray_cow(void);
extern void test_kvmgr_capture(void);
extern void test_zarray_reserve(void);
extern void test_kvmgr_reserve(void);
extern void test_hooks(void);
extern void test_cluster(void);
extern void test_cluster_parse(void);
@ -147,6 +149,8 @@ int main(int argc, char* argv[])
TEST(test_request);
TEST(test_zarray_cow);
TEST(test_kvmgr_capture);
TEST(test_zarray_reserve);
TEST(test_kvmgr_reserve);
TEST(test_hooks);
TEST(test_cluster);
TEST(test_cluster_parse);

136
src/lib/testsuite/test_kvmgr_reserve.c Normal file
View file

@ -0,0 +1,136 @@
/* test_kvmgr_reserve.c — job-array-single-wal-record task 7.1 (bug-1443)
*
* Unit tests for kvmgr_reserve() (kvmgr.c/kvmgr.h) -- the thin, kvmgr-level
* forwarding wrapper around xzarray_reserve() (zarray.c, see
* test_zarray_reserve.c for the rigorous capacity-growth proof at that
* lower level). This file proves kvmgr_reserve() itself:
* - correctly unwraps an xkvmgr_type instance and forwards to its
* internal value store, allowing a subsequent bulk key_add() sequence
* up to the reserved count to succeed;
* - propagates a simulated allocation failure (via the SAME
* xzarray_reserve_test_set_force_fail() seam kvmgr_reserve() forwards
* into -- there is no separate kvmgr-level seam, deliberately, since
* kvmgr_reserve() has no logic of its own beyond the forward);
* - rejects a NULL kvmgr / negative n defensively.
*/
#include <errno.h>
#include <rtti.h>
#include <stdio.h>
#include <stdlib.h>
#include <unity/unity.h>
#include "kv.h"
#include "kvmgr.h"
#include "zarray.h"
static var make_test_kvmgr(uint32_t nslots, uint32_t delta)
{
kvmgr_params_t params = { .key_type = KVMGR_KEY_INT,
.nslots = nslots,
.delta = delta,
.datakiller = NULL };
return makeInstance(&xkvmgr_type, &params);
}
/* ── kvmgr_reserve() allows a subsequent bulk key_add() sequence up to the
* reserved count to succeed (functional forwarding proof) ──────────────── */
static void test_kvmgr_reserve_allows_subsequent_bulk_key_add(void)
{
const int n = 30;
/* Tiny initial capacity/delta -- without reservation this would force
* many incremental growth cycles across the loop below. */
var kvmgr = make_test_kvmgr(2, 1);
TEST_ASSERT_NOT_NULL(kvmgr);
lock(kvmgr, RTTI_LOCK_WRITE);
int reserve_rc = kvmgr_reserve(kvmgr, n + 1);
TEST_ASSERT_TRUE_MESSAGE(reserve_rc >= 0,
"kvmgr_reserve must succeed for a valid kvmgr and count");
for (int i = 0; i < n; i++) {
int idx = key_add(kvmgr, (void*)(long)(1000 + i), (void*)(long)i);
TEST_ASSERT_TRUE_MESSAGE(idx >= 0,
"every key_add within the reserved count must succeed");
}
unlock(kvmgr);
/* Spot-check a few entries resolve correctly. */
void* found = NULL;
TEST_ASSERT_TRUE(key_find(kvmgr, (void*)(long)1000, &found) >= 0);
TEST_ASSERT_EQUAL_INT(0, (int)(long)found);
TEST_ASSERT_TRUE(key_find(kvmgr, (void*)(long)1029, &found) >= 0);
TEST_ASSERT_EQUAL_INT(29, (int)(long)found);
destroy(kvmgr);
printf("test_kvmgr_reserve_allows_subsequent_bulk_key_add: PASS\n");
}
/* ── Simulated allocation failure propagates through kvmgr_reserve() too --
* reusing the SAME xzarray_reserve()-level seam, since kvmgr_reserve() is a
* pure forward with no logic of its own. ───────────────────────────────── */
static void test_kvmgr_reserve_propagates_simulated_allocation_failure(void)
{
var kvmgr = make_test_kvmgr(4, 4);
TEST_ASSERT_NOT_NULL(kvmgr);
xzarray_reserve_test_set_force_fail();
lock(kvmgr, RTTI_LOCK_WRITE);
int rc = kvmgr_reserve(kvmgr, 10);
unlock(kvmgr);
TEST_ASSERT_EQUAL_INT_MESSAGE(-ENOMEM, rc,
"a forced allocation failure must propagate as -ENOMEM through"
" kvmgr_reserve()");
xzarray_reserve_test_clear_force_fail();
lock(kvmgr, RTTI_LOCK_WRITE);
int rc2 = kvmgr_reserve(kvmgr, 10);
unlock(kvmgr);
TEST_ASSERT_TRUE_MESSAGE(rc2 >= 0,
"clearing the force-fail seam must restore normal success");
destroy(kvmgr);
printf("test_kvmgr_reserve_propagates_simulated_allocation_failure: "
"PASS\n");
}
/* ── Defensive / boundary ─────────────────────────────────────────────── */
static void test_kvmgr_reserve_rejects_null_kvmgr(void)
{
int rc = kvmgr_reserve(NULL, 5);
TEST_ASSERT_TRUE(rc < 0);
printf("test_kvmgr_reserve_rejects_null_kvmgr: PASS\n");
}
static void test_kvmgr_reserve_rejects_negative_n(void)
{
var kvmgr = make_test_kvmgr(4, 4);
TEST_ASSERT_NOT_NULL(kvmgr);
lock(kvmgr, RTTI_LOCK_WRITE);
int rc = kvmgr_reserve(kvmgr, -1);
unlock(kvmgr);
TEST_ASSERT_TRUE(rc < 0);
destroy(kvmgr);
printf("test_kvmgr_reserve_rejects_negative_n: PASS\n");
}
/* ── Test group entry point ───────────────────────────────────────────── */
void test_kvmgr_reserve(void)
{
printf("\n>> kvmgr_reserve test\n");
RUN_TEST(test_kvmgr_reserve_allows_subsequent_bulk_key_add);
RUN_TEST(test_kvmgr_reserve_propagates_simulated_allocation_failure);
RUN_TEST(test_kvmgr_reserve_rejects_null_kvmgr);
RUN_TEST(test_kvmgr_reserve_rejects_negative_n);
}

236
src/lib/testsuite/test_zarray_reserve.c Normal file
View file

@ -0,0 +1,236 @@
/* test_zarray_reserve.c — job-array-single-wal-record task 7.1 (bug-1443)
*
* Unit tests for xzarray_reserve() (zarray.c), the new capacity-reservation
* primitive kvmgr_reserve() (kvmgr.c/kvmgr.h) uses to pre-grow an
* xkvmgr_type instance's backing value store before a bulk-insert pass
* (e.g. restore_job_array()'s bulk-insert loop, job.c) -- closing the
* DOMINANT, N-scaling failure trigger for that pass (the large realloc()
* reachable through vstore -> xzarray_ensure_items).
*
* xzarray_t is fully opaque outside zarray.c, so these tests use the two
* new XCE_TESTING-guarded introspection/fault-injection primitives added
* alongside xzarray_reserve() itself:
* - xzarray_test_get_caps() -- read current backing capacity
* - xzarray_reserve_test_set/clear_force_fail() -- simulate an allocation
* failure without needing genuine OOM
*
* Covers:
* - xzarray_reserve() genuinely pre-grows capacity: reserving N+1 items
* up front means a subsequent sequence of N real vstore() calls causes
* NO further capacity growth at all (proven via xzarray_test_get_caps()
* staying constant across every store).
*
* The "+1" is not incidental -- it is required by xzarray_ensure_items()
* (zarray.c, unmodified)'s own strict `mincap < caps` sufficiency check:
* reserving EXACTLY N and then performing N stores still hits
* `mincap == caps` on the very last store, which that strict check
* treats as insufficient and grows again (by a small, delta-bounded
* amount). This is verified directly below (see
* test_xzarray_reserve_off_by_one_without_extra_headroom) rather than
* assumed.
* - Without any reservation, the identical N-store sequence DOES grow
* capacity partway through (the behavior task 7.1 improves on).
* - A simulated allocation failure is correctly propagated as -ENOMEM.
* - Defensive/boundary: NULL array, negative n.
*/
#include <errno.h>
#include <rtti.h>
#include <stdio.h>
#include <stdlib.h>
#include <unity/unity.h>
#include "zarray.h"
static var make_test_xzarray(int delta)
{
xzarray_params_t params = { .delta = delta, .datakiller = NULL };
return makeInstance(&xzarray_type, &params);
}
/* ── xzarray_reserve() genuinely pre-grows capacity (with the verified +1
* headroom) -- zero further growth during N subsequent real stores ──────── */
static void
test_xzarray_reserve_pregrows_capacity_no_growth_during_subsequent_stores(void)
{
const int delta = 1; /* small delta -- growth would otherwise be
* triggered on almost every store, making this the
* worst case (and the one bug-1443 cares about:
* arrays larger than the natural delta margin). */
const int n = 20;
var array = make_test_xzarray(delta);
TEST_ASSERT_NOT_NULL(array);
/* Reserve n+1 -- see the file doc comment for why +1 is required. */
int reserve_rc = xzarray_reserve(array, n + 1);
TEST_ASSERT_TRUE_MESSAGE(reserve_rc >= 0,
"xzarray_reserve must succeed for a valid array and count");
int caps_after_reserve = xzarray_test_get_caps(array);
TEST_ASSERT_TRUE_MESSAGE(caps_after_reserve >= n + 1,
"reservation must genuinely grow capacity to cover n+1 items");
for (int i = 0; i < n; i++) {
char label[32];
snprintf(label, sizeof(label), "item-%d", i);
int idx = vstore(array, (void*)"x");
TEST_ASSERT_TRUE(idx >= 0);
int caps_now = xzarray_test_get_caps(array);
TEST_ASSERT_EQUAL_INT_MESSAGE(caps_after_reserve, caps_now,
"capacity must NOT change during any of the n reserved stores --"
" a change here means a realloc() happened despite a successful"
" reservation, defeating the whole point of task 7.1");
}
destroy(array);
printf("test_xzarray_reserve_pregrows_capacity_no_growth_during_"
"subsequent_stores: PASS\n");
}
/* ── The verified off-by-one: reserving EXACTLY n and then storing n items
* still triggers exactly one (small, delta-bounded) growth on the LAST
* store -- documents/proves why job.c's caller reserves array_size + 1, not
* array_size, to fully close growth during its bulk-insert pass. ────────── */
static void test_xzarray_reserve_off_by_one_without_extra_headroom(void)
{
const int delta = 1;
const int n = 10;
var array = make_test_xzarray(delta);
TEST_ASSERT_NOT_NULL(array);
int reserve_rc = xzarray_reserve(array, n); /* exactly n, no +1 */
TEST_ASSERT_TRUE(reserve_rc >= 0);
int caps_after_reserve = xzarray_test_get_caps(array);
int grew_during_stores = 0;
for (int i = 0; i < n; i++) {
vstore(array, (void*)"x");
if (xzarray_test_get_caps(array) != caps_after_reserve) {
grew_during_stores = 1;
}
}
TEST_ASSERT_TRUE_MESSAGE(grew_during_stores,
"reserving EXACTLY n and then storing n items is expected to still"
" trigger one growth cycle on the last store (xzarray_ensure_items's"
" strict mincap < caps check) -- this is the documented reason"
" job.c's kvmgr_reserve() call reserves array_size + 1, not"
" array_size");
destroy(array);
printf("test_xzarray_reserve_off_by_one_without_extra_headroom: PASS\n");
}
/* ── Baseline contrast: WITHOUT any reservation, the same N-store sequence
* genuinely does grow capacity (the pre-task-7.1 behavior) ─────────────── */
static void test_xzarray_without_reservation_growth_occurs_during_stores(void)
{
const int delta = 1;
const int n = 20;
var array = make_test_xzarray(delta);
TEST_ASSERT_NOT_NULL(array);
int caps_initial = xzarray_test_get_caps(array);
TEST_ASSERT_EQUAL_INT(0, caps_initial);
int grew_during_stores = 0;
for (int i = 0; i < n; i++) {
int before = xzarray_test_get_caps(array);
vstore(array, (void*)"x");
int after = xzarray_test_get_caps(array);
if (after != before) {
grew_during_stores = 1;
}
}
TEST_ASSERT_TRUE_MESSAGE(grew_during_stores,
"without a reservation, growth during the store sequence is"
" expected -- this is the pre-existing baseline task 7.1 improves"
" on, not a new claim");
destroy(array);
printf("test_xzarray_without_reservation_growth_occurs_during_stores: "
"PASS\n");
}
/* ── Simulated allocation failure propagates as -ENOMEM ─────────────────── */
static void test_xzarray_reserve_propagates_simulated_allocation_failure(void)
{
var array = make_test_xzarray(4);
TEST_ASSERT_NOT_NULL(array);
xzarray_reserve_test_set_force_fail();
int rc = xzarray_reserve(array, 10);
TEST_ASSERT_EQUAL_INT_MESSAGE(-ENOMEM, rc,
"a forced allocation failure must propagate as -ENOMEM");
xzarray_reserve_test_clear_force_fail();
/* Clearing the seam must restore normal behavior. */
int rc2 = xzarray_reserve(array, 10);
TEST_ASSERT_TRUE_MESSAGE(rc2 >= 0,
"clearing the force-fail seam must restore normal success");
destroy(array);
printf("test_xzarray_reserve_propagates_simulated_allocation_failure: "
"PASS\n");
}
/* ── Defensive / boundary ─────────────────────────────────────────────── */
static void test_xzarray_reserve_rejects_null_array(void)
{
int rc = xzarray_reserve(NULL, 5);
TEST_ASSERT_TRUE(rc < 0);
printf("test_xzarray_reserve_rejects_null_array: PASS\n");
}
static void test_xzarray_reserve_rejects_negative_n(void)
{
var array = make_test_xzarray(4);
TEST_ASSERT_NOT_NULL(array);
int rc = xzarray_reserve(array, -1);
TEST_ASSERT_TRUE(rc < 0);
destroy(array);
printf("test_xzarray_reserve_rejects_negative_n: PASS\n");
}
static void test_xzarray_reserve_zero_is_noop_success(void)
{
var array = make_test_xzarray(4);
TEST_ASSERT_NOT_NULL(array);
int rc = xzarray_reserve(array, 0);
TEST_ASSERT_TRUE(rc >= 0);
destroy(array);
printf("test_xzarray_reserve_zero_is_noop_success: PASS\n");
}
/* ── Test group entry point ───────────────────────────────────────────── */
void test_zarray_reserve(void)
{
printf("\n>> xzarray_reserve test\n");
RUN_TEST(
test_xzarray_reserve_pregrows_capacity_no_growth_during_subsequent_stores);
RUN_TEST(test_xzarray_reserve_off_by_one_without_extra_headroom);
RUN_TEST(test_xzarray_without_reservation_growth_occurs_during_stores);
RUN_TEST(test_xzarray_reserve_propagates_simulated_allocation_failure);
RUN_TEST(test_xzarray_reserve_rejects_null_array);
RUN_TEST(test_xzarray_reserve_rejects_negative_n);
RUN_TEST(test_xzarray_reserve_zero_is_noop_success);
}

69
src/lib/zarray.c
View file

@ -780,3 +780,72 @@ typeinfo_t xzarray_type = {
.lock = &(intf_lock_t) { .lock = (lock_fn_t)xzarray_lock,
.unlock = (unlock_fn_t)xzarray_unlock } }
};
/* ------------------------------------------------------------------ */
/* xzarray_reserve -- job-array-single-wal-record task 7.1 (bug-1443) */
/* ------------------------------------------------------------------ */
/* NOTE on why these test-support symbols are NOT wrapped in
* `#ifdef XCE_TESTING` here, unlike job.c's analogous seams: job.c is
* recompiled from source directly into each test binary's own source list
* (controller_unit_test_src, meson.build) with `-DXCE_TESTING` supplied as
* that binary's own c_args, so an `#ifdef XCE_TESTING` guard in job.c is
* evaluated per-binary. zarray.c, by contrast, is compiled EXACTLY ONCE, as
* part of the shared `libxce` static_library() target (src/lib/meson.build)
* -- WITHOUT `-DXCE_TESTING` -- and that single libxce.a is then linked,
* unmodified, into every consumer, test binaries included. Guarding these
* definitions behind `#ifdef XCE_TESTING` here would compile them out of
* libxce.a entirely, in every build, making them permanently unresolvable
* link errors for the test binaries that need them (confirmed empirically:
* this was the very shape of the link failure this fix corrects). The
* header-side declarations (zarray.h) remain guarded, so these symbols stay
* un-prototyped -- and therefore uncallable without a compile error -- from
* any production translation unit; only the symbol bytes are unconditionally
* present in libxce.a, not a callable, declared API. The force-fail flag
* below defaults to false and only ever affects xzarray_reserve() itself
* (never any other function), so its unconditional presence is inert unless
* a test explicitly engages it. File-scope, not per-instance -- mirrors
* job.c's own s_job_array_bulk_insert_force_fail_at convention otherwise. */
static _Atomic _Bool s_xzarray_reserve_force_fail = false;
void xzarray_reserve_test_set_force_fail(void)
{
atomic_store_explicit(&s_xzarray_reserve_force_fail, true,
memory_order_relaxed);
}
void xzarray_reserve_test_clear_force_fail(void)
{
atomic_store_explicit(&s_xzarray_reserve_force_fail, false,
memory_order_relaxed);
}
int xzarray_test_get_caps(var array)
{
if (array == NULL) {
return -EINVAL;
}
meta_obj_t* obj = RTTI_DATA_TO_OBJ(array);
xzarray_t* a = RTTI_OBJ_TO_DATA(xzarray_t, obj);
return (int)a->caps;
}
int xzarray_reserve(var array, int n)
{
if ((array == NULL) || (n < 0)) {
return -EINVAL;
}
if (atomic_load_explicit(&s_xzarray_reserve_force_fail,
memory_order_relaxed)) {
errno = ENOMEM; /* simulated resource-exhaustion failure */
return -ENOMEM;
}
meta_obj_t* obj = RTTI_DATA_TO_OBJ(array);
xzarray_t* a = RTTI_OBJ_TO_DATA(xzarray_t, obj);
return xzarray_ensure_items(a, a->count + n);
}