feat(apiserver): fail-fast circuit breaker when backend controller is down (Req 4)

component-failover-adoption Req 4: when the apiserver's controller connection
is down, backend-requiring API requests must return a clear service-unavailable
error rather than hanging. Adds:

- health.go: an atomic circuit breaker (IsDegraded/MarkDegraded/MarkHealthy).
- callBackend: the four write calls (jobsubmit/jobctrl/queuectrl/hostctrl)
  now route through it — degraded ⇒ immediate "backend controller unavailable"
  error without invoking the lib; a transport OR XCE not-ready/masterless/
  standby failure (E_XCE_NOMASTER/REDIRECT/NOT_READY/STANDBY) trips the breaker
  so the NEXT call fails fast. Semantic rejections (-EINVAL/-ENOENT/E_XCE_BAD_*)
  never trip it.
- StartHeartbeat: a failed tick marks degraded; recovery re-joins (rejoinBackend)
  and marks healthy.

Verified live (t1 cluster): kill controller → first submit returns
xcecli_jobsubmit failed: -10013 (bounded, no hang), second submit returns
"backend controller unavailable (reconnecting)" in ~11ms. 7 unit tests.
Recovery-via-heartbeat after a FULL controller death is NOT yet reliable — the
lib transport reconnect does not always re-establish to the restarted
controller (same class as the scheduler reconnect; tracked under task 13).

Co-Authored-By: Claude <noreply@anthropic.com>

src/frontend/apiserver/backend/clib.go

@@ -224,10 +224,12 @@ func StartHeartbeat(ctx context.Context) {
 		rc := C.xcecli_heartbeat(gCtx.ctx, C.int(secs))
 		if rc < 0 {
 			prevFailed.Store(true)
+			MarkDegraded()
 			return
 		}
 		if prevFailed.Swap(false) {
 			rejoinBackend()
+			MarkHealthy()
 		}
 	})
 }
@@ -404,6 +406,49 @@ func handleNotifyData(nd *C.xce_notify_data_t) C.int {
 	return 0
 }
 
+// isBackendUnavailable reports whether a negative xcecli_* return means the
+// backend controller cannot serve the op (transport failure OR an XCE
+// not-ready/masterless/standby code) rather than a semantic rejection. Only
+// these trip the Req 4 circuit breaker — a semantic rejection (-EINVAL/
+// -ENOENT/...) must never mark the backend "down".
+func isBackendUnavailable(rc int) bool {
+	switch rc {
+	case -int(C.ECONNRESET), -int(C.ECONNREFUSED), -int(C.EHOSTUNREACH),
+		-int(C.ENETUNREACH), -int(C.ENETDOWN), -int(C.ETIMEDOUT),
+		-int(C.EPIPE), -int(C.EIO), -int(C.ENOTCONN), -int(C.ESHUTDOWN),
+		-int(C.ENODEV), -int(C.ENOLINK), -int(C.ECONNABORTED):
+		return true
+	// XCE controller role codes (errnox.h): the controller is reachable but
+	// cannot authoritatively serve the op — no master, redirecting, initializing/
+	// electing, or a standby. All mean "backend unavailable right now".
+	case -(int(C.E_XCE_NOMASTER)), -(int(C.E_XCE_REDIRECT)),
+		-(int(C.E_XCE_NOT_READY)), -(int(C.E_XCE_STANDBY)):
+		return true
+	}
+	return false
+}
+
+// callBackend guards a C.xcecli_* write call with the fail-fast circuit
+// breaker (component-failover-adoption Req 4): when degraded, return a
+// service-unavailable error immediately rather than invoking a call that would
+// hang on a dead controller. A transport failure trips the breaker so the NEXT
+// call fails fast (the lib's retry budget already bounded this call's block).
+// fn returns the (negative) xcecli return code converted to a plain int so the
+// breaker logic is unit-testable without cgo.
+func callBackend(name string, fn func() int) error {
+	if IsDegraded() {
+		return fmt.Errorf("%s: backend controller unavailable (reconnecting)", name)
+	}
+	rc := fn()
+	if rc < 0 {
+		if isBackendUnavailable(rc) {
+			MarkDegraded()
+		}
+		return fmt.Errorf("%s failed: %d", name, rc)
+	}
+	return nil
+}
+
 // SubmitJob sends a job submission to the backend via xcecli_jobsubmit.
 // arraySize=1 for normal jobs, >1 for array jobs.
 // The store is updated only when the backend pushes the resulting OnJobCreated
