422 lines
19 KiB
C
422 lines
19 KiB
C
/*
|
||
* 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 <errno.h>
|
||
#include <inttypes.h>
|
||
#include <pthread.h>
|
||
#include <stdatomic.h>
|
||
#include <stddef.h>
|
||
#include <stdint.h>
|
||
#include <stdlib.h>
|
||
#include <unistd.h>
|
||
|
||
#include <tlog.h>
|
||
|
||
#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);
|
||
c->ha.last_known_committed_config_tag = config_tag;
|
||
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 — RED scaffold (ha-convergence-recovery task 5.1).
|
||
*
|
||
* Task 5.2 replaces this body with the real edge detector: evaluate the current
|
||
* write-quorum state (ha_has_write_quorum, ha.c), compare against ha->quorum_lost,
|
||
* and on a transition CAS-flip the flag and emit exactly one WARN with the
|
||
* connected/required counts (Req 8.1, 8.2), invoking ha_quorum_transition_log_hook
|
||
* once under XCE_TESTING. Repeated calls with no state change emit nothing
|
||
* (Req 8.3); a later loss/restore cycle logs again (edge detector, not a latch).
|
||
*
|
||
* This RED stub does NOTHING: no edge detection, no flag flip, no WARN, no hook
|
||
* invocation — so the task 5.1 quorum-transition tests FAIL until 5.2 lands.
|
||
*/
|
||
void ha_log_quorum_transition(ha_state_t *ha)
|
||
{
|
||
(void)ha; /* RED scaffold — task 5.2 implements the edge detection. */
|
||
}
|
||
|
||
#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 */
|