feat(component-failover-adoption): add apiserver's backend-health circuit breaker

Adds an atomic degraded-until state (health.go: IsDegraded/MarkDegraded/
MarkHealthy) that lets request handlers fail fast with a clear
unavailable error instead of blocking on xcecli's own ~100s retry
budget when the backend is known-degraded.

Consolidates what tasks 4.3/4.4 left as duplicated inline
"call xcecli_*; checkAndPerformRejoin()" pairs across five wrapper
functions into one shared callBackend() helper: checks IsDegraded()
before every call (fail fast, skip the C call and the rejoin check
entirely on the degraded path -- checking rejoin there would itself
block up to ~100s, defeating the fail-fast purpose), marks degraded on
-E_XCE_NOT_READY, marks healthy at checkAndPerformRejoin()'s existing
success point (the single call site both the write path and the
idle-period heartbeat path already funnel through).

The circuit breaker's own TTL-based state is deliberately never read
by the rejoin-retry mechanism -- checkAndPerformRejoin() has no
IsDegraded() dependency, so a fast-fail window never suppresses
retry-until-actually-joined; heartbeat.go's unconditional per-tick
call keeps that cadence alive regardless of degraded state, and a
10s TTL (well under the ~100s retry budget) lets write traffic resume
contributing once it clears.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NNB4zGPCsHyDUCDJyMbiQN
This commit is contained in:
Rongsong Shen 2026-08-12 10:52:10 +00:00
parent ecf5eddc61
commit c8325e7575
Signed by: rshen
SSH key fingerprint: SHA256:8Vf2pZ3h6rJa7EhGK+WaAQ9Mta/apZvteW6g9U/WKKk
7 changed files with 648 additions and 39 deletions

View file

