From ecf5eddc61095b47df38a368f06a8abe2f8f9227 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Wed, 12 Aug 2026 09:58:27 +0000 Subject: [PATCH] feat(component-failover-adoption): trigger apiserver's rejoin from the idle-period heartbeat path xcecli_heartbeat() is XCECLI_FF_CALL-wrapped -- a deliberately non-retrying fire-and-forget path that does its own best-effort one-shot reconnect but never calls xcecli_invoke_rejoin() (that only fires from the write-call retry path). Left as-is, a controller failover with no in-flight write request would silently repair the transport connection without ever re-joining, permanently starving push notifications. runHeartbeat() gains two callback hooks fired at the right points -- onFailureToSuccess on a genuine transition, onSuccessTick on every successful tick thereafter -- wired directly to task 4.3's existing setNeedsRejoin()/checkAndPerformRejoin() from StartHeartbeat(). No second rejoin mechanism: checkAndPerformRejoin() is a cheap no-op when nothing is pending, so calling it unconditionally on every successful tick is safe and matches the requirement that retry isn't gated on the transition edge alone. 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 | 13 +- src/frontend/apiserver/backend/heartbeat.go | 53 +++++- .../backend/heartbeat_rejoin_test.go | 132 +++++++++++++ .../apiserver/backend/heartbeat_test.go | 174 +++++++++++++++++- 5 files changed, 363 insertions(+), 12 deletions(-) create mode 100644 src/frontend/apiserver/backend/heartbeat_rejoin_test.go diff --git a/.kiro/specs/component-failover-adoption/tasks.md b/.kiro/specs/component-failover-adoption/tasks.md index a3176870..827c23c3 100644 --- a/.kiro/specs/component-failover-adoption/tasks.md +++ b/.kiro/specs/component-failover-adoption/tasks.md @@ -109,7 +109,7 @@ - _Requirements: 2.1, 2.2, 2.3, 2.4, 2.5_ - _Depends: 4.2_ -- [ ] 4.4 Trigger apiserver's rejoin from the idle-period heartbeat path +- [x] 4.4 Trigger apiserver's rejoin from the idle-period heartbeat path - On a transport failure-to-success transition detected by the heartbeat prober, set the same pending-rejoin flag the write path uses (the heartbeat's own probe mechanism never sets it itself) - On every subsequent heartbeat tick where the probe succeeds and a rejoin is still pending, retry the rejoin — checked every tick, not only on the transition edge, so a rejoin attempt that itself fails still gets retried on a later tick - Do not gate this retry on the separate backend-degraded circuit-breaker state, since that state clears itself on its own timer independent of whether a rejoin actually succeeded @@ -171,5 +171,6 @@ - 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 4.4: thin adapter, not a second rejoin mechanism — `runHeartbeat()`'s new `onFailureToSuccess`/`onSuccessTick` callback params are wired directly to task 4.3's real `setNeedsRejoin`/`checkAndPerformRejoin` (`reconnect.go`) from `StartHeartbeat()`, zero duplicated flag-check logic added in `heartbeat.go` itself. Necessary because `xcecli_heartbeat()` is `XCECLI_FF_CALL`-wrapped (`client_internal.h`), a deliberately non-retrying fire-and-forget path that does its own best-effort one-shot reconnect but NEVER calls `xcecli_invoke_rejoin()` (that only fires from `xcecli_retry_next`, the write-call retry path) — so nothing sets `needsRejoin` on a pure idle-period recovery unless the heartbeat loop does it itself. `rc >= 0` success classification for `xcecli_heartbeat` confirmed against `xceapi.c`/`request.c`: it only ever returns 0 or a negative errno, never a positive value, so this is equivalent to (not just consistent with) the `rc<0`-only-failure convention used elsewhere. `onSuccessTick` fires unconditionally on every successful tick (no pending-check duplicated in `heartbeat.go`) — safe because `checkAndPerformRejoin()` is itself a cheap single-atomic-load no-op when nothing is pending. `prevFailed` (transition-tracking state) is correctly a plain non-atomic local, not `sync/atomic` — verified `runHeartbeat()` has exactly one call site (`StartHeartbeat()`, called once from `cmd/main.go`), one goroutine for the process lifetime; contrast with `needsRejoin` itself, which stays atomic since concurrent HTTP-handler goroutines reach it via the write-path wrappers. Known, accepted, already-documented (task 4.3) tradeoff carried forward: a rejoin attempt from the heartbeat goroutine shares `ac->failover.lock` with concurrent HTTP-handler backend calls, so one can block the other for up to the ~1-2 minute retry budget — not a new issue this task introduces, same mechanism task 4.3 already reviewed and accepted, task 4.4 just adds one more caller of it. - 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 f80cea93..adcda94f 100644 --- a/src/frontend/apiserver/backend/clib.go +++ b/src/frontend/apiserver/backend/clib.go @@ -349,10 +349,17 @@ func Cleanup() { // StartHeartbeat launches a goroutine that sends periodic xcecli_heartbeat // keep-alive messages to the controller. Stops when ctx is cancelled. +// +// Wires runHeartbeat's failure-to-success and every-successful-tick hooks to +// reconnect.go's setNeedsRejoin/checkAndPerformRejoin (task 4.3) -- closing +// the idle-period reconnect gap (task 4.4, Req 2 AC4/AC5): XCECLI_FF_CALL +// (backing xcecli_heartbeat) never invokes the client library's rejoin +// callback, so nothing else would set the pending-rejoin flag if this +// heartbeat prober is the only traffic during a controller outage. func StartHeartbeat(ctx context.Context) { - go runHeartbeat(ctx, func(secs int) { - C.xcecli_heartbeat(gCtx.ctx, C.int(secs)) - }) + go runHeartbeat(ctx, func(secs int) int { + return int(C.xcecli_heartbeat(gCtx.ctx, C.int(secs))) + }, setNeedsRejoin, checkAndPerformRejoin) } // go_push_handler is called by the C library on every incoming request. diff --git a/src/frontend/apiserver/backend/heartbeat.go b/src/frontend/apiserver/backend/heartbeat.go index fff98c78..baa574fb 100644 --- a/src/frontend/apiserver/backend/heartbeat.go +++ b/src/frontend/apiserver/backend/heartbeat.go @@ -18,18 +18,63 @@ func SetHeartbeatInterval(secs int) { } } -// runHeartbeat loops until ctx is cancelled, calling fn(intervalSecs) on each tick. -// Extracted from clib.go for testability without CGO. -func runHeartbeat(ctx context.Context, fn func(int)) { +// runHeartbeat loops until ctx is cancelled, calling fn(intervalSecs) on each +// tick and classifying its returned rc as success (rc >= 0) or failure +// (rc < 0) -- xcecli_heartbeat's own convention: it is XCECLI_FF_CALL-wrapped +// (client_internal.h) and yields xce_heartbeat's rc verbatim, which is +// either 0 (not yet due) or xce_nop's underlying send rc (xceapi.c) -- +// consistent with this codebase's established "rc < 0 is the only failure" +// join/rejoin convention (tasks.md Implementation Notes, task 2.3). +// +// component-failover-adoption task 4.4 (boundary: ApiserverBackendBridge, +// Req 2 AC4/AC5): closes the idle-period reconnect gap. XCECLI_FF_CALL -- +// the macro backing xcecli_heartbeat -- performs a best-effort transport +// reconnect on a known-dead connection but never invokes the client +// library's rejoin callback (xcecli_invoke_rejoin is reached only from +// xcecli_retry_next, the full-retry write path task 4.3 wired up) -- so +// nothing else sets the pending-rejoin flag if this heartbeat prober is the +// ONLY traffic during a controller outage (apiserver idle, no in-flight +// write request). +// +// Tracks success/failure across ticks in a single local variable -- +// single-goroutine state, no atomic needed (unlike reconnect.go's +// needsRejoin, which is genuinely shared across concurrent HTTP request +// goroutines). On a genuine failure-to-success TRANSITION (previous tick +// failed, this tick succeeded), onFailureToSuccess is called once -- +// wired to reconnect.go's setNeedsRejoin in production. On EVERY tick that +// succeeds -- not only the transition tick -- onSuccessTick is called, +// wired to checkAndPerformRejoin (task 4.3), which is itself a cheap no-op +// once no rejoin is pending: this is what makes a rejoin attempt that +// itself fails get retried on the next successful tick rather than +// abandoned (Req 2 AC5). A failed tick does nothing beyond updating the +// local failure-tracking state -- no rejoin logic runs on a failed tick. +// +// Both callbacks are injected (rather than referenced by name) so this +// loop's own tick/transition-tracking logic stays testable without CGO -- +// see heartbeat_test.go for that coverage and heartbeat_rejoin_test.go +// (//go:build cgo) for the end-to-end wiring against the real +// setNeedsRejoin/checkAndPerformRejoin. clib.go's StartHeartbeat wires the +// real production callbacks (those two symbols only exist in the CGO +// build). +func runHeartbeat(ctx context.Context, fn func(int) int, onFailureToSuccess, onSuccessTick func()) { interval := int(atomic.LoadInt32(&heartbeatIntervalSecs)) t := time.NewTimer(time.Duration(interval) * time.Second) defer t.Stop() + + prevFailed := false for { select { case <-ctx.Done(): return case <-t.C: - fn(interval) + succeeded := fn(interval) >= 0 + if succeeded { + if prevFailed { + onFailureToSuccess() + } + onSuccessTick() + } + prevFailed = !succeeded } interval = int(atomic.LoadInt32(&heartbeatIntervalSecs)) t.Reset(time.Duration(interval) * time.Second) diff --git a/src/frontend/apiserver/backend/heartbeat_rejoin_test.go b/src/frontend/apiserver/backend/heartbeat_rejoin_test.go new file mode 100644 index 00000000..35712cbc --- /dev/null +++ b/src/frontend/apiserver/backend/heartbeat_rejoin_test.go @@ -0,0 +1,132 @@ +//go:build cgo + +package backend + +import ( + "context" + "sync/atomic" + "testing" + "time" +) + +// heartbeat_rejoin_test.go exercises runHeartbeat wired to reconnect.go's +// REAL setNeedsRejoin/checkAndPerformRejoin (component-failover-adoption +// task 4.4, boundary: ApiserverBackendBridge, Req 2 AC4/AC5) -- proving the +// idle-period rejoin trigger end-to-end at real tick granularity, the same +// way clib.go's StartHeartbeat wires them in production. heartbeat_test.go +// covers runHeartbeat's own tick/transition-tracking logic via generic +// injected callbacks, without CGO; this file proves those callbacks are +// wired to the right production functions and that the composition +// actually clears/retains needsRejoin correctly across real ticks. +// +// Kept //go:build cgo since needsRejoin/setNeedsRejoin/checkAndPerformRejoin/ +// performRejoinFn only exist in the CGO build (reconnect.go, task 4.3) -- +// this file itself does not `import "C"` (a _test.go file can never do that +// directly, golang/go#13938), matching the reconnect_test.go/candidates_test.go +// precedent already established in this package. +// +// performRejoinFn (reconnect.go's existing test seam, task 4.3) is +// overridden here to script the actual xcecli_join outcome without a live +// controller -- same technique reconnect_test.go's own retry-on-failure +// tests use. scriptedHeartbeatFn (heartbeat_test.go) scripts the heartbeat +// send outcomes. + +func TestRunHeartbeat_IdlePeriodRecovery_SetsNeedsRejoinAndRetriesUntilJoinSucceeds(t *testing.T) { + captureLog(t) // checkAndPerformRejoin logs via utils.Log; nil otherwise + resetNeedsRejoinForTest() + atomic.StoreInt32(&heartbeatIntervalSecs, 1) + + origPerformRejoinFn := performRejoinFn + defer func() { performRejoinFn = origPerformRejoinFn }() + + var joinCalls int32 + joinDone := make(chan struct{}, 8) + performRejoinFn = func() int { + n := atomic.AddInt32(&joinCalls, 1) + joinDone <- struct{}{} + if n == 1 { + return -1 // simulates -E_XCE_NOT_READY: this attempt's own retry budget exhausted + } + return 0 // succeeds on the second, separate attempt + } + + // Heartbeat transport outcomes: fail (primary down), fail, succeed + // (transparent reconnect -- fires the transition), succeed (retry + // tick, while the first rejoin attempt is still pending). + fn := scriptedHeartbeatFn([]int{-1, -1, 0, 0}) + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + runHeartbeat(ctx, fn, setNeedsRejoin, checkAndPerformRejoin) + close(done) + }() + + // Wait for both scripted rejoin attempts: one triggered by the + // transition tick (fails), one retried on the next successful tick + // (succeeds) -- proving the failed rejoin is retried, not abandoned + // (Req 2 AC5), at real heartbeat-tick granularity. + for i := 0; i < 2; i++ { + select { + case <-joinDone: + case <-time.After(6 * time.Second): + t.Fatalf("timed out waiting for rejoin attempt %d", i+1) + } + } + cancel() + <-done + + if got := atomic.LoadInt32(&joinCalls); got != 2 { + t.Errorf("performRejoinFn called %d times, want exactly 2 (first attempt "+ + "fails, retried on the next successful heartbeat tick)", got) + } + if got := atomic.LoadInt32(&needsRejoin); got != 0 { + t.Error("expected needsRejoin to clear once the retried rejoin succeeds") + } +} + +func TestRunHeartbeat_AllHeartbeatsSucceed_NeverSetsNeedsRejoin(t *testing.T) { + resetNeedsRejoinForTest() + atomic.StoreInt32(&heartbeatIntervalSecs, 1) + + origPerformRejoinFn := performRejoinFn + defer func() { performRejoinFn = origPerformRejoinFn }() + + var joinCalls int32 + performRejoinFn = func() int { + atomic.AddInt32(&joinCalls, 1) + return 0 + } + + tickEvents := make(chan struct{}, 8) + fn := func(_ int) int { + tickEvents <- struct{}{} + return 0 // always succeeds -- no failure ever observed + } + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + runHeartbeat(ctx, fn, setNeedsRejoin, checkAndPerformRejoin) + close(done) + }() + + for i := 0; i < 3; i++ { + select { + case <-tickEvents: + case <-time.After(5 * time.Second): + t.Fatalf("timed out waiting for tick %d", i+1) + } + } + cancel() + <-done + + if got := atomic.LoadInt32(&needsRejoin); got != 0 { + t.Error("needsRejoin must stay clear when every heartbeat tick succeeds -- " + + "no failure ever occurred to transition from") + } + if got := atomic.LoadInt32(&joinCalls); got != 0 { + t.Errorf("performRejoinFn called %d times, want 0 (checkAndPerformRejoin "+ + "must no-op when no rejoin is pending)", got) + } +} diff --git a/src/frontend/apiserver/backend/heartbeat_test.go b/src/frontend/apiserver/backend/heartbeat_test.go index 0924d3fd..e0bdcda6 100644 --- a/src/frontend/apiserver/backend/heartbeat_test.go +++ b/src/frontend/apiserver/backend/heartbeat_test.go @@ -31,16 +31,28 @@ func TestSetHeartbeatInterval_NegativeIgnored(t *testing.T) { } } +// noopRejoinHooks are used by tests that only care about runHeartbeat's +// tick/cancellation mechanics, not its failure-to-success/pending-rejoin +// callback behavior (see TestRunHeartbeat_FailureToSuccessTransition_* etc. +// below for that coverage). +func noopRejoinHooks() (func(), func()) { + return func() {}, func() {} +} + func TestRunHeartbeat_CallsFunction(t *testing.T) { atomic.StoreInt32(&heartbeatIntervalSecs, 1) var callCount int64 - fn := func(_ int) { atomic.AddInt64(&callCount, 1) } + fn := func(_ int) int { + atomic.AddInt64(&callCount, 1) + return 0 + } + onFailureToSuccess, onSuccessTick := noopRejoinHooks() ctx, cancel := context.WithCancel(context.Background()) done := make(chan struct{}) go func() { - runHeartbeat(ctx, fn) + runHeartbeat(ctx, fn, onFailureToSuccess, onSuccessTick) close(done) }() @@ -58,13 +70,15 @@ func TestRunHeartbeat_CancelsOnContext(t *testing.T) { cancel() // already cancelled var callCount int64 - fn := func(_ int) { + fn := func(_ int) int { atomic.AddInt64(&callCount, 1) + return 0 } + onFailureToSuccess, onSuccessTick := noopRejoinHooks() done := make(chan struct{}) go func() { - runHeartbeat(ctx, fn) + runHeartbeat(ctx, fn, onFailureToSuccess, onSuccessTick) close(done) }() @@ -79,3 +93,155 @@ func TestRunHeartbeat_CancelsOnContext(t *testing.T) { t.Errorf("fn was called %d times, want 0 for already-cancelled context", got) } } + +// scriptedHeartbeatFn returns a runHeartbeat fn double that yields +// results[0], results[1], ... on successive calls, clamping to the last +// entry for any calls beyond len(results) (defensive against a stray extra +// tick firing before a test's cancel() takes effect -- see the tests below, +// which all stop reading/cancel as soon as the expected callback events are +// observed rather than relying on a fixed sleep window). +func scriptedHeartbeatFn(results []int) func(int) int { + var idx int32 + return func(_ int) int { + i := atomic.AddInt32(&idx, 1) - 1 + if int(i) >= len(results) { + i = int32(len(results) - 1) + } + return results[i] + } +} + +// ════════════════════════════════════════════════════════════════════════ +// component-failover-adoption task 4.4 (Req 2 AC4/AC5) -- idle-period +// rejoin trigger. These tests exercise runHeartbeat's own tick/transition- +// tracking logic via generic injected callbacks, entirely without CGO (the +// production wiring -- reconnect.go's real setNeedsRejoin/ +// checkAndPerformRejoin -- is exercised end-to-end in +// heartbeat_rejoin_test.go, //go:build cgo). +// ════════════════════════════════════════════════════════════════════════ + +func TestRunHeartbeat_FailureToSuccessTransition_FiresOnFailureToSuccessThenOnSuccessTick(t *testing.T) { + atomic.StoreInt32(&heartbeatIntervalSecs, 1) + + // fail, fail, succeed -- the transition happens on tick 3. + fn := scriptedHeartbeatFn([]int{-1, -1, 0}) + + events := make(chan string, 8) + onFailureToSuccess := func() { events <- "failure_to_success" } + onSuccessTick := func() { events <- "success_tick" } + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + runHeartbeat(ctx, fn, onFailureToSuccess, onSuccessTick) + close(done) + }() + + want := []string{"failure_to_success", "success_tick"} + got := make([]string, 0, len(want)) + for range want { + select { + case e := <-events: + got = append(got, e) + case <-time.After(5 * time.Second): + t.Fatalf("timed out waiting for heartbeat callback events, got %v so far", got) + } + } + // Cancel immediately once the expected events are observed, before any + // further tick can fire -- keeps the assertion below exact regardless + // of scheduling slop. + cancel() + <-done + + for i := range want { + if got[i] != want[i] { + t.Fatalf("callback event %d = %q, want %q (full sequence got=%v want=%v) -- "+ + "onFailureToSuccess must fire exactly once, before onSuccessTick, "+ + "only on the fail-then-succeed transition tick", + i, got[i], want[i], got, want) + } + } +} + +func TestRunHeartbeat_EverySuccessfulTick_RetriesWhilePending(t *testing.T) { + atomic.StoreInt32(&heartbeatIntervalSecs, 1) + + // fail, then succeed 3 times in a row: onFailureToSuccess must fire + // only on the first success (the transition), but onSuccessTick must + // fire on ALL THREE successes -- proving the pending-rejoin check is + // retried on every successful tick, not abandoned after the transition + // tick (Req 2 AC5: "checked every tick, not only the transition edge"). + fn := scriptedHeartbeatFn([]int{-1, 0, 0, 0}) + + events := make(chan string, 8) + onFailureToSuccess := func() { events <- "f2s" } + onSuccessTick := func() { events <- "tick" } + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + runHeartbeat(ctx, fn, onFailureToSuccess, onSuccessTick) + close(done) + }() + + want := []string{"f2s", "tick", "tick", "tick"} + got := make([]string, 0, len(want)) + for range want { + select { + case e := <-events: + got = append(got, e) + case <-time.After(6 * time.Second): + t.Fatalf("timed out waiting for heartbeat callback events, got %v so far", got) + } + } + cancel() + <-done + + for i := range want { + if got[i] != want[i] { + t.Fatalf("callback event %d = %q, want %q (full sequence got=%v want=%v)", + i, got[i], want[i], got, want) + } + } +} + +func TestRunHeartbeat_AllTicksSucceed_NeverFiresOnFailureToSuccess(t *testing.T) { + atomic.StoreInt32(&heartbeatIntervalSecs, 1) + + fn := scriptedHeartbeatFn([]int{0, 0, 0}) + + var f2sCalls int32 + onFailureToSuccess := func() { atomic.AddInt32(&f2sCalls, 1) } + + var tickCalls int32 + tickEvents := make(chan struct{}, 8) + onSuccessTick := func() { + atomic.AddInt32(&tickCalls, 1) + tickEvents <- struct{}{} + } + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + runHeartbeat(ctx, fn, onFailureToSuccess, onSuccessTick) + close(done) + }() + + for i := 0; i < 3; i++ { + select { + case <-tickEvents: + case <-time.After(5 * time.Second): + t.Fatalf("timed out waiting for tick %d", i+1) + } + } + cancel() + <-done + + if got := atomic.LoadInt32(&f2sCalls); got != 0 { + t.Errorf("onFailureToSuccess called %d times, want 0 for an all-successful "+ + "run -- no failure was ever observed, so there is no transition to fire on", got) + } + if got := atomic.LoadInt32(&tickCalls); got < 3 { + t.Errorf("onSuccessTick called %d times, want at least 3", got) + } +}