@@ -415,9 +460,10 @@ func SubmitJob(job *models.JobInfo, arraySize int) error {
 	js.array_size = C.uint64_t(arraySize)
 
 	var outJobID C.uint64_t
-	rc := C.xcecli_jobsubmit(gCtx.ctx, js, &outJobID)
-	if rc < 0 {
-		return fmt.Errorf("xcecli_jobsubmit failed: %d", int(rc))
+	if err := callBackend("xcecli_jobsubmit", func() int {
+		return int(C.xcecli_jobsubmit(gCtx.ctx, js, &outJobID))
+	}); err != nil {
+		return err
 	}
 	if outJobID != 0 {
 		job.Unique_JobID = int64(outJobID)
@@ -455,9 +501,10 @@ func ControlJob(jobid uint64, action int, content string) error {
 		C.jobctrl_set_ckptoption(&ctrl, &ckpt)
 	}
 
-	rc := C.xcecli_jobctrl(gCtx.ctx, &ctrl)
-	if rc < 0 {
-		return fmt.Errorf("xcecli_jobctrl failed: %d", int(rc))
+	if err := callBackend("xcecli_jobctrl", func() int {
+		return int(C.xcecli_jobctrl(gCtx.ctx, &ctrl))
+	}); err != nil {
+		return err
 	}
 	return nil
 }
@@ -473,9 +520,10 @@ func ControlQueue(queuename string, action int) error {
 		queue:  cQueue,
 	}
 
-	rc := C.xcecli_queuectrl(gCtx.ctx, &ctrl)
-	if rc < 0 {
-		return fmt.Errorf("xcecli_queuectrl failed: %d", int(rc))
+	if err := callBackend("xcecli_queuectrl", func() int {
+		return int(C.xcecli_queuectrl(gCtx.ctx, &ctrl))
+	}); err != nil {
+		return err
 	}
 	return nil
 }
@@ -490,9 +538,10 @@ func ControlHost(hostname string, action int) error {
 		hostname: cHost,
 	}
 
-	rc := C.xcecli_hostctrl(gCtx.ctx, &ctrl)
-	if rc < 0 {
-		return fmt.Errorf("xcecli_hostctrl failed: %d", int(rc))
+	if err := callBackend("xcecli_hostctrl", func() int {
+		return int(C.xcecli_hostctrl(gCtx.ctx, &ctrl))
+	}); err != nil {
+		return err
 	}
 	return nil
 }

src/frontend/apiserver/backend/health.go

@@ -0,0 +1,24 @@
+package backend
+
+import "sync/atomic"
+
+// backendDegraded is the fail-fast circuit breaker (component-failover-adoption
+// Req 4): when the controller connection is down, backend-requiring API
+// requests return a clear "backend unavailable" error immediately instead of
+// hanging until the connection is restored.
+//
+// Lifecycle:
+//   - tripped by a write-path transport failure (callBackend) and by a failed
+//     heartbeat tick (StartHeartbeat);
+//   - cleared by a successful heartbeat recovery (StartHeartbeat → rejoinBackend
+//     → MarkHealthy), which also re-joins the apiserver session so pushes resume.
+var backendDegraded atomic.Bool
+
+// IsDegraded reports whether the backend controller is currently unreachable.
+func IsDegraded() bool { return backendDegraded.Load() }
+
+// MarkDegraded trips the circuit breaker — backend calls now fail fast.
+func MarkDegraded() { backendDegraded.Store(true) }
+
+// MarkHealthy clears the circuit breaker after a successful reconnect/rejoin.
+func MarkHealthy() { backendDegraded.Store(false) }

src/frontend/apiserver/backend/health_test.go

@@ -0,0 +1,110 @@
+package backend
+
+import "testing"
+
+// Linux errno + XCE error values. Plain ints keep this test cgo-free;
+// callBackend's fn returns a plain int by design for exactly this reason.
+const (
+	ecnreset       = -104    // ECONNRESET
+	einval         = -22     // EINVAL
+	xceNotReady    = -10013  // E_XCE_NOT_READY (controller init/electing)
+	xceBadQueue    = -10008  // E_XCE_BAD_QUEUE (semantic — must NOT trip)
+)
+
+func resetBreaker() { MarkHealthy() }
+
+// ── state transitions ────────────────────────────────────────────────────────
+
+func TestCircuitBreakerStateTransitions(t *testing.T) {
+	resetBreaker()
+	if IsDegraded() {
+		t.Fatal("fresh breaker should be healthy")
+	}
+	MarkDegraded()
+	if !IsDegraded() {
+		t.Fatal("MarkDegraded should trip the breaker")
+	}
+	MarkHealthy()
+	if IsDegraded() {
+		t.Fatal("MarkHealthy should clear the breaker")
+	}
+}
+
+// ── callBackend ──────────────────────────────────────────────────────────────
+
+func TestCallBackend_Degraded_FailsFastWithoutCalling(t *testing.T) {
+	resetBreaker()
+	MarkDegraded()
+
+	called := false
+	err := callBackend("test", func() int {
+		called = true
+		return 0
+	})
+	if err == nil {
+		t.Fatal("degraded call should return an error")
+	}
+	if called {
+		t.Fatal("degraded call must NOT invoke the backend fn")
+	}
+}
+
+func TestCallBackend_TransportError_TripsBreaker(t *testing.T) {
+	resetBreaker()
+
+	err := callBackend("test", func() int { return ecnreset })
+	if err == nil {
+		t.Fatal("transport error should return an error")
+	}
+	if !IsDegraded() {
+		t.Fatal("transport error should trip the circuit breaker")
+	}
+}
+
+func TestCallBackend_XceNotReady_TripsBreaker(t *testing.T) {
+	resetBreaker()
+
+	err := callBackend("test", func() int { return xceNotReady })
+	if err == nil {
+		t.Fatal("XCE not-ready should return an error")
+	}
+	if !IsDegraded() {
+		t.Fatal("XCE not-ready (controller unavailable) should trip the circuit breaker")
+	}
+}
+
+func TestCallBackend_SemanticError_DoesNotTripBreaker(t *testing.T) {
+	resetBreaker()
+
+	err := callBackend("test", func() int { return einval })
+	if err == nil {
+		t.Fatal("semantic error should return an error")
+	}
+	if IsDegraded() {
+		t.Fatal("semantic error must NOT trip the circuit breaker")
+	}
+}
+
+func TestCallBackend_XceSemanticError_DoesNotTripBreaker(t *testing.T) {
+	resetBreaker()
+
+	err := callBackend("test", func() int { return xceBadQueue })
+	if err == nil {
+		t.Fatal("XCE semantic error should return an error")
+	}
+	if IsDegraded() {
+		t.Fatal("XCE semantic rejection must NOT trip the circuit breaker")
+	}
+}
+
+func TestCallBackend_Success_StaysHealthy(t *testing.T) {
+	resetBreaker()
+
+	err := callBackend("test", func() int { return 0 })
+	if err != nil {
+		t.Fatalf("success should return nil, got %v", err)
+	}
+	if IsDegraded() {
+		t.Fatal("success must not trip the circuit breaker")
+	}
+}