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

Fix two flaky test failures: commit sweep timeout + array stats identity

Browse source 
Controller suite flake (user-reported, -j parallel runs): the commit
engine's async timeout sweep armed continuation deadlines with the
hardcoded DEFAULT_SYNC_TIMEOUT_MS (500ms) instead of the configured
sync_timeout_ms, while the blocking persist_commit_barrier honors the
config. Fixtures configuring longer or shorter sync timeouts saw the
sweep resolve -ETIMEDOUT at 500ms before their explicit
truncation/second-call steps, flipping "Expected -17/-2 Was 0/-110"
assertions (test_commit_resolve.c:1458/:1616). commit_submit_async and
commit_sweep_worker now both use p->config.sync_timeout_ms with
DEFAULT_SYNC_TIMEOUT_MS only as the zeroed-config fallback (mirroring
persist_commit_barrier's timeout_ms == 0 convention), and the sweep
WARN reports the effective value.

Apiserver flake (found during full-build verification):
deriveArrayJobStats took the array identity from the first element
seen over store.Jobs.List(), but List() walks a Go map whose order is
randomized per call -- when element rows disagree (the pinned test
case), the stats identity flipped run to run (~12% of -count=1000
runs). The identity carrier is now the lowest Unique_JobID (the array
head, element 1), tracked via arrayAgg.carrierUID; other rows
contribute counters only. Counters were already order-independent and
are unchanged. The test now pins the head-carries-identity contract
explicitly.

Also fixes a latent parse-error trap in the test: a string literal
ending a line (C-style implicit concatenation) triggers Go's automatic
semicolon insertion -- the next line's argument list parses as
"missing ',' before newline". String concatenation across lines must
keep the '+' at the end of the line.

Verified: controller suite 1342/0 (three runs incl. -j 7 parallel
stress); sweep WARNs now match fixture configs (300ms fixture warns at
300, longer-config fixtures emit none); go test -run
TestDeriveArrayJobStats -count=1000 all green (was 118-130 failures);
full ./tools/xce-build.sh all green end to end.

Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
lei.chen 2026-10-22 02:51:42 +00:00
parent 84f8d6b76a
commit 5b9d727086
5 changed files with 108 additions and 56 deletions
Download patch file Download diff file Expand all files Collapse all files

36
src/controller/commit.c
View file

@ -749,13 +749,23 @@ int commit_submit_async(commit_engine_t* ce, uint64_t lsn,
ce->continuations[slot].fn = fn;
ce->continuations[slot].ctx = ctx;
/* raft-apply-decoupling task 1.3: arm this registration's timeout with
* the SAME deadline computation and SAME default constant
* commit_submit's blocking WAIT path uses, so an async registration
* observes identical failure timing to today's blocking path (Req
* 11.1). The dedicated timeout-sweep thread (commit_sweep_worker)
* resolves this continuation FAILED/-ETIMEDOUT if it is still pending
* once this deadline passes. */
wait_deadline(&ce->continuations[slot].deadline, DEFAULT_SYNC_TIMEOUT_MS);
* the SAME deadline computation and SAME configured timeout the blocking
* paths use, so an async registration observes identical failure timing
* to the blocking path (Req 11.1). The configured sync deadline -- not a
* hardcoded constant -- is the source of truth: the blocking
* persist_commit_barrier borrows p->config.sync_timeout_ms when its
* caller passes timeout_ms == 0 (persist.c), and this async path honors
* the same config here. The sweep WARN below reports the same value.
* A zeroed config (unit fixtures memset()ing the persist_t, no
* persist_init) falls back to DEFAULT_SYNC_TIMEOUT_MS, mirroring
* persist_commit_barrier's timeout_ms == 0 convention. The dedicated
* timeout-sweep thread (commit_sweep_worker) resolves this continuation
* FAILED/-ETIMEDOUT if it is still pending once this deadline passes. */
{
uint32_t tmo = ce->p->config.sync_timeout_ms;
wait_deadline(&ce->continuations[slot].deadline,
tmo != 0 ? tmo : DEFAULT_SYNC_TIMEOUT_MS);
}
/* Wake the sweep thread: a freshly armed deadline may be sooner than
* whatever it is currently sleeping toward. */
@ -1106,6 +1116,16 @@ static void* commit_sweep_worker(void* arg)
&next_deadline, &have_next);
if (nc > 0 || nf > 0) {
/* Same effective timeout commit_submit_async armed the deadline
* with: the configured sync timeout, DEFAULT only as the
* zeroed-config (unit fixture) fallback. Read under ce->lock —
* persist config is write-once at init in practice, but the
* read is already inside a lock hold that guarantees the
* engine outlives it. */
uint32_t tmo = ce->p->config.sync_timeout_ms;
if (tmo == 0) {
tmo = DEFAULT_SYNC_TIMEOUT_MS;
}
pthread_mutex_unlock(&ce->lock);
for (int i = 0; i < nc; i++) {
commit_resolve(ce, committed_ready[i], COMMIT_OUTCOME_COMMITTED, 0);
@ -1115,7 +1135,7 @@ static void* commit_sweep_worker(void* arg)
"commit (lsn=%lu committed_lsn=%lu writable_lsn=%lu "
"timeout_ms=%u)",
failed_ready[i], atomic_load(&ce->p->committed_lsn),
ce->writable_lsn, (unsigned)DEFAULT_SYNC_TIMEOUT_MS);
ce->writable_lsn, (unsigned)tmo);
commit_resolve(ce, failed_ready[i], COMMIT_OUTCOME_FAILED,
-ETIMEDOUT);
}

