diff --git a/src/frontend/apiserver/backend/clib.go b/src/frontend/apiserver/backend/clib.go index b64aa6b6..1156e8ff 100644 --- a/src/frontend/apiserver/backend/clib.go +++ b/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 } diff --git a/src/frontend/apiserver/backend/health.go b/src/frontend/apiserver/backend/health.go new file mode 100644 index 00000000..a0be00bc --- /dev/null +++ b/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) } diff --git a/src/frontend/apiserver/backend/health_test.go b/src/frontend/apiserver/backend/health_test.go new file mode 100644 index 00000000..ddc6e7c4 --- /dev/null +++ b/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") + } +}