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

feat(ha-commit-barrier-rollout): wal_classify + classified hook dispatch (task 1.2)

Browse source 
Verified table implemented (STOP/RESUME/UPDATE/FAIL RETAIN per
supersession; WORKER detail switch; CHECKPOINT LOCAL); unknown type =
ERROR + XCE_TESTING abort + production RETAIN fallback. All three
hooks route through wal_hook_append_classified (zero bare async
appends); RETAIN registers with the engine lock-free; facts propagate
local append rc only. 8/8 group tests green; full suite 465/465.
This commit is contained in:
Rongsong Shen 2026-06-12 04:43:50 +00:00
parent fec9519960
commit e8ecc87551
4 changed files with 138 additions and 20 deletions
Download patch file Download diff file Expand all files Collapse all files

144
src/controller/persist.c
View file

@ -9,6 +9,7 @@
#include "controller.capnp.h"
#include "xlist.h"
#include "cluster.h"
#include "commit.h" /* commit_submit RETAIN registration (barrier-rollout 1.2) */
#include <assert.h>
#include <stdbool.h>
@ -1118,16 +1119,104 @@ static uint32_t detail_to_wal_type(int detail)
}
}
/* Write-path commit classification — INERT RED-phase stub
* (ha-commit-barrier-rollout task 1.1; sanctioned scaffolding so the RED
* tests fail on assertions, not at link time). Returns an out-of-enum
* sentinel: classifies nothing yet. The verified classification table
* (design.md §Write-Path Classification) is implemented in task 1.2. */
/* Write-path commit classification (ha-commit-barrier-rollout task 1.2).
* Pure lookup over the VERIFIED table (design.md §Write-Path
* Classification); `detail` disambiguates WAL_WORKER_CTRL only.
* Unclassified types fail fast under XCE_TESTING (abort) and fall back to
* RETAIN in production — never dropped, never falsely confirmed (Req 2.2). */
wal_commit_mode_t wal_classify(uint32_t wal_type, int detail)
{
(void)wal_type;
(void)detail;
return (wal_commit_mode_t)-1; /* invalid sentinel — task 1.2 replaces */
switch (wal_type) {
case WAL_JOB_SUBMIT:
/* client job submit — request-shaped; keeps the in-hook sync path
* (reference implementation, handle_jobsubmit) */
return WAL_MODE_WAIT;
case WAL_QUEUE_CTRL:
/* client queue control — request-shaped; the handler barrier owes
* commit completion before the requester reply (tasks 2-3) */
return WAL_MODE_WAIT;
case WAL_WORKER_CTRL:
/* the only detail-disambiguated type: JOIN/UPDATE have a requester
* awaiting confirmation (WAIT); DEAD is a controller-detected fact.
* Unexpected details take the fact-safe RETAIN side. */
if (detail == HOOK_DETAIL_WORKER_JOIN ||
detail == HOOK_DETAIL_WORKER_UPDATE)
return WAL_MODE_WAIT;
return WAL_MODE_RETAIN;
case WAL_JOB_START:
case WAL_JOB_FINISH:
case WAL_JOB_FAIL:
case WAL_JOB_STOP:
case WAL_JOB_RESUME:
case WAL_JOB_UPDATE:
case WAL_JOB_RUSAGE:
case WAL_JOB_CKPT:
case WAL_JOB_PREASON:
/* worker/scheduler-reported facts — retained until committed.
* FAIL/UPDATE are dual-origin; their request-shaped side is
* realized by the handler barrier in do_jobcontrol_op
* (research.md §2 call-site verification). */
return WAL_MODE_RETAIN;
case WAL_CHECKPOINT:
/* local segment marker — never replicated, by design */
return WAL_MODE_LOCAL;
default:
ERROR("wal_classify: unclassified WAL type %u (detail %d) — "
"RETAIN fallback; add it to the classification table",
wal_type, detail);
#ifdef XCE_TESTING
assert(0 && "unclassified WAL type — add it to wal_classify");
#endif
return WAL_MODE_RETAIN;
}
}
/* Classified append at the hook seam (barrier-rollout task 1.2).
*
* WAIT (queue/worker control): the append stays async here — the handler
* barrier (tasks 2-3) owes commit completion before the requester reply.
* WAL_JOB_SUBMIT never reaches this helper (it keeps the in-hook sync
* path in persist_job_hook_fun).
*
* RETAIN (facts): append + broadcast, then register the appended LSN with
* the commit engine. commit_submit(COMMIT_MODE_RETAIN) is registration-
* free bookkeeping (commit.c — returns 0 immediately; commit state is
* derived from registry watermarks at ACK time), so the call is contract
* documentation: the entry stays pending purely by lsn > committed_lsn
* and commits later via covering ACKs. Commit-progress results are NEVER
* propagated to hook callers — only the local append rc is (Req 3.1).
*
* Lock context: persist_wal_async takes/releases p->lock internally; the
* appended-LSN read below re-takes p->lock briefly (same discipline as
* persist_wal_sync_append) and releases it BEFORE commit_submit, so no
* p->lock is ever held across ce->lock (engine lock-ordering rule). */
static int wal_hook_append_classified(persist_t* p, uint32_t wal_type,
wal_commit_mode_t mode,
const void* payload, uint32_t len)
{
uint64_t appended_lsn;
int rc;
rc = persist_wal_async(p, wal_type, payload, len);
if (rc != 0 || mode != WAL_MODE_RETAIN)
return rc;
if (p->repl.ce == NULL)
return rc; /* engine not wired (bootstrap / engine-less contexts) */
/* current_lsn - 1 is the LSN persist_wal_async just assigned (it
* increments under p->lock). A concurrent append may bump it further;
* that is benign — contiguous advancement means the higher watermark
* still covers this entry. LSN-0 edge: commit_submit rejects lsn == 0
* (first-ever entry of a fresh cluster — accepted interim behavior,
* spec Implementation Notes). */
pthread_mutex_lock(&p->lock);
appended_lsn = p->current_lsn - 1;
pthread_mutex_unlock(&p->lock);
if (appended_lsn != 0)
(void)commit_submit(p->repl.ce, appended_lsn, COMMIT_MODE_RETAIN, 0);
return rc; /* local append rc only — facts have no commit timeout */
}
/* P0 payload: just the detail value (4 bytes).
@ -1142,13 +1231,18 @@ int persist_job_hook_fun(int type, int situation, int detail,
if (situation != HOOK_AFTER_EVENT)
return 0;
persist_t* p = cls;
uint32_t wal_type = detail_to_wal_type(detail);
persist_t* p = cls;
uint32_t wal_type = detail_to_wal_type(detail);
wal_commit_mode_t mode;
if (wal_type == 0)
return 0;
mode = wal_classify(wal_type, detail);
if (wal_type == WAL_JOB_SUBMIT) {
/* AFTER_EVENT hook args: cluster_t* c, xlist_t* jlist */
if (mode == WAL_MODE_WAIT) {
/* WAL_JOB_SUBMIT — the only WAIT type in the job hook; reference
* implementation keeps the existing in-hook sync path.
* AFTER_EVENT hook args: cluster_t* c, xlist_t* jlist */
(void)va_arg(ap, cluster_t*);
xlist_t* jlist = va_arg(ap, xlist_t*);
int rc = 0;
@ -1209,7 +1303,8 @@ int persist_job_hook_fun(int type, int situation, int detail,
}
}
return persist_wal_async(p, wal_type, &js, sizeof(js));
/* mode == WAL_MODE_RETAIN — fact-shaped job state transition */
return wal_hook_append_classified(p, wal_type, mode, &js, sizeof(js));
}
int persist_queue_hook_fun(int type, int situation, int detail,
@ -1219,10 +1314,13 @@ int persist_queue_hook_fun(int type, int situation, int detail,
if (situation != HOOK_AFTER_EVENT)
return 0;
persist_t* p = cls;
uint32_t wal_type = detail_to_wal_type(detail);
persist_t* p = cls;
uint32_t wal_type = detail_to_wal_type(detail);
wal_commit_mode_t mode;
if (wal_type == 0)
return 0;
mode = wal_classify(wal_type, detail);
/* args: cluster_t*, queue_t* */
(void)va_arg(ap, cluster_t*);
@ -1239,7 +1337,10 @@ int persist_queue_hook_fun(int type, int situation, int detail,
memcpy(pkt, &hdr, sizeof(hdr));
memcpy(pkt + sizeof(hdr), name, name_len);
int rc = persist_wal_async(p, wal_type, pkt, (uint32_t)total);
/* mode == WAL_MODE_WAIT — the append stays async; the handler barrier
* owes commit completion before the requester reply (task 3) */
int rc = wal_hook_append_classified(p, wal_type, mode, pkt,
(uint32_t)total);
free(pkt);
return rc;
}
@ -1251,10 +1352,13 @@ int persist_worker_hook_fun(int type, int situation, int detail,
if (situation != HOOK_AFTER_EVENT)
return 0;
persist_t* p = cls;
uint32_t wal_type = detail_to_wal_type(detail);
persist_t* p = cls;
uint32_t wal_type = detail_to_wal_type(detail);
wal_commit_mode_t mode;
if (wal_type == 0)
return 0;
mode = wal_classify(wal_type, detail);
/* WAL_WORKER_CTRL payload layout: [uint32_t detail][capnp worker OR name] */
uint32_t disc = (uint32_t)detail;
@ -1281,7 +1385,9 @@ int persist_worker_hook_fun(int type, int situation, int detail,
memcpy(pkt + WORKER_WAL_HDR_SIZE, buf, capnp_sz);
free(buf);
rc = persist_wal_async(p, wal_type, pkt, (uint32_t)total);
/* JOIN/UPDATE → WAIT (handler barrier owes completion, task 3);
* DEAD → RETAIN (controller-detected fact, registered with the engine) */
rc = wal_hook_append_classified(p, wal_type, mode, pkt, (uint32_t)total);
free(pkt);
return rc;
}