From 2a205291c0109f9c56624eb2d0804c9c170ecea1 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Wed, 12 Aug 2026 09:16:19 +0000 Subject: [PATCH] feat(component-failover-adoption): add apiserver's rejoin logic with correctly-placed cgo registration Registers a rejoin flag-setter (go_backend_rejoin_handler) as the client library's reconnect callback: zero I/O, sets an atomic flag, returns immediately -- avoiding self-deadlock on the library's per-ctx failover lock the same way worker's and scheduler's rejoin callbacks already do. The C shim, its extern declaration, and the xcecli_set_reconnect_handler() registration call all live together in clib.go's xce_backend_init(), placed after xcecli_start()'s success check (xcecli_start() wipes any earlier-registered handler as part of its own init -- the same ordering bug found and fixed in worker's/ scheduler's task 2.3/3.1 work). A shared checkAndPerformRejoin() helper wraps every gCtx.ctx-based backend call (SubmitJob, ControlJob, ControlQueue, ControlHost, FetchFile): after the call returns, if a rejoin is pending, issue xcecli_join() as a separate step and clear the flag only on success -- on failure it stays set so the next backend call's return retries it, no separate retry loop. needsRejoin is a proper sync/atomic flag rather than a plain bool, since apiserver (unlike the single-threaded C daemons) serves concurrent HTTP requests that can all reach the shared helper at once. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01NNB4zGPCsHyDUCDJyMbiQN --- .../component-failover-adoption/tasks.md | 3 +- src/frontend/apiserver/backend/clib.go | 82 +++-- src/frontend/apiserver/backend/reconnect.go | 150 +++++++++ .../apiserver/backend/reconnect_test.go | 292 ++++++++++++++++++ 4 files changed, 497 insertions(+), 30 deletions(-) create mode 100644 src/frontend/apiserver/backend/reconnect.go create mode 100644 src/frontend/apiserver/backend/reconnect_test.go diff --git a/.kiro/specs/component-failover-adoption/tasks.md b/.kiro/specs/component-failover-adoption/tasks.md index 318c2086..a3176870 100644 --- a/.kiro/specs/component-failover-adoption/tasks.md +++ b/.kiro/specs/component-failover-adoption/tasks.md @@ -101,7 +101,7 @@ - _Requirements: 1.1, 1.2, 1.3, 1.5, 1.7, 5.1, 5.2_ - _Depends: 4.1_ -- [ ] 4.3 Add apiserver's rejoin logic with a correctly-placed cgo registration and retry-on-failure +- [x] 4.3 Add apiserver's rejoin logic with a correctly-placed cgo registration and retry-on-failure - Add a flag-setter registered as the library's rejoin callback that does no I/O, only sets a pending-rejoin flag and returns immediately, and a separate function that performs the actual rejoin call - Register the flag-setter's C shim in the same file as its `extern` declaration (cgo symbol resolution is per-file) - Add one shared helper wrapping every backend call: after the call returns, check the pending-rejoin flag; if set, issue the rejoin as a separate step, clearing the flag only if that call succeeds — on its own failure, leave it set so the next backend call's return retries the rejoin @@ -170,5 +170,6 @@ - Task 3.2: implemented as a shared `sched_publish_retry_loop()`/`sched_publish_check_rejoin()` pair called from both `sched_publish_dispatch()` and `sched_publish_pend_reason()`, not literally inlined into `sched_publish_dispatch()` alone as design.md's prose names it — reviewed and judged compliant (same thread, fires once after the whole retry loop's successful return, not per-attempt, not in `sched_allocator`). Testing lesson for any future scheduler.c unit test needing to intercept `xcecli_*` calls: `xcecli_jobdispatch`/`xcecli_set_pend_reason`/`xcecli_join`/`xcecli_start`/`xcecli_set_reconnect_handler` all live in the single `client.c` translation unit inside `libxce.a` alongside symbols scheduler.c genuinely needs — the "test double linked ahead of the archive" trick (used by test_driver_decision_cb.c one layer up) causes duplicate-definition errors here. Use `-Wl,--wrap=` linker interposition instead (see `test_scheduler_decision_publish_retry.c`/its `meson.build` target) when a test needs scheduler.c itself linked live alongside intercepted libxce calls. Also confirmed: `sched_admin_request_test`'s `test_recordstart_symlink_escape_replies_error` fails independent of this spec's changes (its white-box `#include` list doesn't pull in `scheduler.c` at all) — a pre-existing, unrelated, environment-specific (root-in-sandbox symlink-permission) failure; do not re-investigate it as a regression in later tasks. - Task 4.1: first apiserver-group (Go+CGO) task. `xce_parse_peers()` alone (`src/lib/cluster.c`) does syntax parsing only — no semantic checks (accepts port 0, non-numeric port via silent `atoi()`-to-0, empty host). Worker/scheduler always pair it with a second `xce_cluster_config_validate()` call in their `main.c`s; round-1 review caught `parseCandidates()` (`candidates.go`) skipping that second call, silently accepting malformed candidates worker/scheduler reject — a reminder that "the C function I called is correct" isn't the same as "I called every C function the precedent calls." Fix: call `xce_cluster_config_validate(&cCfg)` after `xce_parse_peers()`, stamping a local placeholder into `cCfg.identity` first since that function also checks a non-empty top-level `identity` (unrelated to per-candidate validation, never propagated to the real connection identity elsewhere in `clib.go`). `xce_cluster_config_validate()` checks `hostname`/`port`/top-level `identity` only, never per-candidate `name` — so the `name=` prefix stays legitimately optional. Established convention confirmed pre-existing (not invented this task): a `//go:build cgo` test file with no `import "C"` of its own, pairing with a plain-Go-signature function in a separate cgo-preamble file (`_test.go` files can never `import "C"` themselves — a Go-toolchain restriction, not just a project style choice) — see `wire_dispatch_test.go`/`wire_dispatch_support.go` for the pre-existing precedent this task's `candidates_test.go`/`candidates.go` split follows. Task 4.1 is parsing-only by design — `RegisterToBackend()`'s connect loop and `tryConnect()` stay untouched; task 4.2 is the one that collapses them into a single connect call, so don't expect this diff to touch that territory. Behavior change (deliberate, disclosed, not a regression): a malformed entry now rejects the WHOLE candidate list (matches worker/scheduler's all-or-nothing precedent, `xce_parse_peers()` has no partial-success mode by construction) instead of the old Go parser's skip-with-warning leniency — surfaces via `RegisterToBackend`'s error → `cmd/main.go`'s `logger.Error`+`os.Exit(1)`, loud not silent. Tooling gap for future apiserver tasks (4.2-4.6): `make test-backend`'s documented CGO env-var setup doesn't actually export `CGO_CFLAGS`/`CGO_LDFLAGS` to a general shell (they only exist as meson `custom_target` env vars) — reconstruct via `pkg-config` against the deps `apiserver/meson.build` lists, or run the full `xce-build.sh apiserver asan` (rebuilds `libxce.a` fresh too) rather than a bare `go test`. - Task 4.2: found and fixed a genuine, independently-verified use-after-free while collapsing apiserver's two connect mechanisms into one — the deleted `tryConnect()` did `C.CString(host)` then `defer C.free(cAddr)` right after its single `xce_backend_init()` call returned, but `xcecli_ctx_init()` stores `ctx->conf = cfg` as a raw pointer (`src/lib/client.c:38`, never copied), and `client_failover.c`'s rotation path (`xcecli_retry_next`) re-dereferences `ac->conf->masters[cur]` on *every* failover attempt, not just at init — so any reconnect after the first would read freed heap memory. (`from`/`key` were NOT part of this bug — `xce_ctx_init()` `strdup()`s those synchronously before the defer ever fires.) Fix mirrors worker's/scheduler's own file-scope-`static` pattern for the Go/CGO side: two package-level `var`s (`gClusterCfg C.xce_cluster_config_t`, `gClientCfg C.xce_client_config_t`, alongside the pre-existing `gCtx`/`gIDCtx`) give the CGO structs a stable, process-lifetime address, and the connect path now writes host/from/key into fixed-size C arrays (`writeCCharArray()`) instead of `C.CString`/`C.free`-ing heap pointers the C library retains forever. General lesson for any future CGO code in this codebase that hands a struct to a C API storing a bare pointer to it (check the C function's own source, not just its header, for whether it copies or aliases): a Go function-local var is NOT safe once the call returns unless the C side genuinely copies everything reachable from it — verify via source read, the same standard task 2.1's original aliasing-hazard note set for the C-side daemons. `parseCandidates()` (task 4.1's deliverable) is now production-dead code — `buildClusterConfig()` calls `xce_parse_peers()` directly into the package-level `gClusterCfg` instead, since routing through `parseCandidates()`'s Go-slice return would mean re-parsing/re-copying into a fresh C struct that doesn't have the required stable address; `parseCandidates()` is exercised only by its own tests now, worth a follow-up cleanup task if it stays unused. +- Task 4.3: rejoin flag-setter (`go_backend_rejoin_handler`), its C shim, `extern` declaration, and `xcecli_set_reconnect_handler()` registration call all live together in `clib.go`'s `xce_backend_init()` — confirms design.md's "cgo symbol resolution is per-file" constraint (see `wire_dispatch_support.go`'s own documented rationale): a `C.` reference only resolves against an `extern` declared in THAT file's own cgo preamble, even though the Go `//export` mechanism makes the function callable package-wide — splitting the shim into a separate `reconnect.go` preamble while the extern lived in `clib.go` (or vice versa) would silently fail to link. Registration placed strictly AFTER `xcecli_start()`'s success check (task 2.3's ordering lesson ported to Go: `xcecli_start()` unconditionally wipes `rejoin_cb` to NULL as part of its own init). `needsRejoin` is a proper `sync/atomic` flag, not a plain bool — apiserver serves concurrent HTTP requests (unlike worker's/scheduler's single-threaded daemons), so this is a real correctness requirement here, not just style. `checkAndPerformRejoin()` wired into all 5 `gCtx.ctx`-based backend-call wrappers (SubmitJob/ControlJob/ControlQueue/ControlHost/FetchFile) — note `xcecli_fetchfile()` itself isn't `XCECLI_FAILOVER_CALL`-wrapped in libxce (never triggers the C-side rejoin invoke), so wrapping it here is safe but less load-bearing than the other four; that's a libxce gap, out of this task's `ApiserverBackendBridge` boundary to fix. Reviewer found (non-blocking, correctness-safe): concurrent goroutines can each observe `needsRejoin==1` after a failover and each call the real `xcecli_join()`, serialized by the C library's own per-ctx lock (no corruption, no incorrect flag state) but each can block up to the ~100s retry budget on a redundant join — a latency/traffic concern, not a Requirement-2 violation; a CAS-based single-flight guard is a reasonable fast-follow, not required. Scope reminder for task 4.4/4.5: `ApiserverBackendHealth`/`IsDegraded()`/`MarkHealthy()` deliberately do not exist yet — `checkAndPerformRejoin()` is the POST-call half only; task 4.5 adds a pre-call check to the same helper, not a new one. - Task 3.4: pure test-coverage task, zero production changes to `scheduler.c`/`driver.c` (correct outcome — task 3.2 already implemented the retry/shutdown/rejoin behavior correctly; 3.4 proves it holds at multi-decision QUEUE granularity through the real drain loop, which task 3.2's own tests never exercised, all 6 of which call `sched_publish_dispatch()`/`sched_publish_pend_reason()` directly on one decision). New `test_scheduler_decision_redelivery.c` links real `scheduler_srcs` with `-Wl,--wrap=xcecli_jobdispatch` (task 3.2's established technique) and drives the real `xce_sched_driver_on_decision()` (driver.c) over 3 queued, distinct-jobid decisions. Two load-bearing tests, both RED→GREEN-verified against the real code by two independent reviewer rounds (not just the implementer's claim): ordering (decision 1 stuck-then-succeeds fully completes before decision 2's call ever fires) and bounded shutdown latency (decisions queued behind one that triggers `core.stop` mid-retry get ZERO publish calls — the concrete, fast, deterministic proof that `sched_publish_retry_loop()`'s check-`stop`-before-each-attempt placement is correct, verified by reviewers moving the check to bottom-of-loop and confirming only this one test catches it). Test-fixture gotcha for any future test building a bare `xce_sched_core_t` via `xce_sched_core_init()` alone (not full `sched_init()`): do NOT call `xce_sched_core_fini()` on it — `fini()` calls `dataq_fini()` on `pend_updates[0/1]`, which only `sched_init()` (not `xce_sched_core_init()`) ever `dataq_init()`s, so `fini()` on a bare core SEGVs on a NULL dataq; omit the `fini()` call entirely, matching `test_core_events.c`'s established "fixtures aren't torn down" convention for this exact minimal-fixture pattern. - Task 3.3: fixed a real production gap found along the way — `sched_init()`'s two all-candidates-unreachable/join-failure log lines logged only a candidate COUNT, never host:port names (same gap worker's task 2.3 fixed for the worker). Added `sched_format_candidate_list()`, threading `sched->cluster_cfg` through as a parameter rather than a new file-scope global (unlike worker's `g_cluster_cfg`) since `test_scheduler_connect.c` links `scheduler_srcs` as separate objects, not `#include`, so the `static` helper isn't directly callable from tests — coverage requires a real `dup2()`-based stdout capture around a live `sched_init()` call hitting the failure path, not a white-box direct call (see `capture_stdout_during_sched_init()` in `test_scheduler_connect.c` — reusable pattern for any future scheduler.c log-content assertion). Round-1 review caught this diagnostic fix shipping with zero test coverage (proven by reverting the 59-line diff and rebuilding: all existing tests still passed) — a reminder that "the fix works, I verified it manually/via live logs" is not the same as a committed regression test; always add the assertion, don't just observe the correct behavior once. The worst-case-occupancy backpressure test (`test_scheduler_backpressure.c`) only proves Req 3 AC2 (buffering/block/resume-with-no-loss) — it does not exercise task 3.2's `-E_XCE_NOT_READY` retry path, so don't cite it for Req 7 coverage (that's task 3.4's job). E2E kill-primary/reconnect and all-unreachable-at-startup were validated manually against a real 3-node `/xce-deploy-ha` cluster rather than committed as an automated test, following worker's task 2.3 precedent — this repo has no automated live-multi-process-deploy E2E harness for any C daemon (the only one, `test_e2e_live_ha_push.py`, is Python/apiserver-only); reviewer judged this acceptable both rounds. Manual validation confirmed join, primary-death survival, and lazy socket-orphaning-on-idle (matches design), but could not observe a live rejoin firing against real sockets — blocked by bug-010 (`xcecd` ASAN "already joined thread" abort under `unshare -r`, needed to bypass the worker daemon's root-only check in this sandbox; unconfirmed whether it's a genuine bug or a userns artifact, never tried under real root). The exact rejoin-after-publish-success code path already has solid white-box mock coverage from task 3.2's `test_rejoin_composition_retries_across_publish_calls`, so this residual gap is narrowly "proven against mocks, not yet against a live socket" — worth closing opportunistically if bug-010 ever gets resolved, not a blocker for this spec. diff --git a/src/frontend/apiserver/backend/clib.go b/src/frontend/apiserver/backend/clib.go index be5a4827..f80cea93 100644 --- a/src/frontend/apiserver/backend/clib.go +++ b/src/frontend/apiserver/backend/clib.go @@ -26,14 +26,33 @@ static int xce_backend_handler(xce_request_t *req, void *arg) { return go_push_handler(req, arg); } +// go_backend_rejoin_handler is exported from Go (reconnect.go) and +// registered below as the client library's xcecli_reconnect_cb +// (component-failover-adoption task 4.3). It only sets a flag and returns +// immediately -- see reconnect.go's own doc comment for why (the library +// invokes it while still holding ac->failover.lock). +extern int go_backend_rejoin_handler(xcecli_ctx_t, void*); + +static int xce_backend_rejoin_handler(xcecli_ctx_t ctx, void *arg) { + return go_backend_rejoin_handler(ctx, arg); +} + // xce_backend_init initialises the client context, sets the handler, -// starts the network loop, then joins as HostRole_apiserver. +// starts the network loop, registers the rejoin callback, then joins as +// HostRole_apiserver. static int xce_backend_init(xce_client_config_t *config, backend_ctx_t *ctx) { int rc = xcecli_ctx_init(&(ctx->ctx), config); if (rc < 0) return rc; xcecli_set_handler(ctx->ctx, (xce_request_handler_t)xce_backend_handler, NULL); rc = xcecli_start(ctx->ctx); if (rc < 0) return rc; + // Registered AFTER xcecli_start() succeeds, not before + // (component-failover-adoption task 4.3; matches worker's/scheduler's + // task 2.3/3.1 fix): xcecli_start() unconditionally resets + // ctx->failover.rejoin_cb to NULL as part of initializing per-context + // failover state (client.c) -- registering before this call would be + // silently wiped out immediately afterward. + xcecli_set_reconnect_handler(ctx->ctx, xce_backend_rejoin_handler, NULL); return xcecli_join(ctx->ctx, HostRole_apiserver, NULL); } @@ -520,6 +539,7 @@ func SubmitJob(job *models.JobInfo, arraySize int) error { var outJobID C.uint64_t rc := C.xcecli_jobsubmit(gCtx.ctx, js, &outJobID) + checkAndPerformRejoin() if rc < 0 { return fmt.Errorf("xcecli_jobsubmit failed: %d", int(rc)) } @@ -555,6 +575,7 @@ func ControlJob(jobid uint64, action int, content string) error { } rc := C.xcecli_jobctrl(gCtx.ctx, &ctrl) + checkAndPerformRejoin() if rc < 0 { return fmt.Errorf("xcecli_jobctrl failed: %d", int(rc)) } @@ -573,6 +594,7 @@ func ControlQueue(queuename string, action int) error { } rc := C.xcecli_queuectrl(gCtx.ctx, &ctrl) + checkAndPerformRejoin() if rc < 0 { return fmt.Errorf("xcecli_queuectrl failed: %d", int(rc)) } @@ -590,6 +612,7 @@ func ControlHost(hostname string, action int) error { } rc := C.xcecli_hostctrl(gCtx.ctx, &ctrl) + checkAndPerformRejoin() if rc < 0 { return fmt.Errorf("xcecli_hostctrl failed: %d", int(rc)) } @@ -619,36 +642,37 @@ func signalNumber(signame string, action int) int { // controller→worker chain. The filepath must be the fully resolved absolute // path on the exec host. func FetchFile(jobid uint64, filepath string, offset uint64, maxBytes uint32) (*FetchFileResult, error) { -cFilepath := C.CString(filepath) -defer C.free(unsafe.Pointer(cFilepath)) + cFilepath := C.CString(filepath) + defer C.free(unsafe.Pointer(cFilepath)) -msg := C.xce_fetchfile_t{ -jobid: C.uint64_t(jobid), -filepath: cFilepath, -offset: C.uint64_t(offset), -max_bytes: C.uint32_t(maxBytes), -} + msg := C.xce_fetchfile_t{ + jobid: C.uint64_t(jobid), + filepath: cFilepath, + offset: C.uint64_t(offset), + max_bytes: C.uint32_t(maxBytes), + } -var reply *C.xce_fetchfile_reply_t -rc := C.xcecli_fetchfile(gCtx.ctx, &msg, &reply) -if rc < 0 { -return nil, fmt.Errorf("xcecli_fetchfile failed: %d", int(rc)) -} -if reply == nil { -return nil, fmt.Errorf("xcecli_fetchfile returned nil reply") -} -defer C.free(unsafe.Pointer(reply)) + var reply *C.xce_fetchfile_reply_t + rc := C.xcecli_fetchfile(gCtx.ctx, &msg, &reply) + checkAndPerformRejoin() + if rc < 0 { + return nil, fmt.Errorf("xcecli_fetchfile failed: %d", int(rc)) + } + if reply == nil { + return nil, fmt.Errorf("xcecli_fetchfile returned nil reply") + } + defer C.free(unsafe.Pointer(reply)) -if reply.error != 0 { -return nil, fmt.Errorf("fetchfile remote error: %d", int(reply.error)) -} + if reply.error != 0 { + return nil, fmt.Errorf("fetchfile remote error: %d", int(reply.error)) + } -result := &FetchFileResult{ -FileSize: uint64(reply.file_size), -BytesRead: uint32(reply.bytes_read), -} -if reply.bytes_read > 0 && reply.content != nil { -result.Content = C.GoBytes(unsafe.Pointer(reply.content), C.int(reply.bytes_read)) -} -return result, nil + result := &FetchFileResult{ + FileSize: uint64(reply.file_size), + BytesRead: uint32(reply.bytes_read), + } + if reply.bytes_read > 0 && reply.content != nil { + result.Content = C.GoBytes(unsafe.Pointer(reply.content), C.int(reply.bytes_read)) + } + return result, nil } diff --git a/src/frontend/apiserver/backend/reconnect.go b/src/frontend/apiserver/backend/reconnect.go new file mode 100644 index 00000000..e529ea2f --- /dev/null +++ b/src/frontend/apiserver/backend/reconnect.go @@ -0,0 +1,150 @@ +//go:build cgo + +package backend + +/* +#include "client.h" +#include "controller.capnp.h" +*/ +import "C" + +import ( + logger "apiserver/utils" + "sync/atomic" + "unsafe" + + "go.uber.org/zap" +) + +// reconnect.go implements apiserver's rejoin-after-transparent-reconnect +// logic (component-failover-adoption task 4.3, boundary: +// ApiserverBackendBridge, Req 2 AC1-3/AC5). See design.md's "Rejoin +// callback constraint" section and clib.go's xce_backend_init() for the +// C-side shim/extern/registration this file's exported flag-setter is +// wired to. Per that shim's own doc comment: cgo symbol resolution is +// per-file (see wire_dispatch_support.go's documented rationale for the +// identical go_push_handler precedent), so the shim/extern/registration +// must live in clib.go, not here -- this file holds only the plain Go +// logic (Go-level //export functions are visible package-wide once +// generated, unlike the raw C shim/extern that only resolves within the +// C translation unit of the file that declares it). +// +// Concurrency note: unlike worker's/scheduler's single-threaded C daemons +// (where a plain bool is safe), apiserver serves concurrent HTTP requests +// -- SubmitJob/ControlJob/ControlQueue/ControlHost/FetchFile can all run on +// different goroutines simultaneously. needsRejoin is therefore a proper +// sync/atomic int32, the same pattern already used by heartbeatIntervalSecs +// (heartbeat.go), not a plain bool. + +// needsRejoin flags that the client library has transparently reconnected +// the underlying transport and apiserver must re-issue xcecli_join() as a +// separate, later call before it is a cluster member again. 0 = no rejoin +// pending, 1 = rejoin pending. Set by setNeedsRejoin() (invoked from the +// library while its own lock is held); cleared only once performRejoinFn() +// itself succeeds (Req 2 AC5) -- never merely because it was attempted. +var needsRejoin int32 + +// setNeedsRejoin is the flag-setting logic behind the exported +// go_backend_rejoin_handler below, factored out into a plain-Go-typed +// function so it is directly callable (and independently timeable) from +// reconnect_test.go: a _test.go file can never `import "C"` itself +// (golang/go#13938), so it cannot call go_backend_rejoin_handler directly +// (its signature involves C.int) -- same clib.go/clib_test.go split this +// package already established for currentClusterCandidates()/ +// currentClientConfigNMasters(). +func setNeedsRejoin() { + atomic.StoreInt32(&needsRejoin, 1) +} + +// go_backend_rejoin_handler is the flag-setter registered as the client +// library's xcecli_reconnect_cb, via the C shim (xce_backend_rejoin_handler) +// and xcecli_set_reconnect_handler() call that live in clib.go's +// xce_backend_init() (see that function's own doc comment for why +// registration happens AFTER xcecli_start() succeeds, not before). +// +// The client library invokes this SYNCHRONOUSLY, on the calling goroutine, +// while still holding its own per-ctx lock (ac->failover.lock, a +// non-recursive mutex held for the whole XCECLI_FAILOVER_CALL retry loop -- +// confirmed directly against client_failover.c:341-411, +// xcecli_invoke_rejoin() at line 397). Per design.md's "Rejoin callback +// constraint", this function performs NO I/O and calls NO xcecli_* +// function: doing so would try to re-acquire the same already-held lock +// and self-deadlock the calling goroutine forever. It only sets a flag +// (via setNeedsRejoin) and returns 0 immediately so the interrupted +// operation resumes normally -- the actual rejoin happens later, as a +// separate call, once the triggering call has returned and the lock is +// free (see performPendingRejoin/checkAndPerformRejoin below). +// +//export go_backend_rejoin_handler +func go_backend_rejoin_handler(ctx unsafe.Pointer, arg unsafe.Pointer) C.int { + setNeedsRejoin() + return 0 +} + +// performPendingRejoin issues the actual xcecli_join() call: the same join +// payload used at startup (xce_backend_init, clib.go), as a separate, +// subsequent call once the triggering xcecli_* call has already returned +// and the client library's internal failover lock is free. +// +// Returns the raw rc as a plain Go int (not C.int) so it can be assigned +// to performRejoinFn -- a pure-Go-typed function variable tests override +// with a controllable double, avoiding the need for a live controller or +// a -Wl,--wrap=-style linker interposition to exercise retry-on-failure. +// +// Does not loop/retry on its own failure: retry-on-failure is achieved by +// leaving needsRejoin set so the NEXT backend call's post-call check +// (checkAndPerformRejoin) retries it -- the same "compose with the +// existing per-call cadence, no separate timer" pattern worker's +// checking_connection()/rejoin_call and scheduler's task 3.2 rejoin-check +// composition both already use. +// +// Success is rc >= 0, NOT rc == 0 (matches worker's/scheduler's +// established convention, tasks.md Implementation Notes for task 2.3): a +// successful join returns the controller's reply.status verbatim, which +// can be a non-negative assigned slot/identity value, not always exactly +// 0. Only a genuine failure reply.status is negative. +func performPendingRejoin() int { + rc := C.xcecli_join(gCtx.ctx, C.int(C.HostRole_apiserver), nil) + return int(rc) +} + +// performRejoinFn defaults to the real production call +// (performPendingRejoin above). Tests override it with a controllable +// double so checkAndPerformRejoin()'s retry-until-success composition can +// be verified deterministically -- mirrors worker's identical rejoin_call +// function-pointer testability pattern (controller_connector.c's +// perform_pending_rejoin/rejoin_call). +var performRejoinFn = performPendingRejoin + +// checkAndPerformRejoin is the shared post-call helper every backend-call +// wrapper (SubmitJob, ControlJob, ControlQueue, ControlHost, FetchFile -- +// see clib.go) calls once its own underlying C.xcecli_* call has returned +// (Req 2 AC1-3, AC5): if a rejoin is pending, issue it as a separate step, +// clearing the pending indicator only if that call succeeds. On its own +// failure, the indicator stays set so the next backend call's return +// retries it -- never abandoned. +// +// Runs regardless of the triggering call's own outcome (mirrors worker's +// checking_connection(): a transparent reconnect can be pending +// independent of whether this particular call happened to succeed). +// +// This is deliberately only the POST-call half of what design.md describes +// as the eventual shared helper. Task 4.5 (ApiserverBackendHealth / +// IsDegraded()) will add a PRE-call fail-fast check for the same wrapper +// call sites -- that circuit-breaker state does not exist yet, so there is +// nothing to check before the call in this task. +func checkAndPerformRejoin() { + if atomic.LoadInt32(&needsRejoin) == 0 { + return + } + + rc := performRejoinFn() + if rc >= 0 { + atomic.StoreInt32(&needsRejoin, 0) + logger.Info("checkAndPerformRejoin: rejoin after transparent reconnect succeeded") + return + } + + logger.Warn("checkAndPerformRejoin: rejoin after transparent reconnect failed; will retry on next backend call", + zap.Int("rc", rc)) +} diff --git a/src/frontend/apiserver/backend/reconnect_test.go b/src/frontend/apiserver/backend/reconnect_test.go new file mode 100644 index 00000000..0fbc5b11 --- /dev/null +++ b/src/frontend/apiserver/backend/reconnect_test.go @@ -0,0 +1,292 @@ +//go:build cgo + +package backend + +import ( + "os" + "strings" + "sync/atomic" + "testing" + "time" +) + +// reconnect_test.go exercises reconnect.go's rejoin flag-setter, +// rejoin-performing function, and shared post-call-check helper +// (component-failover-adoption, task 4.3, boundary: ApiserverBackendBridge, +// Req 2 AC1-3/AC5). Kept //go:build cgo since these symbols only exist in +// the CGO build -- this file itself does not `import "C"` (a _test.go file +// can never do that directly, golang/go#13938): it only calls the +// plain-Go-signature functions defined in reconnect.go (setNeedsRejoin, +// checkAndPerformRejoin, performRejoinFn), the same clib.go/clib_test.go +// and candidates.go/candidates_test.go split already established in this +// package. +// +// Test strategy: +// - The flag-setter's "no I/O, returns immediately" contract (Req 2 AC1, +// design.md's "Rejoin callback constraint" -- the library invokes it +// while still holding ac->failover.lock, so any I/O or xcecli_* call +// there self-deadlocks) is proven two ways, mirroring +// test_scheduler_connect.c's identical reasoning for the C-side +// rejoin callback (see that file's own header comment): a +// source-text scan confirming go_backend_rejoin_handler's body makes +// no xcecli_*/blocking call, and a timing-based fast-return assertion +// on setNeedsRejoin (the plain-Go-typed logic go_backend_rejoin_handler +// delegates to -- go_backend_rejoin_handler itself can't be called +// directly from this file since its signature involves C.int). +// - Retry-on-failure (Req 2 AC5) is exercised by overriding +// performRejoinFn with a scripted double across two separate +// checkAndPerformRejoin() calls (simulating two separate +// backend-call-wrapper invocations) -- no live controller or linker +// interposition needed, mirrors worker's rejoin_call function-pointer +// testability pattern (controller_connector.c). +// - The "correctly-placed cgo registration" structural claim (this +// task's own title) is a source-text scan proving the C shim/extern/ +// registration all live in clib.go, not reconnect.go, and that +// registration happens after xcecli_start()'s success check -- same +// genre as clib_test.go's own TestClibSource_* structural regression +// tests. + +func resetNeedsRejoinForTest() { + atomic.StoreInt32(&needsRejoin, 0) +} + +func readSourceForTest(t *testing.T, filename string) string { + t.Helper() + src, err := os.ReadFile(filename) + if err != nil { + t.Fatalf("could not read %s source for scan: %v", filename, err) + } + return string(src) +} + +// stripGoLineComments removes everything from the first "//" to end-of-line +// on every line, for source-text scans that must not be tripped by a +// doc comment merely mentioning a symbol name in prose. Deliberately +// simplistic (no string-literal awareness) -- sufficient for scanning this +// package's own source, which contains no "//" inside a string literal on +// a line this test cares about. +func stripGoLineComments(src string) string { + lines := strings.Split(src, "\n") + for i, line := range lines { + if idx := strings.Index(line, "//"); idx >= 0 { + lines[i] = line[:idx] + } + } + return strings.Join(lines, "\n") +} + +// ════════════════════════════════════════════════════════════════════════ +// Req 2 AC1 -- flag-setter performs zero I/O and returns immediately, even +// conceptually "while the client's lock is held" +// ════════════════════════════════════════════════════════════════════════ + +func TestSetNeedsRejoin_FastReturnAndSetsFlag(t *testing.T) { + resetNeedsRejoinForTest() + + start := time.Now() + setNeedsRejoin() + elapsed := time.Since(start) + + // Generous bound for a flag-only, no-I/O operation -- anything even + // approaching millisecond scale would suggest a blocking call snuck in. + if elapsed > 10*time.Millisecond { + t.Errorf("setNeedsRejoin took %v -- too slow for a flag-only "+ + "operation, suggests I/O or blocking was introduced", elapsed) + } + if atomic.LoadInt32(&needsRejoin) != 1 { + t.Error("expected needsRejoin to be set after setNeedsRejoin()") + } +} + +func TestReconnectSource_FlagSetterBodyHasNoXcecliCallsOrIO(t *testing.T) { + src := readSourceForTest(t, "reconnect.go") + + start := strings.Index(src, "func go_backend_rejoin_handler(") + if start < 0 { + t.Fatal("go_backend_rejoin_handler not found in reconnect.go") + } + end := strings.Index(src[start:], "\n}\n") + if end < 0 { + t.Fatal("could not find end of go_backend_rejoin_handler's body") + } + body := src[start : start+end] + + for _, forbidden := range []string{"C.xcecli_", "net.Dial", "os.Open", "time.Sleep", "<-"} { + if strings.Contains(body, forbidden) { + t.Errorf("go_backend_rejoin_handler body contains %q -- the "+ + "library invokes this callback while still holding "+ + "ac->failover.lock (client_failover.c:341-411); any I/O "+ + "or xcecli_* call here self-deadlocks", forbidden) + } + } +} + +// ════════════════════════════════════════════════════════════════════════ +// Req 2 AC5 -- rejoin retry-on-failure: retried on the NEXT backend call, +// never abandoned +// ════════════════════════════════════════════════════════════════════════ + +func TestCheckAndPerformRejoin_NoPendingRejoin_DoesNothing(t *testing.T) { + resetNeedsRejoinForTest() + orig := performRejoinFn + defer func() { performRejoinFn = orig }() + + called := false + performRejoinFn = func() int { + called = true + return 0 + } + + checkAndPerformRejoin() + + if called { + t.Error("performRejoinFn must not be called when no rejoin is pending") + } +} + +func TestCheckAndPerformRejoin_PendingRejoinSucceeds_ClearsFlag(t *testing.T) { + captureLog(t) // checkAndPerformRejoin logs via utils.Log; nil otherwise + atomic.StoreInt32(&needsRejoin, 1) + orig := performRejoinFn + defer func() { performRejoinFn = orig }() + + performRejoinFn = func() int { return 0 } // rc >= 0 => success + + checkAndPerformRejoin() + + if atomic.LoadInt32(&needsRejoin) != 0 { + t.Error("expected needsRejoin to clear after a successful rejoin") + } +} + +func TestCheckAndPerformRejoin_NonZeroSuccessStillClearsFlag(t *testing.T) { + // Matches worker's/scheduler's established convention: a successful + // join can return a non-negative assigned slot/identity value, not + // always exactly 0 -- rc < 0 is the only correct failure test. + captureLog(t) + atomic.StoreInt32(&needsRejoin, 1) + orig := performRejoinFn + defer func() { performRejoinFn = orig }() + + performRejoinFn = func() int { return 7 } + + checkAndPerformRejoin() + + if atomic.LoadInt32(&needsRejoin) != 0 { + t.Error("expected needsRejoin to clear on a non-zero-but-non-negative rc") + } +} + +func TestCheckAndPerformRejoin_RetriedOnNextBackendCall_NotAbandoned(t *testing.T) { + captureLog(t) + atomic.StoreInt32(&needsRejoin, 1) + orig := performRejoinFn + defer func() { performRejoinFn = orig }() + + calls := 0 + performRejoinFn = func() int { + calls++ + if calls == 1 { + return -1 // simulates -E_XCE_NOT_READY: this attempt's own retry budget exhausted + } + return 3 // succeeds on the second, separate attempt + } + + // First "backend call" wrapper invocation: rejoin attempted, fails -- + // the flag must stay set, not be cleared merely because it was + // attempted. + checkAndPerformRejoin() + if atomic.LoadInt32(&needsRejoin) != 1 { + t.Fatal("expected needsRejoin to remain set after a failed rejoin attempt") + } + if calls != 1 { + t.Fatalf("expected exactly 1 rejoin attempt so far, got %d", calls) + } + + // Second, separate "backend call" wrapper invocation (simulating the + // NEXT backend call, not a retry loop inside the first): rejoin is + // retried and succeeds, flag clears. + checkAndPerformRejoin() + if atomic.LoadInt32(&needsRejoin) != 0 { + t.Error("expected needsRejoin to clear once the retried rejoin succeeds") + } + if calls != 2 { + t.Errorf("expected exactly 2 rejoin attempts total (one per backend call), got %d", calls) + } +} + +// ════════════════════════════════════════════════════════════════════════ +// This task's own title -- correctly-placed cgo registration +// ════════════════════════════════════════════════════════════════════════ + +func TestClibSource_RejoinShimExternAndRegistrationLiveInClibGo(t *testing.T) { + clibSrc := readSourceForTest(t, "clib.go") + + if !strings.Contains(clibSrc, "extern int go_backend_rejoin_handler(") { + t.Error("clib.go must declare `extern int go_backend_rejoin_handler(...)` " + + "in its own cgo preamble -- cgo symbol resolution is per-file, " + + "so a C. reference (here, the shim calling the Go export) " + + "only resolves against an extern declared in the SAME file") + } + if !strings.Contains(clibSrc, "static int xce_backend_rejoin_handler(") { + t.Error("clib.go must define the C shim `xce_backend_rejoin_handler` " + + "that wraps go_backend_rejoin_handler, mirroring the existing " + + "xce_backend_handler/go_push_handler pattern") + } + if !strings.Contains(clibSrc, "xcecli_set_reconnect_handler(") { + t.Error("clib.go must call xcecli_set_reconnect_handler(...) to " + + "register the rejoin shim with the client library") + } + + // Code-only view of reconnect.go (comment lines stripped): reconnect.go's + // own doc comments legitimately MENTION xce_backend_rejoin_handler and + // xcecli_set_reconnect_handler in prose (explaining where they live and + // why) -- that must not trip this check. Only actual C code (the shim + // definition or the registration call) is disallowed here. + reconnectCode := stripGoLineComments(readSourceForTest(t, "reconnect.go")) + if strings.Contains(reconnectCode, "xce_backend_rejoin_handler") { + t.Error("the C shim (xce_backend_rejoin_handler) must live in " + + "clib.go, not reconnect.go -- cgo symbol resolution is per-file") + } + if strings.Contains(reconnectCode, "xcecli_set_reconnect_handler(") { + t.Error("the xcecli_set_reconnect_handler(...) registration call " + + "must live in clib.go, not reconnect.go") + } + + // Ordering: registration happens AFTER xcecli_start()'s success check, + // not before -- xcecli_start() (client.c) unconditionally resets + // ctx->failover.rejoin_cb to NULL as part of initializing per-context + // failover state, so registering before it would be silently wiped + // (tasks.md Implementation Notes, task 2.3's identical worker bug). + startIdx := strings.Index(clibSrc, "rc = xcecli_start(ctx->ctx);") + registerIdx := strings.Index(clibSrc, "xcecli_set_reconnect_handler(ctx->ctx") + if startIdx < 0 { + t.Fatal("expected `rc = xcecli_start(ctx->ctx);` call site not found in clib.go") + } + if registerIdx < 0 { + t.Fatal("expected xcecli_set_reconnect_handler(ctx->ctx, ...) call site not found in clib.go") + } + if registerIdx < startIdx { + t.Error("xcecli_set_reconnect_handler must be registered AFTER " + + "xcecli_start(), not before -- xcecli_start() wipes any " + + "handler registered earlier") + } +} + +// ════════════════════════════════════════════════════════════════════════ +// Shared helper wraps every backend call: structural check that clib.go's +// wrapper functions actually route through checkAndPerformRejoin, not a +// duplicated per-wrapper copy of its logic. +// ════════════════════════════════════════════════════════════════════════ + +func TestClibSource_BackendCallWrappersCallSharedRejoinHelper(t *testing.T) { + clibSrc := readSourceForTest(t, "clib.go") + + count := strings.Count(clibSrc, "checkAndPerformRejoin()") + if count < 4 { + t.Errorf("expected at least 4 call sites of checkAndPerformRejoin() "+ + "in clib.go (SubmitJob/ControlJob/ControlQueue/ControlHost at "+ + "minimum), got %d -- the shared post-call helper must not be "+ + "duplicated or skipped per wrapper", count) + } +}