#pragma once /* config_repl.h — config storage, SHA-256, and replication functions */ #define _GNU_SOURCE #include #include #include #include "myconf.h" #include "clusterconfig.h" #include "netio.h" #include "request.h" #include "wal_types.h" /* WAL_CONFIG_MARKER, WAL_CONFIG_MARKER_PAYLOAD_LEN */ /* Forward decl: ha_stream_wal_to_standby takes a persist handle by pointer only * (for the authoritative per-LSN term index via persist_entry_term). Avoids a * heavy persist.h include in this header. */ struct persist_s; typedef struct persist_s persist_t; /* Max config blob size — reject before allocating */ #define CONF_MAX_SIZE (1 * 1024 * 1024) /* 1 MiB */ /* Sentinel name for the synthetic lost-and-found queue and host */ #define XCE_LOST_AND_FOUND_NAME "lost_and_found" /* Compile-time gate for OP_upgradeConfig (enabled: standbys apply pushed config) */ #ifndef CONF_ENABLE_UPGRADE_CONFIG #define CONF_ENABLE_UPGRADE_CONFIG 1 #endif /* Push timeout for standby fan-out (ms) */ #ifndef CONF_PUSH_TIMEOUT_MS #define CONF_PUSH_TIMEOUT_MS 500 #endif /* Certificate expiry warning threshold (days) */ #define CERT_WARN_DAYS 30 /* ── WAL_CONFIG_MARKER codec (ha-term-safety task 4.1) ─────────────── * * A config-change marker payload is a *reference* to a configuration — its * version tag and full SHA-256 content hash — and carries NO configuration * bytes (R7.2/R7.4). The bytes always travel out-of-band on the config-push * transport; they are never placed in a WAL payload. This codec produces the * fixed, padding-independent on-wire form (big-endian version_tag + raw 32-byte * sha256) that rides inside the raft_entry_blob_t envelope as the payload of a * WAL_CONFIG_MARKER entry. */ /* * config_marker_encode: serialize a {version_tag, sha256} config-change marker * into a fixed WAL_CONFIG_MARKER_PAYLOAD_LEN-byte buffer. NO config bytes are * ever written. version_tag is emitted big-endian; sha256 is the raw 32 bytes. * * buf — output buffer (must be >= WAL_CONFIG_MARKER_PAYLOAD_LEN) * bufsz — capacity of buf * out_len — receives the number of bytes written (always * WAL_CONFIG_MARKER_PAYLOAD_LEN on success); untouched on failure * * Returns 0 on success; -EINVAL if sha256/buf/out_len is NULL; * -ENOSPC if bufsz < WAL_CONFIG_MARKER_PAYLOAD_LEN. */ int config_marker_encode(uint64_t version_tag, const uint8_t sha256[32], uint8_t *buf, size_t bufsz, size_t *out_len); /* * config_marker_decode: deserialize a WAL_CONFIG_MARKER payload produced by * config_marker_encode. Reads exactly WAL_CONFIG_MARKER_PAYLOAD_LEN bytes; a * shorter buffer is rejected without any out-of-bounds read. * * buf — input buffer * len — bytes available in buf * version_tag — receives the host-order version tag (may be NULL to skip) * sha256_out — receives the raw 32-byte content hash (may be NULL to skip) * * Returns 0 on success; -EINVAL if buf is NULL or len < * WAL_CONFIG_MARKER_PAYLOAD_LEN. */ int config_marker_decode(const uint8_t *buf, size_t len, uint64_t *version_tag, uint8_t sha256_out[32]); /* ── Config marker append / committed-config derive (task 4.2) ─────── * * A configuration change is ordered and committed through the SAME replicated * (term,lsn) log as job state: a lightweight WAL_CONFIG_MARKER carrying only a * reference to the configuration ({version_tag, sha256}). The committed * configuration is DERIVED from the latest committed marker within the committed * prefix — there is no separate authoritative config commit index (R7.1/R7.7). */ /* * config_append_marker: append a config-change marker (version_tag + sha256, * NO config bytes — R7.2/R7.4) to the unified replicated log via the single * primary append entry point (ha_primary_log_append → raft_log_append), so the * marker inherits the live-term stamp (R7.3), the same commit rule (R7.11), and * the same (term,lsn) reconcile/truncate (R7.6) as a job-state entry. The * configuration bytes travel out-of-band on the config-push transport (R7.5), * never inside the marker. * * c — cluster (must be the leader for a real raft append) * version_tag — the config version tag (be64 of the first 8 sha256 bytes) * sha256 — the full 32-byte content hash (the reference) * out_lsn — (may be NULL) receives the assigned 1-based LSN on success * * Returns 0 on success; -EINVAL on a NULL argument; otherwise the append rc * propagated from ha_primary_log_append. */ int config_append_marker(cluster_t *c, uint64_t version_tag, const uint8_t sha256[32], uint64_t *out_lsn); /* * config_committed_marker: derive the committed configuration reference from the * LATEST WAL_CONFIG_MARKER at or below committed_lsn (R7.7). Scans the committed * prefix of the local WAL; an uncommitted marker (LSN > committed_lsn) is * deliberately ignored, so a config change is not the committed config until its * marker reaches the committed prefix. * * c — cluster * version_tag — (may be NULL) receives the latest committed marker's tag * sha256_out — (may be NULL) receives the latest committed marker's 32-byte hash * * Returns 0 on success; -ENOENT when no committed config marker exists; * -EINVAL on a NULL cluster; -EIO on a WAL read error. */ int config_committed_marker(const cluster_t *c, uint64_t *version_tag, uint8_t sha256_out[32]); /* ── Config byte delivery: hash-verify-before-activate + serve-fence (task 4.3) * * The committed WAL_CONFIG_MARKER (config_committed_marker) is the AUTHORITY for * which configuration is in effect; the configuration BYTES travel out-of-band on * the existing config-push transport (OPCode_replConfigPush / handle_repl_config_push), * never inside a WAL payload (R7.4/R7.5). config_activate_committed is the consumer * that keys that byte delivery to the marker's {version_tag, sha256}: it obtains the * bytes (already saved locally by the config-push handler when present), VERIFIES * their sha256 against the committed marker BEFORE activating (R7.8), and on a * mismatch (or bytes not yet present locally) keeps the member from serving under * the new configuration and KEEPS RETRYING rather than activating mismatched config * (R7.9) — the "controller never self-terminates" loud non-serving fence. On * recovery it activates the configuration referenced by the LATEST committed marker * (R7.10). */ /* * config_activate_committed: ensure the locally-held config bytes match the latest * committed WAL_CONFIG_MARKER's content hash, then activate (load) that config. * * Behaviour: * - No committed marker (config_committed_marker == -ENOENT): nothing to * activate; clears any serve-fence and returns 0 (the legacy bootstrap config * stays in effect). * - Locally-held bytes' sha256 == marker sha256: HASH-VERIFIED → activate via * load_primary_config, clear the serve-fence, return 0 (R7.8/R7.10). * - Locally-held bytes absent or sha256 != marker sha256: do NOT activate; * set the loud serve-fence (member stays non-serving under the new config, * stays ALIVE) and return -EAGAIN so the caller retries when the matching * bytes arrive out-of-band (R7.9). NEVER activates mismatched bytes; NEVER * self-terminates. * * c — cluster (uses c->myconf for the locally-held bytes, c->ha for the fence) * * Returns 0 on activate (or no-op when no marker); -EAGAIN while fenced * (bytes missing/mismatched); -EINVAL on a NULL/misconfigured cluster; the * load_primary_config rc if the verified bytes fail to load. */ int config_activate_committed(cluster_t *c); /* * config_serve_fenced: 1 when the member is serve-fenced under config_activate_- * committed (locally-held config bytes do not yet hash-match the latest committed * marker), 0 otherwise. The fence is non-serving + non-self-terminating (R7.9). */ int config_serve_fenced(const cluster_t *c); /* * config_schedule_activate_retry: when commit advances WHILE this member is * config serve-fenced, schedule a one-shot OFF-IOMGR delaymgr task that re-runs * config_activate_committed (task 4.3 r2, Finding D / R7.9). Closes the * push-before-commit liveness gap: the config MARKER (WAL) and config BYTES * (out-of-band push) arrive on independent transports, so a standby that saves * the NEW bytes before the NEW marker commits fences against the OLDER committed * marker; nothing else re-runs activation post-commit (on_commit_advanced must * not do disk I/O on the iomgr ACK thread; the primary's config-behind re-push * net does not fire once conf_save_config advanced the advertised tag). This * scheduler is NON-BLOCKING (ordq_put only) — safe to call from the commit-advance * path; the delaymgr worker does the disk reads. No-op when not serve-fenced or * when the delaymgr is not running (unit-test fixture). Returns 0 on enqueue/no-op, * negative on enqueue failure. */ int config_schedule_activate_retry(cluster_t *c); /* * config_activate_retry_fn: the delaymgr task body scheduled by * config_schedule_activate_retry (runs OFF the iomgr ACK thread, may block on * disk). Re-runs config_activate_committed; clears the serve-fence once the * newly-committed marker's bytes verify. Exposed so tests can drive the same * off-thread activation entry the production scheduler enqueues. `arg` is the * cluster_t*. Returns the config_activate_committed rc (0 on activate/clear, * -EAGAIN while still fenced). */ int config_activate_retry_fn(void *arg); /* * config_commit_change: order + commit a config change through the unified * (term,lsn) log (ha-term-safety task 4.2 LIVE wiring). Appends a * WAL_CONFIG_MARKER as the consensus record, pushes the config bytes * out-of-band on the existing config-push transport (R7.5 — byte delivery / * hash-verify / activate is task 4.3), and waits for the MARKER's LSN to reach * the committed prefix via the same WAL commit barrier job state uses (R7.11). * On commit it refreshes committed_config_tag as a NON-AUTHORITATIVE cache that * mirrors the DERIVED committed marker (config_committed_marker is the * authority — R7.7). do_upgrade_config_with_bytes uses this instead of the * retired term-blind committed_config_tag push/ack/advance. * * c — cluster (must be leader for the marker append) * bytes/blen — re-encoded config bytes (pushed out-of-band, never in the WAL) * sha256 — content hash (the marker reference) * version_tag — config version tag * acked_out — (may be NULL) receives the out-of-band transport push count * * Returns 0 once the marker is committed; the barrier rc (-ETIMEDOUT/-EINVAL) * when it never commits; a negative append rc on append failure. */ int config_commit_change(cluster_t *c, const uint8_t *bytes, size_t blen, const uint8_t sha256[32], uint64_t version_tag, int *acked_out); /* ── Phase 1: Storage ────────────────────────────────────────── */ /* * conf_read_current_filename: read HEAD-style pointer file at * /conf/current. Strips trailing newline. Validates format: * must start with "cluster.bin." followed by exactly 12 hex chars. * Returns 0 on success; -ENOENT if absent; -ENODATA if empty (zero-length); * -EINVAL if malformed (path separator, wrong prefix, wrong hex length). * out must be at least NAME_MAX bytes. */ int conf_read_current_filename(const char *workdir, char *out, size_t outsz); /* * conf_save_config: atomically write versioned file and update HEAD pointer. * 1. Writes /cluster.bin..tmp, fsyncs, renames to versioned path. * 2. Writes /current..tmp, fsyncs, renames to conf/current. * 3. fsyncs the conf/ directory. * 4. Trims old versions. * 5. Updates myconf->conffile (heap-allocated via asprintf; caller must free). * 6. Updates myconf->config_version_tag and myconf->config_sha256. * Returns 0 on success, -errno on failure. */ int conf_save_config(myconf_t *myconf, const uint8_t *bytes, size_t len, const uint8_t sha256[32]); /* * conf_read_raw: read versioned file at conffile path. * Allocates *bytes (caller must free). Enforces CONF_MAX_SIZE. * Returns 0 on success; -EFBIG if exceeds limit; -errno on I/O error. */ int conf_read_raw(const char *conffile, uint8_t **bytes, size_t *len); /* * conf_get_saved_tag: return first 8 bytes of SHA-256 of the current * versioned file in host byte order. Returns 0 if conf/current absent. * This is the "no config saved" zero-tag sentinel. */ uint64_t conf_get_saved_tag(const char *workdir); /* * conf_check_stale: returns 1 if the current versioned file's mtime is * older than stale_timeout_s seconds. Returns 0 if within timeout or * if stale_timeout_s == 0 (disabled). Returns 0 if conf/current absent. */ int conf_check_stale(const char *workdir, time_t stale_timeout_s); /* * conf_trim_old_versions: keep at most keep_n versioned cluster.bin.* files * sorted by mtime (newest first). Always preserves the file named by the * current conf/current pointer. Also removes stale .tmp files. */ void conf_trim_old_versions(const char *workdir, int keep_n); /* * config_contains_reserved_name: returns 1 if ClusterConfig contains a * queue or host named "lost_and_found". Returns 0 for NULL config. * Checks queues[i]->queue (queue name) and hconfigs[i]->name (host name). */ int config_contains_reserved_name(cluster_config_t *cfg); /* * create_lost_and_found: ensure the synthetic "lost_and_found" queue and * host exist before persist_recover() runs. Stub for Task 10. */ void create_lost_and_found(cluster_t *c); /* * handle_config_request: handle an OP_replConfig request from a standby. * Reads the current versioned config blob and sends it back. * Stub for Task 8 (full implementation in Task 9). */ int handle_config_request(cluster_t *c, endpoint_t *ep, xce_request_t *req); /* * handle_repl_config_push: standby handler for OPCode_replConfigPush. * Verifies SHA-256, saves config blob to disk, and reloads cluster_config. */ int handle_repl_config_push(cluster_t *c, endpoint_t *ep, xce_request_t *req); /* * push_config_transport_only: best-effort streaming fan-out of a config blob * to all connected standby endpoints — NO barrier wait (ha-commit-barrier- * rollout task 5.2 remediation r1). Spawns one push thread per registered * standby; the sends are non-blocking endpoint_enq()s, so the helper is * iomgr-safe and is the ONLY config-push entry callers reachable from * peer_ctrl_msg_handler / io_worker may use (election promotion, heartbeat- * driven convergence in ha.c, controller join in join.c). Standbys that * miss the push converge later via the OP_replConfig anti-entropy pull and * the heartbeat decision loop. * n_out (may be NULL): receives the registered standby count. * Returns the number of standbys that ACKed the push at the transport level. */ int push_config_transport_only(cluster_t *c, const uint8_t *bytes, size_t blen, const uint8_t sha256[32], uint64_t new_tag, int *n_out); /* * push_config_to_standbys: transport fan-out (push_config_transport_only) * PLUS the majority-quorum config ACK barrier on new_tag (ha-commit-barrier- * rollout task 5.2). The barrier blocks up to CONF_PUSH_TIMEOUT_MS — this * entry is for VTASK-CONTEXT callers ONLY (the OP_upgradeConfig path, * do_upgrade_config_with_bytes); it must never run on the iomgr epoll * thread (no path reachable from peer_ctrl_msg_handler / io_worker — those * callers use push_config_transport_only above). * acked_out (may be NULL): receives the number of standbys that successfully * received the push at the transport level (per-standby push count). * Returns the barrier result: 0 once 1 (self) + standbys with * acked_config_lsn >= new_tag form a majority of configured members * (n_peers <= 1 → immediate 0); -ETIMEDOUT when the quorum never forms * within CONF_PUSH_TIMEOUT_MS; -EINVAL if the commit engine is not wired. * On 0 committed_config_tag has been advanced to new_tag; on failure it is * untouched. The upgrade path treats the result as authoritative. * Only meaningful when CONF_ENABLE_UPGRADE_CONFIG is set, but always declared * so callers can reference it without an #if guard. */ int push_config_to_standbys(cluster_t *c, const uint8_t *bytes, size_t blen, const uint8_t sha256[32], uint64_t new_tag, int *acked_out); /* * handle_upgrade_config: handle an OP_upgradeConfig request from an admin tool. * xce_decode() now extracts sha256 + bytes into req->body.upgradeconfigrequestmsg * before freeing the capnp segment, so no raw packet copy is needed. * Only callable when CONF_ENABLE_UPGRADE_CONFIG is set. */ int handle_upgrade_config(cluster_t *c, endpoint_t *ep, xce_request_t *req); /* * ha_stream_wal_to_standby: batch-stream WAL entries [standby_lsn+1, committed_lsn] * to a standby endpoint via xce_request() (non-blocking). * * Accumulates WAL entries up to WAL_STREAM_BATCH_ENTRIES (256) entries or * WAL_STREAM_BATCH_BYTES (65536) bytes per batch, then enqueues each batch * as a single OPCode_walCatchupEntry message with n_entries set. * All batches for the full range are enqueued in a single invocation — never * blocks the vtaskmgr thread. * * A partial final batch (fewer entries than the batch limit) is sent immediately * without waiting for the batch to fill (Req 5.4). * * Parameters: * ep — outbound endpoint to the standby * wal_path — directory containing WAL segment files (seg-*.wal) * p — persist handle for the AUTHORITATIVE per-LSN term index * (persist_entry_term). Each streamed entry's term and the * AppendEntries prev_log_term MUST come from this index, NOT * the raw on-disk hdr.term — the standby's logOk and the * library send path both use persist_entry_term, which maps a * legacy-0 entry to WAL_LEGACY_TERM. Sourcing raw hdr.term (0) * for a legacy entry diverges from the standby's normalized * term (1) → logOk rejects forever (bug-1028). May be NULL only * in count-only unit tests that do not assert terms. * standby_lsn — standby's last applied LSN (stream starts at standby_lsn+1) * committed_lsn— primary's committed LSN (stream ends at committed_lsn inclusive) * * Returns 0 on success; negative on internal error (e.g., NULL ep). * Requirements: 5.1, 5.2, 5.4, 5.5 (ha-repl-hardening spec); bug-1028. * Boundary: WAL Streaming (config_repl.c) */ #define WAL_STREAM_BATCH_ENTRIES 256 #define WAL_STREAM_BATCH_BYTES 65536 /* bug-1067: leader_term is the streaming leader's CURRENT term — stamped onto * each streamed entry's leader_current_term so the standby intake uses it as the * AppendEntries header term (UpdateTerm/stale-check), distinct from each entry's * stored term (logOk/persist). Historical catch-up entries have a stored term * BELOW leader_term; using the stored term as the AE term stale-rejected them. */ int ha_stream_wal_to_standby(endpoint_t *ep, const char *wal_path, const persist_t *p, uint64_t leader_term, uint64_t standby_lsn, uint64_t committed_lsn); #define SNAPSHOT_CHUNK_SIZE 65536 int ha_send_snapshot_to_standby(endpoint_t *ep, const char *snap_dir, uint64_t checkpoint_lsn); uint64_t ha_compute_oldest_wal_lsn(const char *wal_path); /* * handle_ctrlhello: primary-side handler for OP_hello from a joining standby. * * Implements the state sync sequence (Req 6.5, 6.6, 6.7): * 1. repl_add_standby() + repl_start_catchup() * 2. Reply CtrlHelloReply{committed_lsn, needs_snapshot, config_tag} * 3. If needs_snapshot: persist_checkpoint() + stream OP_snapshotTransfer * 4. Stream OP_walCatchupEntry [applied_lsn+1 → committed_lsn] * 5. On catch-up complete: repl_complete_catchup() * 6. If config_tag mismatch: push_config_to_standbys() */ int handle_ctrlhello(cluster_t *c, endpoint_t *ep, xce_request_t *req); /* * sanitise_cn: copy at most maxlen printable ASCII chars from src to a * static buffer, replacing non-printable/control chars with '?'. * Returns pointer to the static buffer. NOT thread-safe (static storage). */ char *sanitise_cn(const char *src, size_t maxlen); #ifdef XCE_TESTING #include #include #include /* * xce_probe_config_tag_hook: test-only injection point for config tag probing. * When non-NULL, xce_probe_config_tag() calls this instead of the real * network probe, allowing unit tests to return controlled tag values * without real network connections. * Set by test infrastructure (e.g. test_ha_bootstrap_support.c) before tests run. */ extern int (*xce_probe_config_tag_hook)(const char *host, uint16_t port, uint64_t *tag_out); /* * load_primary_config_for_test: test-only wrapper that exercises the path * resolution logic of load_primary_config() without calling FATAL. * Returns 0 if conf/current resolves to a valid filename, negative errno * otherwise. Does NOT perform Cap'n Proto decode or SHA-256 computation. */ int load_primary_config_for_test(cluster_t *c); /* * push_ctx_t: exposed under XCE_TESTING so unit tests can construct * synthetic push contexts to drive wait_all_with_timeout_for_test(). * In production builds this type is private to config_repl.c. */ typedef struct { endpoint_t *ep; xce_peerinfo_t *from; uint64_t new_tag; const uint8_t *bytes; size_t blen; const uint8_t *sha256; int acked; int started; int done; } push_ctx_t; /* * wait_all_with_timeout_for_test: test-only shim exposing the static * wait_all_with_timeout() helper for unit tests. * Joins all started push threads within timeout_ms (CLOCK_MONOTONIC). * Sets *done = 1 when deadline expires. Returns number of acked threads. */ int wait_all_with_timeout_for_test(pthread_t *threads, push_ctx_t *ctxs, int n, uint32_t timeout_ms, atomic_int *done); #endif