Design: HA WAL Replication

Overview

Bridge the WAL replicator and HA connection layer so WAL entries written on the primary are transmitted to standbys, ACKed, and committed via majority quorum. The existing repl_broadcast() is a no-send stub; this design fills in the send path and wires the standby registration/ACK lifecycle.

Architecture

 persist_wal_sync_append (persist.c)
   → repl_broadcast(payload) ───→ xce_wal_batch_req_raw(standby ep)
   → repl_wait_majority_ack() ──→ blocks until majority acked_lsn >= target

 Standby receive (cluster.c):
   OPCode_walCatchupEntry → handle_wal_catchup_entry
   → ha_apply_wal_catchup_entry → persist_wal_write_replicated (durable
     write; entity apply is now deferred to the commit-driven apply pump,
     not immediate — see note below)
   → send WalAckReply(lsn)  [helper: ha_send_wal_ack_reply, ha.c]

 Primary receive (ha.c):
   OPCode_reply + WalAckReply → repl_on_ack → signal ack_cond

Components

1. repl_broadcast send (repl.c)

Fill in the stub at repl.c:193-234. For each non-catch-up registered standby, serialize the WAL entry into a single-entry wal_catchup_entry_t with raw_payload from the caller's buffer, and enqueue via xce_wal_batch_req_raw(ep, &entry, 1).

The existing stub already iterates r->standbys[] and has the catch-up/pending_ack_count guards.
Replace the (void)payload; (void)payload_len; and s->pending_ack_count++ with actual transmission.
The wal_catchup_entry_t is stack-allocated; xce_wal_batch_req_raw copies the payload via capnp serialization (caller retains ownership).

2. Standby registration bridge (ha.c)

Wire repl_add_standby / repl_remove_standby into the HA connection lifecycle:

Register on connect: In raft_cb_become_leader (ha.c:~513-529), when promoting vote eps to ha->standby_endpoints, also call repl_add_standby(&c->persist.repl, ep, host, port). In on_standby_connected (ha.c), when a late-joiner connects, call repl_add_standby.
Deregister on disconnect: In vote_ep_err_handler (ha.c) and standby_ep_err_handler_primary (ha.c), call repl_remove_standby(&c->persist.repl, ep).

The replicator accesses c->persist.repl (already initialized in persist_init).

3. Standby ACK reply (cluster.c / ha.c)

After the standby applies a WAL entry (in ha_apply_single_entry or its caller), send a WalAckReply to the primary via xce_reply:

The WalAckReply carries term = ha->term, lsn = applied_entry->lsn, success = 1.
Use RequestReply_data_walackreplymsg union arm.
Sent from ha_apply_wal_catchup_entry after successful application of each entry in the batch.

4. Primary ACK handler (ha.c)

In repl_ack_decode_and_dispatch (ha.c:2506), add handling for RequestReply_data_walackreplymsg:

Extract the ACKed LSN from WalAckReply.
Map the sending endpoint to a peer index via repl_ep_t comparison.
Call repl_on_ack(&c->persist.repl, ep, acked_lsn).

5. WAL entry format

Reuse OPCode_walCatchupEntry with single-entry mode (n_entries=0):

Field	Source
lsn	`hdr->lsn`
entry_type	`hdr->type`
raw_payload	`payload` buffer from caller
raw_payload_len	`payload_len`

This is backward-compatible with the existing standby-side handler (ha_apply_wal_catchup_entry in ha.c:986) and the raw wire encoder (xce_wal_batch_req_raw in request.c).

Wire Protocol

Primary                              Standby
  │                                    │
  │── OPCode_walCatchupEntry ──────→   │  (single entry, lsn + payload)
  │                                    │  → apply_wal_entry
  │   ←── OPCode_reply ────────────    │  (WalAckReply { lsn, success })
  │   → repl_on_ack                    │
  │   → signal majority_ack_cond       │

File Structure Plan

File	Action	Responsibility
`src/controller/repl.c`	MODIFY	Fill in `repl_broadcast()` TCP send
`src/controller/ha.c`	MODIFY	Register/deregister standbys with replicator; handle incoming WalAckReply
`src/controller/cluster.c`	MODIFY	Send WalAckReply after WAL entry applied
`src/controller/testsuite/test_ha_wal_repl.c`	CREATE	Unit tests for broadcast send, ACK round-trip, registration lifecycle
`.kiro/specs/ha-wal-replication/tasks.md`	CREATE	Implementation task list

Testing Strategy

Unit (broadcast send): Create a mock endpoint, call repl_broadcast with a test entry, verify xce_wal_batch_req_raw is invoked with correct LSN and payload.
Unit (registration): Call ha_discover_primary path that adds a standby, verify r->standbys[i].ep != NULL and repl_broadcast iterates it.
Unit (ACK round-trip): Encode a WalAckReply, decode it on the primary side, verify repl_on_ack updates acked_lsn and signals the condition.
Integration: Start 3-node cluster, submit a job, verify WAL appears on all standbys, kill primary, verify promoted standby has the job.
ASAN: All new tests pass under ASAN with zero leaks.

Out of Boundary

Changes to repl_wait_majority_ack quorum logic (already correct).
Changes to the HA heartbeat-driven decision loop (WAL streaming for catching-up standbys is already handled by ha-repl-hardening).
Config replication path (uses OPCode_replConfigPush, not repl_broadcast).
WAL entry serialization format changes (reuses wal_catchup_entry_t).

Boundary Commitments

repl_broadcast transmits via xce_wal_batch_req_raw (single-entry mode), NOT a new custom wire format.
Standby registration uses the existing repl_add_standby / repl_remove_standby API — no new replicator functions.
The standby ACK uses the existing WalAckReply capnp struct — no schema changes needed.

5.8 KiB Raw Blame History