feat(ha-entity-snapshots): add spec + task 1.1 test fixture

Adds the ha-entity-snapshots Kiro spec (requirements, design, research,
tasks) for task 5.10a / raft-ha-core-safety: making the HA checkpoint an
actually-restorable copy of committed job/queue state instead of a
position-only sentinel, closing the residual chaotic-soak divergence gap.

Task 1.1 (foundation): reusable test fixture snapent_make/snapent_teardown
in a new test_persist_snapshot_entities.c, seeding N jobs (varying state)
and M queues (varying status) into a real persist_t+cluster_t for later
write/load/rotation-guard tests to build on. Build clean; new suite 1/1;
regression spot-checks (test_persist_hooks 19/19, test_commit_barrier_reply
22/22) clean.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01JDYFbrgyDHbvNYcxApKhrX
This commit is contained in:
Rongsong Shen 2026-07-01 04:55:55 +00:00
parent 0d70d5e412
commit 10f9bbfd9d
8 changed files with 1253 additions and 1 deletions

View file

@ -73,7 +73,7 @@ extern void test_ha_vep_teardown(void);
extern void test_ha_dispatch_integration(void);
extern void test_ha_core_safety(void);
extern void test_ha_partition(void);
extern void test_ha_partition(void);
extern void test_persist_snapshot_entities(void);
void setUp(void) { return; }
@ -216,6 +216,7 @@ int main(int argc, char* argv[])
TEST(test_ha_dispatch_integration);
TEST(test_ha_core_safety);
TEST(test_ha_partition);
TEST(test_persist_snapshot_entities);
return UNITY_END();
}

View file

