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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NNB4zGPCsHyDUCDJyMbiQN

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.

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)

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)
+	}
+}

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)
+	}
+}