12
src/controller/include/commit.h
View file

@ -51,9 +51,10 @@ typedef struct commit_continuation {
/* raft-apply-decoupling task 1.3: absolute CLOCK_REALTIME deadline this
* registration must resolve by if majority is never reached -- computed
* at registration time via the SAME wait_deadline() helper and the SAME
* DEFAULT_SYNC_TIMEOUT_MS constant commit_submit's blocking WAIT path
* uses, so an async registration observes identical failure timing to
* today's blocking path (Req 11.1). Only meaningful while lsn != 0;
* configured sync timeout (p->config.sync_timeout_ms;
* DEFAULT_SYNC_TIMEOUT_MS as the zeroed-config fallback) the blocking
* paths use, so an async registration observes identical failure timing
* to the blocking path (Req 11.1). Only meaningful while lsn != 0;
* overwritten on every fresh registration of a reused slot. Swept by the
* commit engine's dedicated timeout-sweep thread (see commit.c). */
struct timespec deadline;
@ -170,8 +171,9 @@ void commit_blocking_release(commit_engine_t* ce);
* outcome is known.
*
* raft-apply-decoupling task 1.3: registration also arms an implicit
* DEFAULT_SYNC_TIMEOUT_MS deadline (identical constant and clock the
* blocking WAIT path uses) -- if lsn has not resolved by then, the commit
* deadline -- the SAME configured sync timeout (p->config.sync_timeout_ms,
* DEFAULT_SYNC_TIMEOUT_MS only as the zeroed-config fallback) and clock the
* blocking WAIT path uses -- if lsn has not resolved by then, the commit
* engine's own timeout-sweep thread resolves it FAILED/-ETIMEDOUT with no
* caller action required. A genuine majority ack that lands at or after the
* deadline still wins: the sweep thread always checks the majority

86
src/frontend/apiserver/controllers/arraystats.go
View file

@ -24,6 +24,15 @@ import (
// arrayAgg accumulates one array's counters while scanning store.Jobs.
type arrayAgg struct {
info models.ArrayJobInfo
// carrierUID is the element row currently supplying info's identity
// fields. Identity selection must be deterministic: store.Jobs.List()
// walks a Go map, whose order is randomized per call, so "first element
// seen" would make the row -- and the stats identity -- flip run to run
// whenever element rows disagree. The lowest Unique_JobID (the array
// head, element 1) wins instead: every element of one array carries the
// same submit spec in production, so this only matters for the
// disagreement case, and there it always answers the same way.
carrierUID uint64
}
// deriveArrayJobStats scans store.Jobs and aggregates every array element row
@ -39,41 +48,30 @@ func deriveArrayJobStats() []models.ArrayJobInfo {
base := int64(backend.ExtractArrayBaseID(raw))
agg, ok := aggs[base]
if !ok {
agg = &arrayAgg{info: models.ArrayJobInfo{
JobID: base,
ArrayJobName: job.JobName,
SubmitUser: job.SubmitUser,
QueueName: job.QueueName,
UserGroup: job.UserGroup,
JobGroup: job.JobGroup,
JobDescription: job.JobDescription,
JobName: job.JobName,
ProjectName: job.ProjectName,
LicenseProject: job.LicenseProject,
SLA: job.SLA,
}}
agg = &arrayAgg{info: models.ArrayJobInfo{JobID: base}}
aggs[base] = agg
}
agg.info.NumJobs++
// Classify through JobStateStr so the counters always agree with the
// status string every other view shows (suspend states are stored as
// RUN|USUSP / RUN|SSUSP / PEND|PSUSP combinations there).
switch models.JobStateStr(job.JobStatus) {
case "USUSP":
agg.info.NumUSusp++
case "SSUSP":
agg.info.NumSSusp++
case "PSUSP":
agg.info.NumPSusp++
case "RUN":
agg.info.NumRun++
case "PEND":
agg.info.NumPend++
case "EXIT":
agg.info.NumExit++
case "DONE":
agg.info.NumDone++
if ok && raw >= agg.carrierUID {
// A non-head element: counters only, identity untouched.
agg.info.NumJobs++
aggCountStatus(agg, job.JobStatus)
continue
}
// First-seen (new agg) or a strictly lower-ID element: (re)supply
// the identity from this row.
agg.carrierUID = raw
agg.info.ArrayJobName = job.JobName
agg.info.SubmitUser = job.SubmitUser
agg.info.QueueName = job.QueueName
agg.info.UserGroup = job.UserGroup
agg.info.JobGroup = job.JobGroup
agg.info.JobDescription = job.JobDescription
agg.info.JobName = job.JobName
agg.info.ProjectName = job.ProjectName
agg.info.LicenseProject = job.LicenseProject
agg.info.SLA = job.SLA
agg.info.NumJobs++
aggCountStatus(agg, job.JobStatus)
}
out := make([]models.ArrayJobInfo, 0, len(aggs))
@ -84,6 +82,30 @@ func deriveArrayJobStats() []models.ArrayJobInfo {
return out
}
// aggCountStatus classifies one element row into the aggregate's counters
// (shared by the identity-carrier and counter-only branches above).
func aggCountStatus(agg *arrayAgg, status int) {
// Classify through JobStateStr so the counters always agree with the
// status string every other view shows (suspend states are stored as
// RUN|USUSP / RUN|SSUSP / PEND|PSUSP combinations there).
switch models.JobStateStr(status) {
case "USUSP":
agg.info.NumUSusp++
case "SSUSP":
agg.info.NumSSusp++
case "PSUSP":
agg.info.NumPSusp++
case "RUN":
agg.info.NumRun++
case "PEND":
agg.info.NumPend++
case "EXIT":
agg.info.NumExit++
case "DONE":
agg.info.NumDone++
}
}
// arrayStatsToOrderMaps converts derived stats rows to the OrderMap shape
// QueryData hands to ExecuteQuery (mirrors store.StoreToOrderMaps).
func arrayStatsToOrderMaps(rows []models.ArrayJobInfo) []utils.OrderMap {

23
src/frontend/apiserver/controllers/arraystats_test.go
View file

@ -84,23 +84,26 @@ func TestDeriveArrayJobStats_CountsPerStatus(t *testing.T) {
func TestDeriveArrayJobStats_IdentityFromFirstElement(t *testing.T) {
store.InitStores()
e := mkArrayElement(arrayStatsBase[0], 2, "RUN")
e.UserGroup = "ug1"
e.ProjectName = "proj"
store.Jobs.Put(uint64(e.Unique_JobID), e)
// A later element with different identity fields must not win (elements
// of one array share the submit spec; first-seen supplies the identity).
e2 := mkArrayElement(arrayStatsBase[0], 1, "DONE")
e2.SubmitUser = "bob"
// The identity carrier is the array HEAD (lowest element ID), no matter
// which order the rows were put or List() walks them in (map order is
// randomized). The head (element 1) carries identity fields that must
// win; element 2's different fields must NOT overwrite them.
e2 := mkArrayElement(arrayStatsBase[0], 2, "RUN")
e2.UserGroup = "ug2"
e2.ProjectName = "proj2"
store.Jobs.Put(uint64(e2.Unique_JobID), e2)
e := mkArrayElement(arrayStatsBase[0], 1, "DONE")
e.SubmitUser = "bob"
store.Jobs.Put(uint64(e.Unique_JobID), e)
rows := deriveArrayJobStats()
if len(rows) != 1 {
t.Fatalf("expected 1 array row, got %d", len(rows))
}
got := rows[0]
if got.SubmitUser != "alice" || got.UserGroup != "ug1" || got.ProjectName != "proj" {
t.Errorf("identity fields: user=%q ug=%q proj=%q, want alice/ug1/proj",
if got.SubmitUser != "bob" || got.UserGroup != "" || got.ProjectName != "" {
t.Errorf("identity fields: user=%q ug=%q proj=%q, want bob/// "+
"(the head element, not element 2, must supply the identity)",
got.SubmitUser, got.UserGroup, got.ProjectName)
}
}

7
src/include/persist.h
View file

@ -758,7 +758,12 @@ wal_commit_mode_t wal_classify(uint32_t wal_type, int detail);
* -XCE_ERR_NOT_READY — engine admission cap exhausted (retriable)
* timeout_ms = 0 falls back to p->config.sync_timeout_ms. n_peers <= 1
* returns 0 immediately (engine self-quorum, standalone preserved).
* Call only from vtask threads with no entity locks held. */
* Call only from vtask threads with no entity locks held.
*
* The async sibling persist_submit_job_payloads_async /
* persist_submit_job_array_async arms the same p->config.sync_timeout_ms
* deadline (commit_submit_async) -- the configured value, not a hardcoded
* constant, governs both paths' failure timing. */
int persist_commit_barrier(persist_t* p, uint32_t timeout_ms);
/* Force WAL fdatasync (for tests and clean shutdown) */