@ -13,6 +13,7 @@ package backend
#include "controller.capnp.h"
#include "val_helpers.h"
#include "xce/cluster.h"
#include "xce/errnox.h"
// backend_ctx_t wraps xcecli_ctx_t for use as the C handler's arg.
typedef struct {
@ -78,9 +79,11 @@ import (
"apiserver/models"
logger "apiserver/utils"
"context"
"errors"
"fmt"
"strings"
"sync"
"time"
"unsafe"
"go.uber.org/zap"
@ -534,6 +537,85 @@ func handleNotifyData(nd *C.xce_notify_data_t) C.int {
return 0
}
// ErrBackendUnavailable is returned immediately by callBackend (and
// therefore by every wrapper below) when ApiserverBackendHealth reports the
// backend connection degraded (Req 4.1) -- a clear, distinct error a caller
// can recognize as "backend down, retrying" rather than an unrelated
// internal failure, without ever invoking the underlying blocking
// C.xcecli_* call.
var ErrBackendUnavailable = errors.New(
"apiserver: backend connection unavailable (circuit breaker open); try again shortly")
// backendDegradedTTL is how long callBackend marks the backend degraded
// (ApiserverBackendHealth.MarkDegraded) after a call's own outcome
// indicates its retry budget was exhausted, before request handlers resume
// probing it directly again -- the half-open half of the circuit breaker
// (design.md's ApiserverBackendHealth). Chosen well below xcecli's own
// full per-call retry budget (XCECLI_FAILOVER_CALL: max_retries=20,
// backoff capped at 5000ms -- roughly 1-2 minutes total,
// client_failover.c:218,220): 10s is short enough that a backend which
// recovers quickly gets re-probed promptly, while still long enough that
// every request arriving during a genuine outage fails fast instead of
// each independently re-discovering the same ~100s exhausted budget one
// blocking call at a time.
const backendDegradedTTL = 10 * time.Second
// notReadyRC exposes -E_XCE_NOT_READY as a plain Go int, for
// health_rejoin_test.go (a //go:build cgo test file that cannot `import "C"`
// itself, golang/go#13938) to script a callBackend closure returning this
// exact retry-budget-exhaustion value without hardcoding the raw errno
// number -- same plain-Go-signature test-seam rationale as performRejoinFn
// (reconnect.go) and currentClientConfigNMasters() (above).
var notReadyRC = -int(C.E_XCE_NOT_READY)
// callBackend is the single shared pre-call/post-call helper every
// gCtx.ctx-based backend-call wrapper below (SubmitJob, ControlJob,
// ControlQueue, ControlHost, FetchFile) calls around its own single
// C.xcecli_* invocation (component-failover-adoption task 4.5, boundary:
// ApiserverBackendBridge/ApiserverBackendHealth) -- implemented once here,
// not copy-pasted per wrapper. fn performs exactly one C.xcecli_* call and
// returns its raw rc as a plain Go int (not C.int) -- the same
// plain-Go-signature test seam pattern as reconnect.go's performRejoinFn,
// letting a //go:build cgo test file (which can never `import "C"` itself,
// golang/go#13938) drive this helper directly with a scripted closure, with
// no live backend connection or even an initialized gCtx required.
//
// Pre-call (Req 4.1): if IsDegraded(), fn is never invoked at all -- the
// wrapped call (and its own up-to-~100s xcecli retry budget) is skipped
// entirely, returning ErrBackendUnavailable immediately. checkAndPerformRejoin
// is deliberately NOT called on this path either: it can itself block on
// xcecli_join's own up-to-~100s retry budget, which would defeat the whole
// point of failing fast here (Req 4.1's bounded, sub-second response). The
// rejoin-retry cadence during this window keeps running regardless, driven
// by heartbeat.go's own unconditional per-tick call to checkAndPerformRejoin
// (task 4.4) -- which never reads IsDegraded() -- so the pending-rejoin
// flag's retry-until-actually-joined behavior is never suppressed by this
// short-circuit (see health.go's own doc comment for why these two states
// must stay independent).
//
// Post-call (only reached when not degraded): the existing rejoin check
// (checkAndPerformRejoin, task 4.3/4.4, unchanged) always runs, regardless
// of fn's own outcome -- a pending rejoin is orthogonal to whether this
// particular call happened to succeed. If fn's rc is exactly
// -E_XCE_NOT_READY (this call's own retry budget exhausted), MarkDegraded
// opens the circuit for subsequent calls.
func callBackend(name string, fn func() int) (rc int, err error) {
if IsDegraded() {
return 0, ErrBackendUnavailable
}
rc = fn()
checkAndPerformRejoin()
if rc == notReadyRC {
MarkDegraded(backendDegradedTTL)
}
if rc < 0 {
return rc, fmt.Errorf("%s failed: %d", name, rc)
}
return rc, 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
@ -545,10 +627,11 @@ 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)
checkAndPerformRejoin()
if rc < 0 {
return fmt.Errorf("xcecli_jobsubmit failed: %d", int(rc))
_, err := callBackend("xcecli_jobsubmit", func() int {
return int(C.xcecli_jobsubmit(gCtx.ctx, js, &outJobID))
})
if err != nil {
return err
}
if outJobID != 0 {
job.Unique_JobID = int64(outJobID)
@ -581,12 +664,10 @@ func ControlJob(jobid uint64, action int, content string) error {
C.jobctrl_set_ckptoption(&ctrl, &ckpt)
}
rc := C.xcecli_jobctrl(gCtx.ctx, &ctrl)
checkAndPerformRejoin()
if rc < 0 {
return fmt.Errorf("xcecli_jobctrl failed: %d", int(rc))
}
return nil
_, err := callBackend("xcecli_jobctrl", func() int {
return int(C.xcecli_jobctrl(gCtx.ctx, &ctrl))
})
return err
}
// ControlQueue sends a queue open/close/activate/deactivate action to the
@ -600,12 +681,10 @@ func ControlQueue(queuename string, action int) error {
queue: cQueue,
}
rc := C.xcecli_queuectrl(gCtx.ctx, &ctrl)
checkAndPerformRejoin()
if rc < 0 {
return fmt.Errorf("xcecli_queuectrl failed: %d", int(rc))
}
return nil
_, err := callBackend("xcecli_queuectrl", func() int {
return int(C.xcecli_queuectrl(gCtx.ctx, &ctrl))
})
return err
}
// ControlHost sends a host open/close action to the backend via xcecli_hostctrl.
@ -618,12 +697,10 @@ func ControlHost(hostname string, action int) error {
hostname: cHost,
}
rc := C.xcecli_hostctrl(gCtx.ctx, &ctrl)
checkAndPerformRejoin()
if rc < 0 {
return fmt.Errorf("xcecli_hostctrl failed: %d", int(rc))
}
return nil
_, err := callBackend("xcecli_hostctrl", func() int {
return int(C.xcecli_hostctrl(gCtx.ctx, &ctrl))
})
return err
}
// signalNumber maps a job action and optional signal name to a POSIX signal
@ -660,10 +737,11 @@ func FetchFile(jobid uint64, filepath string, offset uint64, maxBytes uint32) (*
}
var reply *C.xce_fetchfile_reply_t
rc := C.xcecli_fetchfile(gCtx.ctx, &msg, &reply)
checkAndPerformRejoin()
if rc < 0 {
return nil, fmt.Errorf("xcecli_fetchfile failed: %d", int(rc))
_, err := callBackend("xcecli_fetchfile", func() int {
return int(C.xcecli_fetchfile(gCtx.ctx, &msg, &reply))
})
if err != nil {
return nil, err
}
if reply == nil {
return nil, fmt.Errorf("xcecli_fetchfile returned nil reply")