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>
This commit is contained in:
lei.chen 2026-08-14 05:20:52 +00:00
parent d4b98aab01
commit 2bbe894f99
3 changed files with 195 additions and 12 deletions

View file

@ -224,10 +224,12 @@ func StartHeartbeat(ctx context.Context) {
rc := C.xcecli_heartbeat(gCtx.ctx, C.int(secs)) rc := C.xcecli_heartbeat(gCtx.ctx, C.int(secs))
if rc < 0 { if rc < 0 {
prevFailed.Store(true) prevFailed.Store(true)
MarkDegraded()
return return
} }
if prevFailed.Swap(false) { if prevFailed.Swap(false) {
rejoinBackend() rejoinBackend()
MarkHealthy()
} }
}) })
} }
@ -404,6 +406,49 @@ func handleNotifyData(nd *C.xce_notify_data_t) C.int {
return 0 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. // SubmitJob sends a job submission to the backend via xcecli_jobsubmit.
// arraySize=1 for normal jobs, >1 for array jobs. // arraySize=1 for normal jobs, >1 for array jobs.
// The store is updated only when the backend pushes the resulting OnJobCreated // 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) js.array_size = C.uint64_t(arraySize)
var outJobID C.uint64_t var outJobID C.uint64_t
rc := C.xcecli_jobsubmit(gCtx.ctx, js, &outJobID) if err := callBackend("xcecli_jobsubmit", func() int {
if rc < 0 { return int(C.xcecli_jobsubmit(gCtx.ctx, js, &outJobID))
return fmt.Errorf("xcecli_jobsubmit failed: %d", int(rc)) }); err != nil {
return err
} }
if outJobID != 0 { if outJobID != 0 {
job.Unique_JobID = int64(outJobID) job.Unique_JobID = int64(outJobID)
@ -455,9 +501,10 @@ func ControlJob(jobid uint64, action int, content string) error {
C.jobctrl_set_ckptoption(&ctrl, &ckpt) C.jobctrl_set_ckptoption(&ctrl, &ckpt)
} }
rc := C.xcecli_jobctrl(gCtx.ctx, &ctrl) if err := callBackend("xcecli_jobctrl", func() int {
if rc < 0 { return int(C.xcecli_jobctrl(gCtx.ctx, &ctrl))
return fmt.Errorf("xcecli_jobctrl failed: %d", int(rc)) }); err != nil {
return err
} }
return nil return nil
} }
@ -473,9 +520,10 @@ func ControlQueue(queuename string, action int) error {
queue: cQueue, queue: cQueue,
} }
rc := C.xcecli_queuectrl(gCtx.ctx, &ctrl) if err := callBackend("xcecli_queuectrl", func() int {
if rc < 0 { return int(C.xcecli_queuectrl(gCtx.ctx, &ctrl))
return fmt.Errorf("xcecli_queuectrl failed: %d", int(rc)) }); err != nil {
return err
} }
return nil return nil
} }
@ -490,9 +538,10 @@ func ControlHost(hostname string, action int) error {
hostname: cHost, hostname: cHost,
} }
rc := C.xcecli_hostctrl(gCtx.ctx, &ctrl) if err := callBackend("xcecli_hostctrl", func() int {
if rc < 0 { return int(C.xcecli_hostctrl(gCtx.ctx, &ctrl))
return fmt.Errorf("xcecli_hostctrl failed: %d", int(rc)) }); err != nil {
return err
} }
return nil return nil
} }

View file

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

View file

@ -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")
}
}