diff --git a/src/frontend/apiserver/backend/clib.go b/src/frontend/apiserver/backend/clib.go index f99aa5e9..b64aa6b6 100644 --- a/src/frontend/apiserver/backend/clib.go +++ b/src/frontend/apiserver/backend/clib.go @@ -36,6 +36,15 @@ static int xce_backend_init(xce_client_config_t *config, backend_ctx_t *ctx) { return xcecli_join(ctx->ctx, HostRole_apiserver, NULL); } +// xce_backend_rejoin re-registers the apiserver with the (re)connected +// controller. The controller resets its peer table on restart, so after a +// transport reconnect the apiserver must re-join to resume push notifications +// (component-failover-adoption Req 2.4). Call this from Go AFTER a heartbeat +// has re-established the transport — never from the lib's own reconnect hook. +static int xce_backend_rejoin(backend_ctx_t *ctx) { + return xcecli_join(ctx->ctx, HostRole_apiserver, NULL); +} + // Helpers to extract typed row pointers from a dataset_t. static xce_job_t **dataset_jobs(dataset_t *d) { return (xce_job_t **)d->rows; } static xce_worker_t **dataset_workers(dataset_t *d) { return (xce_worker_t **)d->rows; } @@ -60,6 +69,7 @@ import ( "context" "fmt" "sync" + "sync/atomic" "unsafe" "go.uber.org/zap" @@ -185,11 +195,40 @@ func Cleanup() { C.idgen_ctx_fini(&gIDCtx) } +// rejoinBackend re-registers the apiserver with the controller after a +// transport reconnect (component-failover-adoption Req 2.4). xcecli_heartbeat +// is fire-and-forget: it best-effort reconnects the transport but never +// re-joins, so without an explicit rejoin the controller's restarted peer +// table never learns this apiserver again and push notifications stop. +func rejoinBackend() { + if gCtx.ctx == nil { + return + } + rc := C.xce_backend_rejoin(&gCtx) + if rc < 0 { + logger.Warn("backend rejoin failed (will retry on next heartbeat)", + zap.Int("rc", int(rc))) + return + } + logger.Info("backend re-joined controller after reconnect") +} + // StartHeartbeat launches a goroutine that sends periodic xcecli_heartbeat -// keep-alive messages to the controller. Stops when ctx is cancelled. +// keep-alive messages to the controller. Stops when ctx is cancelled. When a +// heartbeat transitions from failure to success (the transport reconnect +// succeeded), it triggers an explicit backend rejoin so the apiserver resumes +// receiving push notifications without a process restart. func StartHeartbeat(ctx context.Context) { + var prevFailed atomic.Bool go runHeartbeat(ctx, func(secs int) { - C.xcecli_heartbeat(gCtx.ctx, C.int(secs)) + rc := C.xcecli_heartbeat(gCtx.ctx, C.int(secs)) + if rc < 0 { + prevFailed.Store(true) + return + } + if prevFailed.Swap(false) { + rejoinBackend() + } }) } diff --git a/src/worker/src/controller_connector.c b/src/worker/src/controller_connector.c index 6a466dbc..02440dc3 100755 --- a/src/worker/src/controller_connector.c +++ b/src/worker/src/controller_connector.c @@ -160,16 +160,6 @@ static void init_client_config(xce_client_config_t* config) config->key); } -static void handle_disconnect() -{ - XCECD_LOG(XCE_LOG_DEBUG, "Handling disconnect and shutdown XCECD..."); - - // Shutdown worker and waiting for restart - cc_ctx->running = false; - - XCECD_LOG(XCE_LOG_DEBUG, "Disconnect handled."); -} - /** * @brief Handles I/O events for the endpoint, focusing on error conditions * and connection liveness checks using EPOLLIN/EPOLLOUT. @@ -178,12 +168,19 @@ static void handle_disconnect() */ int checking_connection() { - const int MAX_RETRY_COUNT = 5; - const int RETRY_INTERVAL = 5; // seconds - int retry_count = 0; + const int RETRY_INTERVAL = 5; // seconds + int retry_count = 0; int ret; - while (retry_count < MAX_RETRY_COUNT) { + /* component-failover-adoption Req 2/Req 3: keep retrying indefinitely so + * the worker self-heals when its controller (or a failover primary) + * becomes reachable again — never give up after a bounded retry count and + * never call handle_disconnect() on a transient outage (that sets + * cc_ctx->running=false and exits the connector thread, requiring an + * operator restart). The connector thread is dedicated to this loop, so + * the unbounded retry/sleep cannot block the worker's other liveness + * duties (hardware/process monitoring run on separate threads). */ + while (1) { ret = xcecli_hello(xcecli_ctx, "hello"); if (ret == 0) { // Connection successful @@ -197,8 +194,8 @@ int checking_connection() retry_count++; XCECD_LOG(XCE_LOG_ERROR, - "Connection to controller is broken. Retry %d/%d failed", - retry_count, MAX_RETRY_COUNT); + "Connection to controller is broken. Retry %d failed", + retry_count); if (0 == xcecd_join_cluster(&host_attrs)) { XCECD_LOG(XCE_LOG_INFO, @@ -211,26 +208,13 @@ int checking_connection() * terminal state for processes that are gone). */ (void)resend_job_status(*cc_ctx->worker_list); - // api_ctx_t* api_ctx = (api_ctx_t*)xcecli_ctx; - // cc_ctx->xce_ctx = &(api_ctx->ctx); return 0; } - if (retry_count < MAX_RETRY_COUNT) { - XCECD_LOG(XCE_LOG_INFO, "Retrying connection in %d seconds...", - RETRY_INTERVAL); - sleep(RETRY_INTERVAL); - } + XCECD_LOG(XCE_LOG_INFO, "Retrying connection in %d seconds...", + RETRY_INTERVAL); + sleep(RETRY_INTERVAL); } - - // All retries exhausted - XCECD_LOG(XCE_LOG_ERROR, - "Connection to controller failed after %d retries. Handling " - "disconnect...", - MAX_RETRY_COUNT); - handle_disconnect(); - - return 0; } /**