/* * ha_convergence.c — demotion convergence + self-restart primitive * (ha-convergence-recovery Requirement 1). * * ─────────────────────────────────────────────────────────────────────────── * This TU owns the self-restart primitive and the deferred demotion-convergence * task body. The synchronous demotion BRANCH (hook disable + task scheduling) * lives in raft_cb_become_follower (ha.c) because that callback is file-static; * it calls ha_schedule_demotion_convergence() declared here. * * Split of responsibilities (design "Demotion Convergence + Self-Restart"): * - raft_cb_become_follower (ha.c): read prior role → if was-PRIMARY, disable * the 3 persistence hooks INLINE (Req 1.1), * then schedule the one-shot convergence * task off the callback thread (D1). * - ha_demotion_convergence_fn (here, delaymgr task body): truncate above * committed_lsn (Req 1.2) → WARN range * (Req 1.5) → ha_self_restart (Req 1.3). * - ha_self_restart (here): flush logs → execv(saved argv); exec * return → ERROR + abort (FATAL, Req 1.4); * under XCE_TESTING route to the restart / * fatal seams (Req 1.6, 1.4). * * controller_saved_argv lives in main.c (the real argv capture); this TU only * consumes it through the ha.h accessor. * * fd hygiene (Req 1.3 / design fd audit): every controller-to-controller socket * — listener, accepted peer/replication connections, and outbound vote/peer * connect sockets — is created FD_CLOEXEC in src/lib/netio.c (socket_create * sets it at netio.c:219; socket_accept at netio.c:308). The re-exec'd image * therefore inherits no live sockets and re-creates all endpoints through the * boot path; no extra close-before-exec sweep is required here. * ─────────────────────────────────────────────────────────────────────────── */ #define _GNU_SOURCE #include #include #include #include #include #include #include #include #include #include "cluster.h" #include "delay.h" #include "ha.h" #include "persist.h" #include "xce/cluster_def.h" #ifdef XCE_TESTING /* XCE_TESTING seams (Req 1.6, 1.4) — defined here, declared in ha.h. * Tests set these before exercising the restart/fatal contract. */ int (*ha_restart_hook)(cluster_t *c) = NULL; void (*ha_fatal_hook)(const char *msg) = NULL; /* Quorum-transition emission-count seam (task 5.x, Req 8.1-8.3) — defined here, * declared in ha.h. ha_log_quorum_transition invokes this once per WARN it * emits so the counter-hook tests can assert once-per-transition + counts. * NULL until a test installs it. */ void (*ha_quorum_transition_log_hook)(int lost, int connected, int required) = NULL; #endif /* * ha_self_restart — the truncate-and-restart primitive's restart half. * * Production (Req 1.3, 1.4): flush logs, then execv() the saved argv so the * controller rebuilds in-memory state purely from the surviving (committed) * WAL prefix via boot recovery. If exec cannot be performed, terminate * FATALLY (abort) — a demoted ex-primary must NEVER continue running with * potentially divergent state. NEVER returns in production. * * Test (Req 1.6): under XCE_TESTING invoke ha_restart_hook(c) instead of * execv and return its result on the success path; a non-zero hook return * models an exec failure and routes to ha_fatal_hook in place of abort, so a * test binary can assert the FATAL path was taken without dying (Req 1.4). */ int ha_self_restart(cluster_t *c) { #ifdef XCE_TESTING /* Test contract: the restart seam stands in for execv. */ int rc = 0; if (ha_restart_hook != NULL) rc = ha_restart_hook(c); if (rc != 0 || ha_restart_hook == NULL) { /* Restart could not be performed → FATAL contract (Req 1.4). */ if (ha_fatal_hook != NULL) ha_fatal_hook("ha_self_restart: restart seam failed (FATAL)"); return rc != 0 ? rc : -1; } return 0; #else (void)c; char **argv = controller_saved_argv(); if (argv == NULL || argv[0] == NULL) { ERROR("HA self-restart: no saved argv — cannot re-exec (FATAL)"); tlog_exit(); /* flush pending log buffer before dying */ abort(); } INFO("HA self-restart: re-executing %s to rebuild from committed state", argv[0]); tlog_exit(); /* flush + drain the async log buffer before exec replaces us */ execv(argv[0], argv); /* execv only returns on failure — FATAL (Req 1.4): never continue serving * as a divergent ex-primary. */ ERROR("HA self-restart: execv(%s) failed: %d — aborting (FATAL)", argv[0], errno); abort(); return -1; /* unreachable */ #endif } /* * ha_demotion_convergence_ctx_t — heap context for the one-shot delaymgr task. * Holds only the cluster pointer; committed_lsn is read at task time (not * schedule time) so a late-advancing commit still truncates to the right * boundary. */ typedef struct { cluster_t *c; } ha_demotion_convergence_ctx_t; /* * ha_demotion_convergence_fn — delaymgr task body (runs OFF the become-follower * callback thread, D1). Truncate above committed_lsn (Req 1.2), WARN with the * truncated range (Req 1.5), then self-restart (Req 1.3). On truncation failure * (Req error strategy) ERROR then restart anyway — the rejoin reconciliation * loop converges the untruncated case at the next join. * * In XCE_TESTING the test fixture drives this entry directly (un-started * delaymgr); ha_self_restart routes to the restart seam there. The ctx is * freed before restart in production restart never returns, so only the test * path reaches the free — kept on every non-restarting exit for safety. */ int ha_demotion_convergence_fn(void *arg) { ha_demotion_convergence_ctx_t *ctx = (ha_demotion_convergence_ctx_t *)arg; if (ctx == NULL || ctx->c == NULL) { free(ctx); return -EINVAL; } cluster_t *c = ctx->c; uint64_t committed = atomic_load_explicit(&c->persist.committed_lsn, memory_order_acquire); /* Prior log tip for the WARN range (read under the persist lock). */ pthread_mutex_lock(&c->persist.lock); uint64_t prior_tip = c->persist.current_lsn > 0 ? c->persist.current_lsn - 1 : 0; pthread_mutex_unlock(&c->persist.lock); int trc = persist_truncate_wal_above(&c->persist, committed); if (trc < 0) { ERROR("HA demotion convergence: truncate above committed_lsn=%" PRIu64 " failed: %d — restarting anyway (rejoin reconciliation will " "converge)", committed, trc); } else { /* Req 1.5: log the discarded range — (committed_lsn, prior tip]. */ WARN("HA demotion convergence: truncated WAL above committed_lsn=%" PRIu64 " (prior tip=%" PRIu64 "); discarding provisional suffix, " "restarting to rebuild from committed state", committed, prior_tip); } /* Free the task ctx before restart (production exec never returns). */ free(ctx); /* Req 1.3: rebuild in-memory state through boot recovery. */ return ha_self_restart(c); } /* * ha_schedule_demotion_convergence — called from raft_cb_become_follower's * was-PRIMARY branch AFTER the persistence hooks have been disabled inline. * Schedules exactly one one-shot delaymgr task running ha_demotion_convergence_fn * so the file I/O (truncation) and exec never run on the callback thread (D1). * * Returns 0 on success, negative on allocation/scheduling failure. On failure * the node stays demoted with hooks disabled; the untruncated suffix is then * reconciled by rejoin reconciliation at the next join (crash-safe by design). */ int ha_schedule_demotion_convergence(cluster_t *c) { if (c == NULL) return -EINVAL; ha_demotion_convergence_ctx_t *ctx = calloc(1, sizeof(*ctx)); if (ctx == NULL) { ERROR("HA demotion convergence: ctx alloc failed — suffix will be " "reconciled at next rejoin"); return -ENOMEM; } ctx->c = c; /* One-shot (runcount=1), minimal delay: the work must run promptly but off * the iomgr callback thread. delay_task_ms only enqueues (ordq_put); the * delaymgr worker runs it. In XCE_TESTING the fixture drives the queued * task synchronously via its stored entry/args. */ int rc = delay_task_ms(&c->delaymgr, 1, 1, (delay_task_entry_fn_t)ha_demotion_convergence_fn, ctx); if (rc < 0) { ERROR("HA demotion convergence: delay_task_ms failed: %d", rc); free(ctx); return rc; } return 0; } /* * ha_rejoin_reconcile_ctx_t — heap context for the one-shot rejoin delaymgr task. * Carries the clamped cutoff computed at trigger time (under the same lock that * read the tip) plus the prior tip for the WARN range, so the deferred body does * not re-read state that may have advanced between schedule and run. */ typedef struct { cluster_t *c; uint64_t cutoff; /* truncate above this LSN (clamped, Req 2.2) */ uint64_t prior_tip; /* local tip at trigger time (WARN range, Req 2.1) */ } ha_rejoin_reconcile_ctx_t; /* * ha_rejoin_reconcile_fn — delaymgr task body (runs OFF the CtrlHelloReply intake * thread, which may be iomgr; D5/D1). Truncate above the clamped cutoff (Req 2.1, * 2.2), WARN with the discarded range, then ha_self_restart (Req 2.1). On * truncation failure ERROR then restart anyway — boot recovery + a subsequent * rejoin converge the untruncated case (crash-safe by design, mirrors demotion). */ int ha_rejoin_reconcile_fn(void *arg) { ha_rejoin_reconcile_ctx_t *ctx = (ha_rejoin_reconcile_ctx_t *)arg; if (ctx == NULL || ctx->c == NULL) { free(ctx); return -EINVAL; } cluster_t *c = ctx->c; uint64_t cutoff = ctx->cutoff; uint64_t prior_tip = ctx->prior_tip; int trc = persist_truncate_wal_above(&c->persist, cutoff); if (trc < 0) { ERROR("HA rejoin reconcile: truncate above cutoff=%" PRIu64 " failed: %d — restarting anyway (boot recovery + next rejoin will " "converge)", cutoff, trc); } else { /* Req 2.1: log the discarded range — (cutoff, prior tip]. */ WARN("HA rejoin reconcile: local tip=%" PRIu64 " exceeded the cluster " "committed boundary; truncated WAL above cutoff=%" PRIu64 " (clamped to the locally committed prefix), restarting once to " "rejoin from converged state", prior_tip, cutoff); } /* Free the task ctx before restart (production exec never returns). */ free(ctx); /* Req 2.1: rebuild in-memory state through boot recovery. */ return ha_self_restart(c); } /* * ha_rejoin_reconcile — converge a rejoining node whose local WAL log runs ahead * of the cluster's committed boundary (ha-convergence-recovery Req 2.1–2.3, D5). * * Called from the CtrlHelloReply intake (cluster_apply_ctrlhello_reply) after the * primary-reported committed boundary is stored — the moment the rejoining node * learns where the cluster has committed. * * Decision (design "Rejoin Reconciliation", research D5): * - local_tip = persist.current_lsn - 1. * - If local_tip <= primary_committed_lsn → return 0, NO side effects (Req 2.3). * - If the rejoin_reconciled once-guard is already set → return 0; never restart * a second time in one process lifetime (Req 2.1 defensive guard; "once" is * also structural — after restart the tip equals the cutoff, so the condition * cannot re-fire for the same divergence). * - Else: cutoff = max(primary_committed_lsn, local committed_lsn) — the CLAMP * (Req 2.2): never truncate at or below the locally known committed prefix. * Set the once-guard, then defer truncation + restart to a delaymgr one-shot * (the intake may run on iomgr, so file I/O / exec must leave it — D5/D1). * * Returns 0 on the no-action paths and on a successfully scheduled reconcile; * negative on a scheduling/allocation failure (the once-guard is rolled back so a * later rejoin can retry). */ int ha_rejoin_reconcile(cluster_t *c, uint64_t primary_committed_lsn) { if (c == NULL) return -EINVAL; /* Read the local tip + committed prefix together under the persist lock so * the clamp decision is consistent. */ pthread_mutex_lock(&c->persist.lock); uint64_t current_lsn = c->persist.current_lsn; pthread_mutex_unlock(&c->persist.lock); uint64_t local_tip = current_lsn > 0 ? current_lsn - 1 : 0; /* Req 2.3: tip does not exceed the boundary → join with no truncation/restart. */ if (local_tip <= primary_committed_lsn) return 0; /* Req 2.1 (defensive once-guard): reconcile at most once per process lifetime. * CAS 0→1 so a concurrent/duplicate CtrlHelloReply cannot schedule twice. */ int expected = 0; if (!atomic_compare_exchange_strong_explicit( &c->ha.rejoin_reconciled, &expected, 1, memory_order_acq_rel, memory_order_acquire)) return 0; /* already reconciled — no second restart */ /* Req 2.2 (clamp): never truncate at/below the locally known committed prefix. * cutoff = max(primary_committed_lsn, local committed_lsn). */ uint64_t local_committed = atomic_load_explicit(&c->persist.committed_lsn, memory_order_acquire); uint64_t cutoff = primary_committed_lsn > local_committed ? primary_committed_lsn : local_committed; ha_rejoin_reconcile_ctx_t *ctx = calloc(1, sizeof(*ctx)); if (ctx == NULL) { ERROR("HA rejoin reconcile: ctx alloc failed — divergent suffix will be " "reconciled at the next rejoin"); atomic_store_explicit(&c->ha.rejoin_reconciled, 0, memory_order_release); /* roll back: allow retry */ return -ENOMEM; } ctx->c = c; ctx->cutoff = cutoff; ctx->prior_tip = local_tip; /* D5/D1: defer truncation + restart off the (possibly iomgr) intake thread. * One-shot, minimal delay — mirrors ha_schedule_demotion_convergence. Under * XCE_TESTING the fixture drives the queued task synchronously. */ int rc = delay_task_ms(&c->delaymgr, 1, 1, (delay_task_entry_fn_t)ha_rejoin_reconcile_fn, ctx); if (rc < 0) { ERROR("HA rejoin reconcile: delay_task_ms failed: %d", rc); free(ctx); atomic_store_explicit(&c->ha.rejoin_reconciled, 0, memory_order_release); /* roll back: allow retry */ return rc; } return 0; } /* * cluster_apply_ctrlhello_reply — standby-side CtrlHelloReply intake body. * * Lives in this (unit-linkable) TU rather than cluster.c so the rejoin * intake-integration test can drive the real path: cluster.c (the daemon-only * vtask handler graph) is NOT part of controller_unit_test_src, but * ha_convergence.c is. cluster.c's myreq_handler calls this on every * CtrlHelloReply; the function stores the primary-reported committed boundary + * config tag (existing behavior), then is the rejoin trigger point. * * Rejoin trigger (Req 2.1–2.3, design "Rejoin Reconciliation", research D5): * AFTER the store, ha_rejoin_reconcile(c, committed_lsn) evaluates the local tip * against the primary-reported boundary and (when the tip runs ahead) defers a * clamped truncate + one-time restart. ha_rejoin_reconcile is internally a * no-op when the tip is at/below the boundary or already reconciled, so it is * safe to call on every CtrlHelloReply. */ void cluster_apply_ctrlhello_reply(cluster_t *c, uint64_t committed_lsn, uint64_t config_tag) { atomic_store_explicit(&c->ha.last_known_committed_lsn, committed_lsn, memory_order_release); atomic_store_explicit(&c->ha.last_known_committed_config_tag, config_tag, memory_order_release); INFO("HA: CtrlHelloReply from primary: committed_lsn=%" PRIu64 " config_tag=%" PRIu64, committed_lsn, config_tag); /* Req 2.1–2.3: reconcile a log that runs ahead of the cluster boundary. */ (void)ha_rejoin_reconcile(c, committed_lsn); } /* * ha_log_quorum_transition — write-quorum loss/restore edge logger * (ha-convergence-recovery task 5.2, Req 8.1, 8.2, 8.3). * * Evaluate the current write-quorum state and compare it against the * ha_state_t.quorum_lost edge-detector flag. On a TRANSITION — and only on a * transition — atomically flip the flag via CAS and emit exactly one WARN * carrying the connected/required member counts: * - loss: "HA quorum lost: connected=%d required=%d" (Req 8.1) * - restore: "HA quorum restored: connected=%d required=%d" (Req 8.2) * * The flag makes the logging an edge detector, NOT a latch: repeated calls while * the state is unchanged emit nothing (Req 8.3 — no log storms across the * dispatch-gate backstop's per-rejection calls), but a later loss/restore cycle * logs again. The CAS guarantees once-per-transition under concurrency: only the * caller that actually flips the flag emits the WARN. * * Observability only — this never returns a status and never gates an operation * outcome; the dispatch gate's existing -XCE_ERR_NOT_READY behavior is unchanged. * * connected/required mirror the ha_has_write_quorum BODY (ha.c): with the repl * back-pointer wired, connected = repl_active_standby_count(repl) and the * threshold required = n_peers/2 (integer division — for n_peers=3 → 1). When * the back-pointer is NULL (legacy contexts), connected falls back to the * standby_endpoints count, matching the gate's own fallback exactly. The quorum * decision math is NOT changed here — this only reports what that body decides. */ void ha_log_quorum_transition(ha_state_t *ha) { if (ha == NULL) return; /* Single-node clusters always hold quorum (ha_has_write_quorum returns true * for n_peers <= 1); there is no transition to observe. */ if (ha->n_peers <= 1) return; /* Mirror ha_has_write_quorum's connected-count derivation (ha.c body). */ int connected; if (ha->repl != NULL) connected = repl_active_standby_count(ha->repl); else connected = ha->standby_endpoints ? ha->standby_endpoints->count : 0; int required = (int)(ha->n_peers / 2); int lost_now = (connected < required) ? 1 : 0; /* Edge detection: only the caller that flips quorum_lost from its prior value * to lost_now emits. CAS expected=!lost_now → desired=lost_now; if the flag * already equals lost_now (no transition) the CAS fails and we return * silently (Req 8.3). */ int expected = lost_now ? 0 : 1; if (!atomic_compare_exchange_strong_explicit( &ha->quorum_lost, &expected, lost_now, memory_order_acq_rel, memory_order_acquire)) return; /* no edge (or another thread already logged it) — no WARN */ if (lost_now) WARN("HA quorum lost: connected=%d required=%d", connected, required); else WARN("HA quorum restored: connected=%d required=%d", connected, required); #ifdef XCE_TESTING /* Emission-count seam (Req 8.1-8.3 observable): invoked once per WARN. */ if (ha_quorum_transition_log_hook != NULL) ha_quorum_transition_log_hook(lost_now, connected, required); #endif } #ifdef XCE_TESTING /* ─────────────────────────────────────────────────────────────────────────── * ha_walack_continuation_for_test — task 4.2 GREEN. * * Test-only shim onto the REAL post-ACK catch-up decision ha_walack_on_ack * (config_repl.c, D6). The production walAck branch of * repl_ack_decode_and_dispatch (ha.c) calls the SAME ha_walack_on_ack after * repl_on_ack, so production and the unit tests exercise identical logic * (derive target = max(committed_lsn, last_written); behind target → schedule * one single-flight continuation; at parity → repl_complete_catchup, Req * 5.1/5.2/7.1). This shim exists only because the production caller decodes a * real Cap'n Proto WalAckReply off the wire, which is too heavy to drive from a * unit fixture. * ─────────────────────────────────────────────────────────────────────────── */ int ha_walack_continuation_for_test(cluster_t *c, void *ep, int peer_idx, uint64_t acked_lsn) { return ha_walack_on_ack(c, ep, peer_idx, acked_lsn); } #endif /* XCE_TESTING */