@ -0,0 +1,306 @@
/*
* test_persist_snapshot_entities.c — entity-bearing checkpoint test suite
* (ha-entity-snapshots).
*
* Task 1.1 (Requirements 1.1, 2.1): a reusable test fixture wrapping a real
* persist_t embedded in a minimal test cluster_t whose jobmgr/qmgr tables are
* seeded with N jobs (varying state) and M queues (varying status) — the
* foundation later tasks in this file build on:
* - 2.1/2.2: write_snapshot/load_snapshot job round-trip, per-entity
* consistency, partial-write discard
* - 3.1/3.2: queue admin-status round-trip
* - 4.2: rotation-guard (in-flight checkpoint survives rotate_snapshots)
* - 5.1/5.2: restore-before-advance, failure-safety, legacy-format handling
*
* Modeled on:
* - test_persist_hooks.c's make_hook_test_persist/cleanup_hook_test_persist
* (bare persist_t over a private tmpdir, no HA/commit-engine wiring)
* - test_commit_barrier_reply.c's cbrp_setup / test_egress_filter.c's
* ef_setup (persist_t embedded in a real cluster_t with jobmgr/qmgr wired
* as xkvmgr instances, back-pointer registered via persist_init so
* persist_start can register hooks)
*
* Deliberately OMITTED vs. cbrp_setup/ef_setup: commit engine, standby
* registry, repl_broadcast_send_hook, and the LSN-0 priming append. None of
* those are needed here — create_job/makeJob and makeQueue insert directly
* into the cluster's tables with no WAL append or replication involved, and
* write_snapshot/persist_checkpoint never call repl_broadcast (checkpoint
* markers are local-only; see persist.c persist_write_checkpoint_marker).
* Adding that machinery here would be unused scaffolding.
*
* job_t/queue_t insertion mirrors the established test idiom:
* - ef_make_job (test_egress_filter.c): zero an xce_job_t, set
* jobid/state/submit, call makeJob(jobmgr, &jdef) — makeJob deep-copies
* the submit payload (jobj.c job_init -> makeJobsub), so a single reused
* scratch job_submit_data_t is safe across a seeding loop.
* - cbrp_arrange_queuectrl (test_commit_barrier_reply.c): zero an
* xce_queue_t, set queue (name) + status, call makeQueue(qmgr, &qdef) —
* queue_init deep-copies the name via makeString, and the underlying
* string-keyed kvtab strdup's the key on insert (kv.c assign_string_key),
* so a single reused scratch name buffer is likewise safe.
*
* NOTE (design.md / bug investigation 2026-06-30): queues are config-
* provisioned at startup, not created from WAL/snapshot data — restore_queue_state
* can only re-apply status to a queue that already exists in qmgr. Seeding
* queues directly here (as makeQueue does) stands in for that config-load
* step, matching what a real config-provisioned queue table looks like by
* the time write_snapshot/load_snapshot ever touch it.
*/
#define _GNU_SOURCE
#include <rtti.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <unity/unity.h>
#include "cluster.h"
#include "jobj.h"
#include "qobj.h"
#include "xce/cluster_def.h"
#include "xce/types.h"
/* ------------------------------------------------------------------ */
/* Deterministic seed data — later tests predict seeded values from the */
/* loop index via these helpers rather than the fixture returning arrays */
/* ------------------------------------------------------------------ */
#define SNAPENT_JOBID_BASE (1000ULL)
static const unsigned int g_snapent_job_states[] = {
JOB_STAT_PEND, JOB_STAT_RUN, JOB_STAT_EXIT, JOB_STAT_DONE, JOB_STAT_PSUSP,
};
#define SNAPENT_N_JOB_STATES \
(sizeof(g_snapent_job_states) / sizeof(g_snapent_job_states[0]))
static const uint32_t g_snapent_queue_statuses[] = {
QUEUE_STAT_OPEN, QUEUE_STAT_CLOSED, QUEUE_STAT_INACT_ADMIN,
(QUEUE_STAT_CLOSED | QUEUE_STAT_INACT_ADMIN),
};
#define SNAPENT_N_QUEUE_STATUSES \
(sizeof(g_snapent_queue_statuses) / sizeof(g_snapent_queue_statuses[0]))
static uint64_t snapent_jobid(int i)
{
return SNAPENT_JOBID_BASE + (uint64_t)i;
}
static unsigned int snapent_job_state(int i)
{
return g_snapent_job_states[i % (int)SNAPENT_N_JOB_STATES];
}
static void snapent_queue_name(int i, char* buf, size_t buf_sz)
{
snprintf(buf, buf_sz, "snapent-q%d", i);
}
static uint32_t snapent_queue_status(int i)
{
return g_snapent_queue_statuses[i % (int)SNAPENT_N_QUEUE_STATUSES];
}
/* ------------------------------------------------------------------ */
/* Fixture: persist_t + cluster_t with seeded job/queue tables */
/* ------------------------------------------------------------------ */
typedef struct {
cluster_t* c;
char tmpdir[256];
} snapent_fixture_t;
/* jobj.c's jobsub/startinfo tables are process-global and idempotently
* guarded — mirror the per-file static flag used by test_egress_filter.c /
* test_commit_barrier_reply.c so this file can run standalone or alongside
* them in the same controller_unit_test binary. */
static int g_snapent_jobobj_ready;
static void snapent_seed_jobs(cluster_t* c, int n)
{
static job_submit_data_t submit_scratch;
for (int i = 0; i < n; i++) {
xce_job_t jdef;
memset(&jdef, 0, sizeof(jdef));
memset(&submit_scratch, 0, sizeof(submit_scratch));
jdef.jobid = snapent_jobid(i);
jdef.state = snapent_job_state(i);
jdef.submit = &submit_scratch;
TEST_ASSERT_NOT_NULL_MESSAGE(makeJob(c->jobmgr, &jdef),
"seed job insertion must succeed");
}
}
static void snapent_seed_queues(cluster_t* c, int m)
{
for (int i = 0; i < m; i++) {
char name[32];
snapent_queue_name(i, name, sizeof(name));
xce_queue_t qdef;
memset(&qdef, 0, sizeof(qdef));
qdef.queue = name; /* deep-copied via makeString in queue_init */
qdef.status = snapent_queue_status(i);
TEST_ASSERT_NOT_NULL_MESSAGE(makeQueue(c->qmgr, &qdef),
"seed queue insertion must succeed");
}
}
/* make an entity-snapshot fixture: a real persist_t (private tmpdir) embedded
* in a minimal cluster_t whose jobmgr/qmgr are seeded with n_jobs/m_queues.
* persist_init registers the cluster back-pointer (p->cluster) so future
* write_snapshot/load_snapshot entity collection can walk c->jobmgr/c->qmgr —
* the same access path collect_jobs/collect_queues (dataex.c) use. */
static snapent_fixture_t* snapent_make(int n_jobs, int m_queues)
{
kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT,
.nslots = 64,
.delta = 16 };
kvmgr_params_t str_params = { .key_type = KVMGR_KEY_STRING,
.nslots = 64,
.delta = 16 };
snapent_fixture_t* f = calloc(1, sizeof(*f));
if (f == NULL)
return NULL;
f->c = calloc(1, sizeof(cluster_t));
if (f->c == NULL) {
free(f);
return NULL;
}
f->c->myhost = &f->c->myhost_info;
f->c->myhost_info.name = (char*)"snapent-test";
f->c->myhost_info.key = (char*)"";
TEST_ASSERT_EQUAL_INT(
0, xce_hook_init(&f->c->hooksys, XCE_CONTROLLER_HOOK_MAX));
f->c->jobmgr = makeInstance(&xkvmgr_type, &int_params);
f->c->qmgr = makeInstance(&xkvmgr_type, &str_params);
TEST_ASSERT_NOT_NULL(f->c->jobmgr);
TEST_ASSERT_NOT_NULL(f->c->qmgr);
zarray_init(&f->c->sharedinst, 64, NULL);
if (!g_snapent_jobobj_ready) {
TEST_ASSERT_EQUAL_INT(0, job_obj_init(64, 64));
g_snapent_jobobj_ready = 1;
}
persist_config_t cfg = persist_config_defaults();
cfg.wal_sync_ms = 5000; /* slow background fsync — avoid races */
snprintf(f->tmpdir, sizeof(f->tmpdir), "/tmp/xce_snapent_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(f->tmpdir));
persist_set_workdir(&f->c->persist, f->tmpdir, "snapent");
TEST_ASSERT_EQUAL_INT(0, persist_init(&f->c->persist, &cfg, f->c));
TEST_ASSERT_EQUAL_INT(0, persist_start(&f->c->persist));
snapent_seed_jobs(f->c, n_jobs);
snapent_seed_queues(f->c, m_queues);
return f;
}
static void snapent_teardown(snapent_fixture_t* f)
{
if (f == NULL)
return;
persist_stop(&f->c->persist);
persist_fini(&f->c->persist);
zarray_fini(&f->c->sharedinst);
if (f->c->qmgr) dereference(f->c->qmgr);
if (f->c->jobmgr) dereference(f->c->jobmgr);
xce_hook_fini(&f->c->hooksys);
free(f->c);
free(f);
}
/* Lookup helpers — mirror ef_find_job (test_egress_filter.c). */
static job_t* snapent_find_job(cluster_t* c, uint64_t jobid)
{
job_t* j = NULL;
int found = 0;
lock(c->jobmgr, RTTI_LOCK_READ);
found = key_find(c->jobmgr, (void*)jobid, (void**)&j);
unlock(c->jobmgr);
return (found > 0) ? j : NULL;
}
static queue_t* snapent_find_queue(cluster_t* c, const char* name)
{
queue_t* q = NULL;
int found = 0;
lock(c->qmgr, RTTI_LOCK_READ);
found = key_find(c->qmgr, (void*)name, (void**)&q);
unlock(c->qmgr);
return (found > 0) ? q : NULL;
}
/* ------------------------------------------------------------------ */
/* Test 1: fixture seeds N jobs (varying state) + M queues (varying */
/* status) and both are independently observable via key_find */
/* ------------------------------------------------------------------ */
static void test_snapent_fixture_seeds_jobs_and_queues(void)
{
const int n_jobs = 5;
const int m_queues = 4;
snapent_fixture_t* f = snapent_make(n_jobs, m_queues);
TEST_ASSERT_NOT_NULL(f);
TEST_ASSERT_NOT_NULL(f->c);
/* persist_init must have registered the cluster back-pointer — this is
* what lets write_snapshot/load_snapshot reach c->jobmgr/c->qmgr. */
TEST_ASSERT_EQUAL_PTR(f->c, f->c->persist.cluster);
for (int i = 0; i < n_jobs; i++) {
job_t* j = snapent_find_job(f->c, snapent_jobid(i));
TEST_ASSERT_NOT_NULL_MESSAGE(j, "seeded job must be present in jobmgr");
TEST_ASSERT_EQUAL_UINT64(snapent_jobid(i), j->job.jobid);
TEST_ASSERT_EQUAL_UINT32(snapent_job_state(i), j->job.state);
}
for (int i = 0; i < m_queues; i++) {
char name[32];
snapent_queue_name(i, name, sizeof(name));
queue_t* q = snapent_find_queue(f->c, name);
TEST_ASSERT_NOT_NULL_MESSAGE(q, "seeded queue must be present in qmgr");
TEST_ASSERT_EQUAL_UINT32(snapent_queue_status(i), q->queue.status);
}
/* Sanity: lookups for entities that were never seeded must miss. */
TEST_ASSERT_NULL(snapent_find_job(f->c, snapent_jobid(n_jobs) + 999));
TEST_ASSERT_NULL(snapent_find_queue(f->c, "snapent-does-not-exist"));
snapent_teardown(f);
printf("test_snapent_fixture_seeds_jobs_and_queues: PASS\n");
}
/* ------------------------------------------------------------------ */
/* Suite entry point */
/* ------------------------------------------------------------------ */
void test_persist_snapshot_entities(void)
{
printf("\n>> entity-bearing snapshot fixture tests\n\n");
RUN_TEST(test_snapent_fixture_seeds_jobs_and_queues);
}