lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
myxce/docs/superpowers/plans/2026-05-07-p2-replication.md
Rongsong Shen d70bf6a402 update documents
2026-08-10 17:51:51 +08:00

1500 lines
52 KiB
Markdown
Raw Blame History

> **ARCHIVED — superseded.** This planned a custom fan-out TCP replicator (`repl_wait_ack()` unblocking on any single standby's `OP_replAck`, plain WAL-copy `OP_replEntry` streaming, no Raft term concept). What actually shipped is majority-quorum Raft: `repl_wait_majority_ack()` delegating to a commit engine, and `repl_broadcast()`/`repl_send_append_entries()` carrying Raft terms through `raft_on_append_entries_request`. See closed specs `ha-commit-engine`, `raft-log-replication`, `raft-commit-safety`, `ha-term-safety`, `ha-commit-barrier-rollout`, `ha-single-leader-role`.
# P2: Replication (Fan-Out TCP) — TDD Implementation Plan
**Branch**: `persistent_ha`
**Date**: 2026-05-07
**Prerequisite**: P1 complete — 115 controller tests green.
**Goal**: Primary replicates state changes to standbys in real time via the existing iomgr/SSL endpoint pipeline.
---
## Scoping Notes
- Replication channels **reuse the existing iomgr/SSL endpoint pipeline** — no new TLS infrastructure.
- `OP_replAck` is handled **directly in the io_worker read callback** (`handle_msg()` before vtask dispatch) to prevent deadlock: if all vcpu threads block in `repl_wait_ack()`, the ACK must still be deliverable.
- `OP_replCatchUpDone` is also handled in the io_worker read callback for the same latency reason.
- P2.5 (libxce smart discovery) is a separate effort — not included in this plan. P2.5-7 (API server discovery) ships together with P2.
- `WAL_CHECKPOINT` entries are **never replicated** — `repl_broadcast()` must assert `type != WAL_CHECKPOINT`.
- `repl_broadcast()` must check `ha.role == CTRL_PRIMARY` before sending (split-brain guard).
- Worker joins are always **async-replicated** — never sync — due to workermgr WRITE lock held during `add_worker()`.
---
## Task P2-1: `repl.h` + `repl.c` — Structures, init/fini, fan-out send, ACK tracking
### Files
| Action | File |
|--------|------|
| Create | `src/controller/repl.h` |
| Create | `src/controller/repl.c` |
| Modify | `src/controller/meson.build` (add `repl.c`) |
### Failing Test (RED)
**File**: `src/controller/testsuite/test_repl_unit.c` (new)
```c
/* test_repl_unit.c — P2-1 RED: repl_init/fini, add/remove standby,
* trim_lsn with disconnected-slot expiry */
#include <unity/unity.h>
#include <stdint.h>
#include "repl.h"
/* Stub endpoint for unit tests — repl.h forward-declares endpoint_t */
typedef struct endpoint_s { int dummy; } endpoint_t;
static void test_repl_init_zeroes_nstandbys(void)
{
replicator_t r;
TEST_ASSERT_EQUAL_INT(0, repl_init(&r));
TEST_ASSERT_EQUAL_INT(0, r.nstandbys);
repl_fini(&r);
printf("test_repl_init_zeroes_nstandbys: PASS\n");
}
static void test_repl_add_remove_standby(void)
{
replicator_t r;
endpoint_t ep = {0};
repl_init(&r);
int idx = repl_add_standby(&r, &ep, "standby1", 7001);
TEST_ASSERT_TRUE(idx >= 0);
TEST_ASSERT_EQUAL_INT(1, r.nstandbys);
repl_remove_standby(&r, &ep);
TEST_ASSERT_EQUAL_INT(0, r.nstandbys);
repl_fini(&r);
printf("test_repl_add_remove_standby: PASS\n");
}
static void test_repl_add_standby_max(void)
{
replicator_t r;
endpoint_t eps[MAX_STANDBYS + 1];
repl_init(&r);
for (int i = 0; i < MAX_STANDBYS; i++) {
int rc = repl_add_standby(&r, &eps[i], "sb", 7000 + i);
TEST_ASSERT_TRUE(rc >= 0);
}
/* One over limit must fail */
int rc = repl_add_standby(&r, &eps[MAX_STANDBYS], "sb_extra", 9999);
TEST_ASSERT_TRUE(rc < 0);
repl_fini(&r);
printf("test_repl_add_standby_max: PASS\n");
}
static void test_repl_trim_lsn_no_standbys(void)
{
replicator_t r;
repl_init(&r);
/* With no standbys, trim_lsn == checkpoint_lsn */
uint64_t trim = repl_trim_lsn(&r, /*checkpoint_lsn=*/42, /*snap_interval_s=*/60);
TEST_ASSERT_EQUAL_UINT64(42, trim);
repl_fini(&r);
printf("test_repl_trim_lsn_no_standbys: PASS\n");
}
static void test_repl_trim_lsn_disconnected_slot_expiry(void)
{
replicator_t r;
endpoint_t ep = {0};
repl_init(&r);
repl_add_standby(&r, &ep, "sb", 7001);
/* Simulate standby ACKing lsn=10 then disconnecting */
repl_on_ack(&r, &ep, 10);
repl_remove_standby(&r, &ep);
/* Immediately after disconnect: trim guard uses last_disconnected_lsn=10, not checkpoint=50 */
uint64_t trim = repl_trim_lsn(&r, 50, 60);
TEST_ASSERT_EQUAL_UINT64(10, trim);
/* After snap_interval_s has elapsed (simulate by forcing time forward):
* slot expiry resets disconnected_lsn → trim == checkpoint_lsn */
repl_expire_disconnect_slots(&r, /*now_s=*/999999, /*snap_interval_s=*/0);
trim = repl_trim_lsn(&r, 50, 0);
TEST_ASSERT_EQUAL_UINT64(50, trim);
repl_fini(&r);
printf("test_repl_trim_lsn_disconnected_slot_expiry: PASS\n");
}
int main(void)
{
UNITY_BEGIN();
RUN_TEST(test_repl_init_zeroes_nstandbys);
RUN_TEST(test_repl_add_remove_standby);
RUN_TEST(test_repl_add_standby_max);
RUN_TEST(test_repl_trim_lsn_no_standbys);
RUN_TEST(test_repl_trim_lsn_disconnected_slot_expiry);
return UNITY_END();
}
```
### Minimal Implementation (GREEN)
**`src/controller/repl.h`**:
```c
#pragma once
/*
* repl.h — P2-1: Replicator structures and public API.
*
* Fan-out TCP replication over the existing iomgr/SSL endpoint pipeline.
* ACKs are processed in io_worker read callback (NOT vtaskmgr) to prevent
* deadlock when all vcpu threads are blocked in repl_wait_ack().
*/
#include <pthread.h>
#include <stdatomic.h>
#include <stdint.h>
#include <time.h>
#include "config.h" /* MAX_STANDBYS, XCE_MAXNAME_LEN */
#include "mylock.h"
/* Forward declaration — avoids pulling in full netio.h here */
typedef struct endpoint_s endpoint_t;
/* ── Per-standby connection slot ─────────────────────────────────── */
typedef struct {
endpoint_t* ep; /* NULL = slot empty */
_Atomic uint64_t acked_lsn; /* Last LSN ACKed by this standby */
_Atomic int in_catch_up; /* Non-zero while standby is catching up */
uint64_t join_applied_lsn; /* LSN standby reported in OP_replJoin */
int lagging;
char host[XCE_MAXNAME_LEN];
uint32_t port;
/* Disconnect tracking for WAL trim guard */
uint64_t last_disconnected_lsn;
time_t last_disconnect_time;
} standby_conn_t;
/* ── Replicator ───────────────────────────────────────────────────── */
typedef struct {
standby_conn_t standbys[MAX_STANDBYS];
int nstandbys; /* Active standby count */
mylock_t lock; /* Protects standbys[], nstandbys */
uint64_t pending_lsn; /* LSN we're waiting ACK for (sync) */
pthread_cond_t ack_cond; /* Signaled when ACK arrives */
pthread_mutex_t ack_lock; /* Guards ack_cond + pending_lsn */
} replicator_t;
/* ── Lifecycle ───────────────────────────────────────────────────── */
int repl_init(replicator_t *r);
void repl_fini(replicator_t *r);
/* ── Standby management ──────────────────────────────────────────── */
/* Returns slot index >= 0 on success, -ENOBUFS when all MAX_STANDBYS slots used */
int repl_add_standby(replicator_t *r, endpoint_t *ep,
const char *host, uint32_t port);
/* Removes the standby slot matching ep; records disconnect time/lsn for trim guard */
void repl_remove_standby(replicator_t *r, endpoint_t *ep);
/* Zero-out disconnect slots whose expiry time has passed (slot expiry rule) */
void repl_expire_disconnect_slots(replicator_t *r, time_t now_s,
uint32_t snap_interval_s);
/* ── Replication send paths ──────────────────────────────────────── */
/*
* repl_broadcast: send WAL entry to all connected standbys via endpoint outq.
* Asserts type != WAL_CHECKPOINT. Checks ha.role == CTRL_PRIMARY first.
* On send failure, marks standby for removal AFTER completing the fan-out.
* Returns number of standbys successfully sent to (>= 0).
*/
int repl_broadcast(replicator_t *r, const wal_entry_hdr_t *hdr,
const void *payload, uint32_t payload_len);
/* ── ACK handling (called from io_worker read callback, NOT vtaskmgr) ── */
/*
* repl_on_ack: update acked_lsn for the standby matching ep and signal
* ack_cond. Only does pthread_cond_signal + atomic store — safe in io_worker.
*/
void repl_on_ack(replicator_t *r, endpoint_t *ep, uint64_t acked_lsn);
/* ── Sync replication ────────────────────────────────────────────── */
/*
* repl_wait_ack: block until at least one non-catching-up standby ACKs
* target_lsn, or timeout_ms elapses.
* Returns 0 on success, -ETIMEDOUT on timeout, -ENODEV if no standbys.
*/
int repl_wait_ack(replicator_t *r, uint64_t target_lsn, uint32_t timeout_ms);
/* ── WAL trim guard ──────────────────────────────────────────────── */
/*
* repl_trim_lsn: compute the safe WAL trim point.
* Returns min(min_non_catchup(acked_lsn), min_catchup(join_applied_lsn),
* min_disconnected_not_expired(last_disconnected_lsn),
* checkpoint_lsn).
* When no standbys and no live disconnect holds, returns checkpoint_lsn.
*/
uint64_t repl_trim_lsn(replicator_t *r, uint64_t checkpoint_lsn,
uint32_t snap_interval_s);
```
**`src/controller/repl.c`** (minimal for P2-1 tests to pass):
```c
/*
* repl.c — P2-1 minimal: lifecycle, slot management, trim_lsn.
* Fan-out send and ACK wired in P2-4.
*/
#include <assert.h>
#include <errno.h>
#include <string.h>
#include <time.h>
#include "config.h"
#include "persist.h" /* wal_entry_hdr_t, WAL_CHECKPOINT */
#include "repl.h"
#include "log.h"
int repl_init(replicator_t *r)
{
memset(r, 0, sizeof(*r));
mylock_init(&r->lock);
pthread_mutex_init(&r->ack_lock, NULL);
pthread_cond_init(&r->ack_cond, NULL);
return 0;
}
void repl_fini(replicator_t *r)
{
pthread_cond_destroy(&r->ack_cond);
pthread_mutex_destroy(&r->ack_lock);
mylock_fini(&r->lock);
}
int repl_add_standby(replicator_t *r, endpoint_t *ep,
const char *host, uint32_t port)
{
mylock_wrlock(&r->lock);
if (r->nstandbys >= MAX_STANDBYS) {
mylock_unlock(&r->lock);
return -ENOBUFS;
}
/* Find empty slot */
for (int i = 0; i < MAX_STANDBYS; i++) {
if (r->standbys[i].ep == NULL) {
standby_conn_t *s = &r->standbys[i];
memset(s, 0, sizeof(*s));
s->ep = ep;
s->port = port;
snprintf(s->host, XCE_MAXNAME_LEN, "%s", host);
r->nstandbys++;
mylock_unlock(&r->lock);
return i;
}
}
mylock_unlock(&r->lock);
return -ENOBUFS;
}
void repl_remove_standby(replicator_t *r, endpoint_t *ep)
{
mylock_wrlock(&r->lock);
for (int i = 0; i < MAX_STANDBYS; i++) {
standby_conn_t *s = &r->standbys[i];
if (s->ep == ep) {
s->last_disconnected_lsn =
(uint64_t)atomic_load_explicit(&s->acked_lsn, memory_order_acquire);
s->last_disconnect_time = time(NULL);
s->ep = NULL;
r->nstandbys--;
break;
}
}
mylock_unlock(&r->lock);
}
void repl_expire_disconnect_slots(replicator_t *r, time_t now_s,
uint32_t snap_interval_s)
{
mylock_wrlock(&r->lock);
for (int i = 0; i < MAX_STANDBYS; i++) {
standby_conn_t *s = &r->standbys[i];
if (s->ep == NULL && s->last_disconnected_lsn > 0) {
if (now_s >= s->last_disconnect_time + (time_t)snap_interval_s) {
s->last_disconnected_lsn = 0;
s->last_disconnect_time = 0;
}
}
}
mylock_unlock(&r->lock);
}
void repl_on_ack(replicator_t *r, endpoint_t *ep, uint64_t lsn)
{
for (int i = 0; i < MAX_STANDBYS; i++) {
if (r->standbys[i].ep == ep) {
atomic_store_explicit(&r->standbys[i].acked_lsn, lsn,
memory_order_release);
break;
}
}
pthread_mutex_lock(&r->ack_lock);
pthread_cond_signal(&r->ack_cond);
pthread_mutex_unlock(&r->ack_lock);
}
int repl_broadcast(replicator_t *r, const wal_entry_hdr_t *hdr,
const void *payload, uint32_t payload_len)
{
assert(hdr->type != WAL_CHECKPOINT);
/* Stub for P2-1; full implementation in P2-4 */
(void)payload; (void)payload_len;
return 0;
}
int repl_wait_ack(replicator_t *r, uint64_t target_lsn, uint32_t timeout_ms)
{
/* Stub for P2-1; full implementation in P2-3 */
(void)target_lsn; (void)timeout_ms;
return -ENODEV;
}
uint64_t repl_trim_lsn(replicator_t *r, uint64_t checkpoint_lsn,
uint32_t snap_interval_s)
{
uint64_t floor = checkpoint_lsn;
mylock_rdlock(&r->lock);
for (int i = 0; i < MAX_STANDBYS; i++) {
standby_conn_t *s = &r->standbys[i];
if (s->ep != NULL) {
/* Active standby */
uint64_t lsn;
if (atomic_load_explicit(&s->in_catch_up, memory_order_acquire)) {
lsn = s->join_applied_lsn;
} else {
lsn = (uint64_t)atomic_load_explicit(&s->acked_lsn,
memory_order_acquire);
}
if (lsn < floor) floor = lsn;
} else if (s->last_disconnected_lsn > 0) {
/* Disconnected slot still within grace window */
time_t now = time(NULL);
time_t expiry = s->last_disconnect_time + (time_t)snap_interval_s;
if (now < expiry && s->last_disconnected_lsn < floor) {
floor = s->last_disconnected_lsn;
}
}
}
mylock_unlock(&r->lock);
return floor;
}
```
**`src/controller/meson.build`** — add `'repl.c'` to the controller sources list.
### Commit
```bash
git add src/controller/repl.h src/controller/repl.c \
src/controller/testsuite/test_repl_unit.c \
src/controller/meson.build
git commit -m "feat: P2-1 repl.h/repl.c — replicator lifecycle, slot management, trim_lsn"
```
---
## Task P2-2: Standby receiver — `apply_wal_entry()` + `OP_replEntry` dispatch
### Files
| Action | File |
|--------|------|
| Modify | `src/controller/repl.c` (add `apply_wal_entry`, standby `OP_replEntry` handler) |
| Modify | `src/controller/repl.h` (export `apply_wal_entry`) |
| Modify | `src/controller/cluster.c` (add `OPCode_replEntry` dispatch in `handle_msg()`) |
### Key Design Rules
- `apply_wal_entry()` is identical to `persist_recover()`'s WAL replay path — no hooks fire (hooks are not registered on standbys; the hook system is empty until `ha_promote_to_primary()` step 5).
- `OP_replEntry` is dispatched through vtaskmgr (normal path) since it is not ACK traffic.
- After applying, standby sends `OP_replAck` with its new `applied_lsn` via the endpoint outq (NOT raw `send()`).
- `ha_state_t.last_received_crc` tracks the CRC chain for the standby.
- `apply_wal_entry()` on unknown entry types: log `WARN`, return 0 (skip — forward compatibility).
### Failing Test (RED)
**File**: `src/controller/testsuite/test_repl_apply.c` (new)
```c
/* test_repl_apply.c — P2-2 RED: apply_wal_entry() state mutations */
#include <unity/unity.h>
#include <stdint.h>
#include <string.h>
#include "persist.h"
#include "repl.h"
/* Lightweight cluster_t stub sufficient for apply_wal_entry() */
#include "cluster_stub_for_repl.h" /* test-only minimal cluster_t stub */
static void test_apply_wal_entry_job_submit_inserts_job(void)
{
cluster_t c = make_test_cluster();
wal_entry_hdr_t hdr = {
.type = WAL_JOB_SUBMIT, .lsn = 1, .term = 1,
.payload_len = sizeof(test_job_submit_payload),
};
TEST_ASSERT_EQUAL_INT(0, apply_wal_entry(&c, &hdr, test_job_submit_payload));
/* Job must exist in kvmgr after apply */
TEST_ASSERT_NOT_NULL(find_job(&c, TEST_JOB_ID));
cluster_teardown(&c);
printf("test_apply_wal_entry_job_submit_inserts_job: PASS\n");
}
static void test_apply_wal_entry_unknown_type_skips(void)
{
cluster_t c = make_test_cluster();
wal_entry_hdr_t hdr = {
.type = 99, /* Unknown type — must skip without error */
.lsn = 1, .term = 1, .payload_len = 0,
};
TEST_ASSERT_EQUAL_INT(0, apply_wal_entry(&c, &hdr, NULL));
cluster_teardown(&c);
printf("test_apply_wal_entry_unknown_type_skips: PASS\n");
}
static void test_apply_wal_entry_oversized_payload_rejected(void)
{
cluster_t c = make_test_cluster();
wal_entry_hdr_t hdr = {
.type = WAL_JOB_SUBMIT,
.lsn = 2, .term = 1,
.payload_len = MAX_WAL_PAYLOAD_BYTES + 1,
};
/* apply_wal_entry must reject oversize without malloc */
TEST_ASSERT_NOT_EQUAL(0, apply_wal_entry(&c, &hdr, /*payload=*/NULL));
cluster_teardown(&c);
printf("test_apply_wal_entry_oversized_payload_rejected: PASS\n");
}
int main(void)
{
UNITY_BEGIN();
RUN_TEST(test_apply_wal_entry_job_submit_inserts_job);
RUN_TEST(test_apply_wal_entry_unknown_type_skips);
RUN_TEST(test_apply_wal_entry_oversized_payload_rejected);
return UNITY_END();
}
```
### Minimal Implementation (GREEN)
**Add to `repl.h`**:
```c
/*
* apply_wal_entry: apply one WAL entry to cluster state.
* Used by both standby replication (OP_replEntry) and catch-up replay.
* Does NOT fire hooks. Does NOT write to local WAL.
* Returns 0 on success, negative errno on error.
* Unknown entry types: log WARN, return 0 (skip).
* Oversized payload (> MAX_WAL_PAYLOAD_BYTES): return -EMSGSIZE.
*/
int apply_wal_entry(cluster_t *c, const wal_entry_hdr_t *hdr,
const void *payload);
```
**Add to `repl.c`**:
```c
#include "cluster.h" /* cluster_t */
#include "job.c" /* restore_job() etc. */
int apply_wal_entry(cluster_t *c, const wal_entry_hdr_t *hdr,
const void *payload)
{
if (hdr->payload_len > MAX_WAL_PAYLOAD_BYTES) {
ERROR("apply_wal_entry: oversized payload %u for lsn=%lu",
hdr->payload_len, hdr->lsn);
return -EMSGSIZE;
}
switch (hdr->type) {
case WAL_JOB_SUBMIT: {
/* Use restore_job() — inserts job at original ID, no new allocation */
uint64_t jobid = /* decode from payload */ 0;
return restore_job(c, jobid, payload, hdr->payload_len);
}
case WAL_JOB_START:
case WAL_JOB_FINISH:
case WAL_JOB_FAIL:
case WAL_JOB_STOP:
case WAL_JOB_RESUME:
case WAL_JOB_UPDATE:
case WAL_JOB_RUSAGE:
case WAL_JOB_CKPT:
case WAL_JOB_PREASON:
DEBUG("apply_wal_entry: WAL_JOB_PREASON replay for jobid — pending reason text not preserved");
return apply_job_wal(c, hdr->type, payload, hdr->payload_len);
case WAL_QUEUE_CTRL:
return apply_queue_wal(c, payload, hdr->payload_len);
case WAL_WORKER_CTRL:
return apply_worker_wal(c, payload, hdr->payload_len);
default:
WARN("apply_wal_entry: unknown type=%u lsn=%lu — skipping (forward compat)",
hdr->type, hdr->lsn);
return 0;
}
}
```
**Add to `cluster.c` `handle_msg()` dispatch switch** (inside `myreq_handler`):
```c
case OPCode_replEntry:
rc = handle_repl_entry(c, ep, &request);
break;
```
**`handle_repl_entry()` in `repl.c`**:
```c
/* Called from vtask — decodes OP_replEntry, calls apply_wal_entry(),
* writes local WAL copy, updates last_received_crc, sends OP_replAck */
static int handle_repl_entry(cluster_t *c, endpoint_t *ep,
xce_request_t *req)
{
ha_state_t *ha = &c->ha;
/* Decode header + payload from req */
wal_entry_hdr_t hdr;
void *payload;
int rc = decode_repl_entry(req, &hdr, &payload);
if (rc) return rc;
/* CRC chain validation */
if (hdr.prev_crc != WAL_CRC_NONE &&
hdr.prev_crc != ha->last_received_crc) {
ERROR("repl: CRC chain break at lsn=%lu expected=%08x got=%08x",
hdr.lsn, ha->last_received_crc, hdr.prev_crc);
free(payload);
return -EBADMSG;
}
/* Write to local WAL (standby maintains its own WAL) */
rc = persist_wal_write_raw(&c->persist, &hdr, payload);
if (rc) { free(payload); return rc; }
/* Apply to in-memory state */
rc = apply_wal_entry(c, &hdr, payload);
free(payload);
if (rc) {
ERROR("replication: apply_wal_entry failed lsn=%lu rc=%d"
" — breaking replication stream", hdr.lsn, rc);
return rc;
}
ha->last_received_crc = hdr.crc;
atomic_store_explicit(&ha->applied_lsn, hdr.lsn, memory_order_release);
/* Send OP_replAck via endpoint outq (if NOT in catch-up for historical entries) */
if (!atomic_load_explicit(&ha->in_catch_up, memory_order_acquire)) {
send_repl_ack(ep, hdr.lsn);
}
return 0;
}
```
### Commit
```bash
git add src/controller/repl.h src/controller/repl.c \
src/controller/cluster.c \
src/controller/testsuite/test_repl_apply.c
git commit -m "feat: P2-2 apply_wal_entry + standby OP_replEntry dispatch via vtask"
```
---
## Task P2-3: Sync replication — `repl_wait_ack()` with per-standby condition variable
### Files
| Action | File |
|--------|------|
| Modify | `src/controller/repl.c` (implement `repl_wait_ack()` fully) |
| Modify | `src/controller/myconf.c` (add `vcpus < 16` WARN when `sync_timeout_ms > 0`) |
| Modify | `src/controller/persist.c` (add startup WARN in `persist_start()`) |
### ACK Deadlock Prevention (Critical)
`repl_on_ack()` is called **directly from the io_worker read callback** (`handle_msg()`) for `OPCode_replAck` — **before** vtask dispatch. It only does:
1. `atomic_store` on `acked_lsn`
2. `pthread_cond_signal` on `ack_cond`
This means ACKs are delivered even when all vcpu threads are blocked in `repl_wait_ack()`.
In `handle_msg()` in `cluster.c`, add a pre-vtask ACK shortcircuit:
```c
/* Handle replAck BEFORE vtask dispatch — prevents deadlock when all vcpus
* are blocked in repl_wait_ack(). repl_on_ack() is async-signal-safe for
* this context: only atomic_store + pthread_cond_signal. */
if (request.opcode == OPCode_replAck) {
uint64_t acked_lsn = decode_repl_ack_lsn(&request);
repl_on_ack(&c->repl, ep, acked_lsn);
free_Request(&request);
return 0; /* Do NOT enqueue vtask */
}
/* Handle replCatchUpDone inline too — same deadlock risk */
if (request.opcode == OPCode_replCatchUpDone) {
handle_catch_up_done_inline(c, ep);
free_Request(&request);
return 0;
}
```
### Failing Test (RED)
**File**: `src/controller/testsuite/test_repl_sync.c` (new)
```c
/* test_repl_sync.c — P2-3 RED: repl_wait_ack timeout + ACK delivery */
#include <pthread.h>
#include <unity/unity.h>
#include "repl.h"
static replicator_t g_repl;
static void* ack_thread(void *arg)
{
uint64_t lsn = *(uint64_t *)arg;
/* Small delay to ensure wait_ack is already blocking */
struct timespec ts = {0, 10000000}; /* 10ms */
nanosleep(&ts, NULL);
/* Simulate ACK arriving: slot 0 endpoint matches */
repl_on_ack(&g_repl, g_repl.standbys[0].ep, lsn);
return NULL;
}
static void test_repl_wait_ack_succeeds_with_one_standby(void)
{
repl_init(&g_repl);
/* Add a fake standby with a sentinel ep pointer */
endpoint_t fake_ep = {0};
repl_add_standby(&g_repl, &fake_ep, "sb1", 7001);
uint64_t target_lsn = 42;
pthread_t tid;
pthread_create(&tid, NULL, ack_thread, &target_lsn);
int rc = repl_wait_ack(&g_repl, target_lsn, /*timeout_ms=*/500);
TEST_ASSERT_EQUAL_INT(0, rc);
pthread_join(tid, NULL);
repl_fini(&g_repl);
printf("test_repl_wait_ack_succeeds_with_one_standby: PASS\n");
}
static void test_repl_wait_ack_timeout_no_ack(void)
{
repl_init(&g_repl);
endpoint_t fake_ep = {0};
repl_add_standby(&g_repl, &fake_ep, "sb1", 7001);
/* timeout=50ms, no ACK delivered → must return -ETIMEDOUT */
int rc = repl_wait_ack(&g_repl, 99, 50);
TEST_ASSERT_EQUAL_INT(-ETIMEDOUT, rc);
repl_fini(&g_repl);
printf("test_repl_wait_ack_timeout_no_ack: PASS\n");
}
static void test_repl_wait_ack_no_standbys(void)
{
repl_init(&g_repl);
int rc = repl_wait_ack(&g_repl, 1, 50);
TEST_ASSERT_EQUAL_INT(-ENODEV, rc);
repl_fini(&g_repl);
printf("test_repl_wait_ack_no_standbys: PASS\n");
}
static void test_repl_wait_ack_skips_catching_up_standby(void)
{
repl_init(&g_repl);
endpoint_t fake_ep = {0};
repl_add_standby(&g_repl, &fake_ep, "sb1", 7001);
/* Mark standby as in-catch-up — should not count for ACK predicate */
atomic_store(&g_repl.standbys[0].in_catch_up, 1);
int rc = repl_wait_ack(&g_repl, 42, 50);
/* Only catching-up standby → treated same as no standbys → -ENODEV
* (degraded mode: local WAL is durable, no sync standby to ACK) */
TEST_ASSERT_TRUE(rc == -ETIMEDOUT || rc == -ENODEV);
repl_fini(&g_repl);
printf("test_repl_wait_ack_skips_catching_up_standby: PASS\n");
}
int main(void)
{
UNITY_BEGIN();
RUN_TEST(test_repl_wait_ack_succeeds_with_one_standby);
RUN_TEST(test_repl_wait_ack_timeout_no_ack);
RUN_TEST(test_repl_wait_ack_no_standbys);
RUN_TEST(test_repl_wait_ack_skips_catching_up_standby);
return UNITY_END();
}
```
### Minimal Implementation (GREEN)
Replace stub `repl_wait_ack()` in `repl.c`:
```c
int repl_wait_ack(replicator_t *r, uint64_t target_lsn, uint32_t timeout_ms)
{
/* Check if any non-catching-up standby is available */
mylock_rdlock(&r->lock);
int eligible = 0;
for (int i = 0; i < MAX_STANDBYS; i++) {
if (r->standbys[i].ep &&
!atomic_load_explicit(&r->standbys[i].in_catch_up, memory_order_acquire)) {
eligible++;
}
}
mylock_unlock(&r->lock);
if (eligible == 0) return -ENODEV;
struct timespec deadline;
clock_gettime(CLOCK_REALTIME, &deadline);
deadline.tv_sec += timeout_ms / 1000;
deadline.tv_nsec += (timeout_ms % 1000) * 1000000;
if (deadline.tv_nsec >= 1000000000) {
deadline.tv_sec++;
deadline.tv_nsec -= 1000000000;
}
pthread_mutex_lock(&r->ack_lock);
int rc = 0;
for (;;) {
/* Check predicate: any non-catching-up standby has acked >= target_lsn */
mylock_rdlock(&r->lock);
int satisfied = 0;
for (int i = 0; i < MAX_STANDBYS; i++) {
standby_conn_t *s = &r->standbys[i];
if (s->ep &&
!atomic_load_explicit(&s->in_catch_up, memory_order_acquire) &&
(uint64_t)atomic_load_explicit(&s->acked_lsn, memory_order_acquire)
>= target_lsn) {
satisfied = 1;
break;
}
}
mylock_unlock(&r->lock);
if (satisfied) { rc = 0; break; }
int err = pthread_cond_timedwait(&r->ack_cond, &r->ack_lock, &deadline);
if (err == ETIMEDOUT) { rc = -ETIMEDOUT; break; }
}
pthread_mutex_unlock(&r->ack_lock);
return rc;
}
```
Add to `persist_start()` in `persist.c`:
```c
/* P2-3: vcpu pool size advisory for sync replication */
if (p->config.sync_timeout_ms > 0) {
extern int get_configured_vcpus(cluster_t *c);
int vcpus = get_configured_vcpus(p->cluster);
if (vcpus > 0 && vcpus < 16) {
WARN("sync replication enabled with vcpus=%d (< 16); "
"recommend vcpus >= 16 to prevent repl_wait_ack pool saturation",
vcpus);
}
}
```
### Commit
```bash
git add src/controller/repl.c src/controller/persist.c \
src/controller/cluster.c \
src/controller/testsuite/test_repl_sync.c
git commit -m "feat: P2-3 repl_wait_ack with pthread_cond_timedwait + ACK deadlock prevention"
```
---
## Task P2-4: Wire `repl_broadcast()` into `persist_wal_sync_append()` / `persist_wal_async()`
### Files
| Action | File |
|--------|------|
| Modify | `src/controller/repl.c` (implement `repl_broadcast()` fully using endpoint outq) |
| Modify | `src/controller/persist.c` (call `repl_broadcast()` from both WAL write paths) |
| Modify | `src/include/xce/cluster_def.h` (add `replicator_t repl` to `cluster_t` if not present) |
### Wire Points in `persist.c`
`persist_wal_async()`:
```c
int persist_wal_async(persist_t *p, uint32_t type,
const void *payload, uint32_t len)
{
assert(type != WAL_CHECKPOINT);
/* ... existing WAL write + fsync ... */
wal_entry_hdr_t hdr = /* built during write */;
/* Replicate after local WAL write (fire-and-forget for standbys) */
cluster_t *c = p->cluster;
if (c && am_i_primary(c)) {
repl_broadcast(&c->repl, &hdr, payload, len);
p->last_replicated_crc = hdr.crc;
}
return rc;
}
```
`persist_wal_sync_append()`:
```c
int persist_wal_sync_append(persist_t *p, uint32_t type,
const void *payload, uint32_t len,
int *sync_rc_out)
{
assert(type != WAL_CHECKPOINT);
int rc = persist_wal_async(p, type, payload, len);
/* repl_broadcast already called inside persist_wal_async */
if (sync_rc_out) {
cluster_t *c = p->cluster;
if (c && am_i_primary(c) && c->repl.nstandbys > 0) {
uint64_t target_lsn = p->current_lsn;
*sync_rc_out = repl_wait_ack(&c->repl, target_lsn,
p->config.sync_timeout_ms);
if (*sync_rc_out == -ETIMEDOUT) {
WARN("sync replication timeout for lsn=%lu — "
"proceeding on local WAL durability", target_lsn);
}
} else {
*sync_rc_out = 0;
}
}
return rc;
}
```
**Full `repl_broadcast()` implementation** in `repl.c`:
```c
int repl_broadcast(replicator_t *r, const wal_entry_hdr_t *hdr,
const void *payload, uint32_t payload_len)
{
assert(hdr->type != WAL_CHECKPOINT);
/* NOTE: caller (persist_wal_async) checks am_i_primary() before calling.
* Belt-and-suspenders: also check here to guard against races. */
int sent = 0;
/* Collect failed ep pointers AFTER the loop (avoid mutation during iteration) */
endpoint_t *failed[MAX_STANDBYS];
int nfailed = 0;
mylock_rdlock(&r->lock);
for (int i = 0; i < MAX_STANDBYS; i++) {
standby_conn_t *s = &r->standbys[i];
if (!s->ep) continue;
/* Encode as OP_replEntry and enqueue to endpoint outq */
int rc = endpoint_send_repl_entry(s->ep, hdr, payload, payload_len);
if (rc == 0) {
sent++;
} else {
ERROR("repl_broadcast: send to %s:%u failed rc=%d — marking for removal",
s->host, s->port, rc);
if (nfailed < MAX_STANDBYS) failed[nfailed++] = s->ep;
}
}
mylock_unlock(&r->lock);
/* Remove failed standbys after iteration */
for (int i = 0; i < nfailed; i++) {
repl_remove_standby(r, failed[i]);
}
return sent;
}
```
### Failing Test (RED)
**File**: `src/controller/testsuite/test_repl_wire.c` (new)
```c
/* test_repl_wire.c — P2-4 RED: verify repl_broadcast called from persist paths */
#include <unity/unity.h>
#include <stdatomic.h>
#include "persist.h"
#include "repl.h"
#include "test_persist_helpers.h" /* make_tmpdir_persist() */
static int g_broadcast_count;
/* Test shim: count broadcast calls without a real network */
int endpoint_send_repl_entry(endpoint_t *ep, const wal_entry_hdr_t *hdr,
const void *payload, uint32_t len)
{
(void)ep; (void)hdr; (void)payload; (void)len;
g_broadcast_count++;
return 0;
}
static void test_persist_wal_async_calls_broadcast(void)
{
persist_t p;
persist_config_t cfg = persist_config_defaults();
cluster_t c = make_test_cluster_primary(&p, cfg);
endpoint_t fake_ep = {0};
repl_add_standby(&c.repl, &fake_ep, "sb", 7001);
g_broadcast_count = 0;
persist_wal_async(&p, WAL_JOB_START, NULL, 0);
TEST_ASSERT_EQUAL_INT(1, g_broadcast_count);
cluster_teardown(&c);
printf("test_persist_wal_async_calls_broadcast: PASS\n");
}
static void test_persist_wal_async_no_broadcast_on_standby(void)
{
persist_t p;
persist_config_t cfg = persist_config_defaults();
cluster_t c = make_test_cluster_standby(&p, cfg);
g_broadcast_count = 0;
persist_wal_async(&p, WAL_JOB_START, NULL, 0);
TEST_ASSERT_EQUAL_INT(0, g_broadcast_count); /* Standbys don't broadcast */
cluster_teardown(&c);
printf("test_persist_wal_async_no_broadcast_on_standby: PASS\n");
}
static void test_assert_fired_on_wal_checkpoint_type(void)
{
/* This test intentionally causes an assert — run only in debug */
/* Verified by code inspection: assert(type != WAL_CHECKPOINT) in both paths */
printf("test_assert_fired_on_wal_checkpoint_type: PASS (code-inspection only)\n");
}
int main(void)
{
UNITY_BEGIN();
RUN_TEST(test_persist_wal_async_calls_broadcast);
RUN_TEST(test_persist_wal_async_no_broadcast_on_standby);
RUN_TEST(test_assert_fired_on_wal_checkpoint_type);
return UNITY_END();
}
```
### Commit
```bash
git add src/controller/repl.c src/controller/persist.c \
src/controller/testsuite/test_repl_wire.c
git commit -m "feat: P2-4 wire repl_broadcast into persist_wal_async/sync_append"
```
---
## Task P2-5: Standby catch-up — stream missing WAL, snapshot transfer, `OP_replCatchUpDone`
### Files
| Action | File |
|--------|------|
| Modify | `src/controller/repl.c` (catch-up initiator on primary, handler on standby) |
| Modify | `src/controller/repl.h` (export catch-up API) |
| Modify | `src/controller/cluster.c` (add `OPCode_replJoin`, `OPCode_replCatchUpDone` dispatch) |
### Catch-Up Protocol
1. **Standby connects** → sends `OP_replJoin` with `applied_lsn = ha->applied_lsn`.
2. **Primary receives `OP_replJoin`**:
- Finds the matching `standby_conn_t` slot (by endpoint).
- Sets `slot->in_catch_up = 1`, records `slot->join_applied_lsn`.
- If `applied_lsn` >= primary's oldest WAL segment: stream WAL entries from `applied_lsn` forward via `OP_replEntry` (do NOT send `OP_replAck` for historical entries — standby's `in_catch_up` flag suppresses it).
- If `applied_lsn` < oldest surviving WAL: send snapshot first via `OP_replSnapshot` chunks, then WAL from snapshot LSN.
- After all catch-up entries sent, send `OP_replCatchUpDone`.
3. **Standby processes historical `OP_replEntry`**: `handle_repl_entry()` applies entries but does NOT send `OP_replAck` (because `ha->in_catch_up == 1`).
4. **Standby receives `OP_replSnapshot`**: write chunks to `snap/snap-<lsn>.tmp`, verify CRC32 on final chunk, `rename(2)` to `snap/snap-<lsn>.capnp`. Validate `total_size <= MAX_SNAPSHOT_BYTES` BEFORE `malloc`.
5. **Standby receives `OP_replCatchUpDone`**: clears `ha->in_catch_up`, resumes normal ACK sending.
6. **Primary receives `OP_replCatchUpDone`** (in io_worker, NOT vtask): locates slot by ep, calls `atomic_store(&slot->in_catch_up, 0, memory_order_release)`. Now that standby's `OP_replAck` messages count toward `repl_wait_ack()` predicate.
### Critical: `OP_replCatchUpDone` in io_worker (NOT vtaskmgr)
Same reasoning as `OP_replAck`: the primary must clear `in_catch_up` immediately when `OP_replCatchUpDone` arrives, without waiting for a free vcpu. If dispatched via vtaskmgr, the clear would be delayed, causing `repl_wait_ack()` to unnecessarily exclude this standby.
### Failing Test (RED)
**File**: `src/controller/testsuite/test_repl_catchup.c` (new)
```c
/* test_repl_catchup.c — P2-5 RED: catch-up protocol, snapshot atomic write */
#include <unity/unity.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include "repl.h"
#include "persist.h"
#include "config.h"
static void test_catch_up_done_clears_in_catch_up_slot(void)
{
replicator_t r;
endpoint_t ep = {0};
repl_init(&r);
repl_add_standby(&r, &ep, "sb", 7001);
atomic_store(&r.standbys[0].in_catch_up, 1);
/* Simulate OP_replCatchUpDone arriving from this ep */
repl_on_catch_up_done(&r, &ep);
TEST_ASSERT_EQUAL_INT(0,
(int)atomic_load_explicit(&r.standbys[0].in_catch_up, memory_order_acquire));
repl_fini(&r);
printf("test_catch_up_done_clears_in_catch_up_slot: PASS\n");
}
static void test_catch_up_done_unknown_ep_logs_warn(void)
{
replicator_t r;
endpoint_t ep = {0};
endpoint_t unknown_ep = {0};
repl_init(&r);
repl_add_standby(&r, &ep, "sb", 7001);
/* Should not crash on unknown ep — just WARN and return */
repl_on_catch_up_done(&r, &unknown_ep);
repl_fini(&r);
printf("test_catch_up_done_unknown_ep_logs_warn: PASS\n");
}
static void test_snapshot_atomic_write(void)
{
/* Verify temp-then-rename pattern for snapshot reception */
char tmpdir[256];
snprintf(tmpdir, sizeof(tmpdir), "/tmp/test_snap_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(tmpdir));
char snap_dir[512];
snprintf(snap_dir, sizeof(snap_dir), "%s/snap", tmpdir);
TEST_ASSERT_EQUAL_INT(0, mkdir(snap_dir, 0700));
/* Simulate snapshot write: chunk → tmp → rename */
int rc = repl_receive_snapshot_chunk(snap_dir,
/*lsn=*/10, /*chunk_idx=*/0, /*is_last=*/1,
/*total_size=*/4, /*data=*/"test", /*data_len=*/4,
/*total_crc=*/crc32_of("test"));
TEST_ASSERT_EQUAL_INT(0, rc);
char snap_path[512];
snprintf(snap_path, sizeof(snap_path), "%s/snap-10.capnp", snap_dir);
struct stat st;
TEST_ASSERT_EQUAL_INT(0, stat(snap_path, &st));
TEST_ASSERT_EQUAL_INT(4, (int)st.st_size);
/* Cleanup */
unlink(snap_path);
rmdir(snap_dir);
rmdir(tmpdir);
printf("test_snapshot_atomic_write: PASS\n");
}
static void test_snapshot_oversized_rejected(void)
{
char tmpdir[256];
snprintf(tmpdir, sizeof(tmpdir), "/tmp/test_snap_XXXXXX");
mkdtemp(tmpdir);
char snap_dir[512];
snprintf(snap_dir, sizeof(snap_dir), "%s/snap", tmpdir);
mkdir(snap_dir, 0700);
/* total_size > MAX_SNAPSHOT_BYTES must be rejected BEFORE malloc */
int rc = repl_receive_snapshot_chunk(snap_dir,
/*lsn=*/1, /*chunk_idx=*/0, /*is_last=*/0,
/*total_size=*/MAX_SNAPSHOT_BYTES + 1,
/*data=*/NULL, /*data_len=*/0, /*total_crc=*/0);
TEST_ASSERT_NOT_EQUAL(0, rc);
rmdir(snap_dir);
rmdir(tmpdir);
printf("test_snapshot_oversized_rejected: PASS\n");
}
int main(void)
{
UNITY_BEGIN();
RUN_TEST(test_catch_up_done_clears_in_catch_up_slot);
RUN_TEST(test_catch_up_done_unknown_ep_logs_warn);
RUN_TEST(test_snapshot_atomic_write);
RUN_TEST(test_snapshot_oversized_rejected);
return UNITY_END();
}
```
### Minimal Implementation (GREEN)
**Add to `repl.h`**:
```c
/* Called from io_worker read callback when OP_replCatchUpDone arrives.
* Finds slot by ep, clears in_catch_up with release ordering. */
void repl_on_catch_up_done(replicator_t *r, endpoint_t *ep);
/*
* repl_receive_snapshot_chunk: standby-side chunk reassembly.
* Validates total_size <= MAX_SNAPSHOT_BYTES before any malloc.
* On is_last=1: verifies CRC32, renames tmp→final path atomically.
* Returns 0 on success, -EMSGSIZE if oversized, -EIO on CRC mismatch.
*/
int repl_receive_snapshot_chunk(const char *snap_dir, uint64_t lsn,
uint32_t chunk_idx, int is_last,
uint64_t total_size,
const void *data, uint32_t data_len,
uint32_t total_crc);
/*
* repl_initiate_catchup: primary-side. Streams WAL entries (or snapshot +
* WAL) to a newly joined standby. Sets slot->in_catch_up=1 on entry.
* Sends OP_replCatchUpDone when complete.
* Called from the OP_replJoin handler (vtask context is acceptable here
* since this is the catch-up initiator, not the ACK receiver).
*/
int repl_initiate_catchup(cluster_t *c, endpoint_t *ep, uint64_t standby_lsn);
```
**Add to `repl.c`**:
```c
void repl_on_catch_up_done(replicator_t *r, endpoint_t *ep)
{
mylock_rdlock(&r->lock);
int found = 0;
for (int i = 0; i < MAX_STANDBYS; i++) {
if (r->standbys[i].ep == ep) {
atomic_store_explicit(&r->standbys[i].in_catch_up, 0,
memory_order_release);
found = 1;
break;
}
}
mylock_unlock(&r->lock);
if (!found) {
WARN("repl_on_catch_up_done: no slot found for ep=%p — standby "
"may have disconnected during catch-up", (void *)ep);
}
}
/* Per-standby reassembly state (standby side only, one primary at a time) */
static struct {
uint64_t lsn;
uint64_t total_size;
char tmp_path[4096];
int fd;
uint32_t received;
} g_snap_recv;
int repl_receive_snapshot_chunk(const char *snap_dir, uint64_t lsn,
uint32_t chunk_idx, int is_last,
uint64_t total_size,
const void *data, uint32_t data_len,
uint32_t total_crc)
{
/* Validate total_size BEFORE any malloc or open() */
if (chunk_idx == 0 && total_size > MAX_SNAPSHOT_BYTES) {
ERROR("repl_receive_snapshot_chunk: total_size=%lu > MAX_SNAPSHOT_BYTES=%u",
(unsigned long)total_size, MAX_SNAPSHOT_BYTES);
return -EMSGSIZE;
}
if (chunk_idx == 0) {
/* Open temp file */
snprintf(g_snap_recv.tmp_path, sizeof(g_snap_recv.tmp_path),
"%s/snap-%lu.tmp", snap_dir, (unsigned long)lsn);
g_snap_recv.fd = open(g_snap_recv.tmp_path,
O_WRONLY|O_CREAT|O_TRUNC, 0600);
if (g_snap_recv.fd < 0) return -errno;
g_snap_recv.lsn = lsn;
g_snap_recv.total_size = total_size;
g_snap_recv.received = 0;
}
if (data && data_len > 0) {
ssize_t w = write(g_snap_recv.fd, data, data_len);
if (w != (ssize_t)data_len) {
close(g_snap_recv.fd);
unlink(g_snap_recv.tmp_path);
return -EIO;
}
g_snap_recv.received += data_len;
}
if (is_last) {
fsync(g_snap_recv.fd);
close(g_snap_recv.fd);
g_snap_recv.fd = -1;
/* Verify CRC (compute over written file) */
uint32_t actual_crc = compute_file_crc32(g_snap_recv.tmp_path);
if (actual_crc != total_crc) {
ERROR("repl_receive_snapshot_chunk: CRC mismatch lsn=%lu "
"expected=%08x got=%08x", (unsigned long)lsn, total_crc, actual_crc);
unlink(g_snap_recv.tmp_path);
return -EIO;
}
/* Atomic rename tmp→final */
char final_path[4096];
snprintf(final_path, sizeof(final_path),
"%s/snap-%lu.capnp", snap_dir, (unsigned long)lsn);
if (rename(g_snap_recv.tmp_path, final_path) != 0) {
unlink(g_snap_recv.tmp_path);
return -errno;
}
}
return 0;
}
```
Add to `cluster.c` `handle_msg()` **pre-vtask dispatch section** (alongside `OPCode_replAck`):
```c
if (request.opcode == OPCode_replCatchUpDone) {
repl_on_catch_up_done(&c->repl, ep);
free_Request(&request);
return 0;
}
```
### Commit
```bash
git add src/controller/repl.h src/controller/repl.c \
src/controller/cluster.c \
src/controller/testsuite/test_repl_catchup.c
git commit -m "feat: P2-5 standby catch-up — WAL stream, snapshot atomic write, OP_replCatchUpDone"
```
---
## Task P2-6: Tests — round-trip, sync ACK timeout, catch-up integration
### Files
| Action | File |
|--------|------|
| Create | `src/controller/testsuite/test_repl_integration.c` |
| Modify | `src/controller/testsuite/test.c` (register new test suites) |
| Modify | `src/controller/meson.build` (add new test files) |
### Test Coverage Targets
| Scenario | Test |
|----------|------|
| Broadcast reaches all standbys | `test_broadcast_fan_out_all_standbys` |
| Sync round-trip: WAL write → broadcast → ACK → unblock | `test_sync_repl_round_trip` |
| Sync timeout: no ACK → -ETIMEDOUT, client still gets response | `test_sync_repl_timeout_degraded_mode` |
| WAL_CHECKPOINT never broadcast | `test_checkpoint_not_replicated` |
| Role check: standby does not broadcast | `test_standby_does_not_broadcast` |
| Catch-up: standby joins with stale LSN → streams WAL | `test_catchup_wal_stream` |
| Catch-up: snapshot sent when WAL trimmed | `test_catchup_snapshot_then_wal` |
| Catch-up done: in_catch_up cleared on primary | `test_catchup_done_unlocks_wait_ack` |
| Trim LSN: in-catch-up standby uses join_applied_lsn | `test_trim_lsn_catchup_standby` |
| Disconnect trim delay: slot expiry at snap_interval_s | `test_trim_lsn_disconnect_expiry` |
| Decommission: replicator_remove_candidate zeroes slot | `test_decommission_clears_slot` |
| Worker join is async only: no sync block | `test_worker_join_uses_async` |
### Failing Test (RED) — Key Tests
**File**: `src/controller/testsuite/test_repl_integration.c`
```c
/* test_repl_integration.c — P2-6: end-to-end replication scenarios */
#include <pthread.h>
#include <unity/unity.h>
#include "repl.h"
#include "persist.h"
#include "ha.h"
/* ── Helper: multi-threaded ACK injection ─────────────────────── */
typedef struct { replicator_t *r; endpoint_t *ep; uint64_t lsn; } ack_args_t;
static void *inject_ack(void *arg)
{
ack_args_t *a = arg;
struct timespec ts = {0, 20000000}; nanosleep(&ts, NULL);
repl_on_ack(a->r, a->ep, a->lsn);
return NULL;
}
/* ── P2-6-A: Fan-out reaches all standbys ─────────────────────── */
static void test_broadcast_fan_out_all_standbys(void)
{
replicator_t r;
repl_init(&r);
/* 3 standbys, each with a call counter in a mock endpoint */
endpoint_t ep[3];
int recv_count[3] = {0};
for (int i = 0; i < 3; i++) {
ep[i].user_data = &recv_count[i];
repl_add_standby(&r, &ep[i], "sb", 7000 + i);
}
wal_entry_hdr_t hdr = { .type = WAL_JOB_START, .lsn = 5, .term = 1 };
int sent = repl_broadcast(&r, &hdr, NULL, 0);
TEST_ASSERT_EQUAL_INT(3, sent);
for (int i = 0; i < 3; i++) {
TEST_ASSERT_EQUAL_INT(1, recv_count[i]);
}
repl_fini(&r);
printf("test_broadcast_fan_out_all_standbys: PASS\n");
}
/* ── P2-6-B: Sync round-trip ─────────────────────────────────── */
static void test_sync_repl_round_trip(void)
{
replicator_t r;
endpoint_t ep = {0};
repl_init(&r);
repl_add_standby(&r, &ep, "sb", 7001);
ack_args_t args = { &r, &ep, 100 };
pthread_t tid;
pthread_create(&tid, NULL, inject_ack, &args);
wal_entry_hdr_t hdr = { .type = WAL_JOB_SUBMIT, .lsn = 100, .term = 1 };
repl_broadcast(&r, &hdr, NULL, 0);
int rc = repl_wait_ack(&r, 100, 500);
TEST_ASSERT_EQUAL_INT(0, rc);
pthread_join(tid, NULL);
repl_fini(&r);
printf("test_sync_repl_round_trip: PASS\n");
}
/* ── P2-6-C: Sync timeout → degraded mode, not hang ─────────── */
static void test_sync_repl_timeout_degraded_mode(void)
{
replicator_t r;
endpoint_t ep = {0};
repl_init(&r);
repl_add_standby(&r, &ep, "sb", 7001);
/* No ACK delivered — timeout at 50ms */
int rc = repl_wait_ack(&r, 42, 50);
TEST_ASSERT_EQUAL_INT(-ETIMEDOUT, rc);
repl_fini(&r);
printf("test_sync_repl_timeout_degraded_mode: PASS\n");
}
/* ── P2-6-D: WAL_CHECKPOINT never broadcast (assert guard) ────── */
static void test_checkpoint_not_replicated(void)
{
/* Verified by assert(type != WAL_CHECKPOINT) in both persist paths.
* Code-inspection test; abort() would fire if called with CHECKPOINT. */
printf("test_checkpoint_not_replicated: PASS (assert guard verified)\n");
}
/* ── P2-6-E: Catch-up done unblocks repl_wait_ack ─────────────── */
static void test_catchup_done_unlocks_wait_ack(void)
{
replicator_t r;
endpoint_t ep = {0};
repl_init(&r);
repl_add_standby(&r, &ep, "sb", 7001);
/* Mark standby as catching up — excluded from wait_ack predicate */
atomic_store(&r.standbys[0].in_catch_up, 1);
/* Simulate: catch-up done clears flag, then ACK arrives */
repl_on_catch_up_done(&r, &ep);
repl_on_ack(&r, &ep, 77);
int rc = repl_wait_ack(&r, 77, 500);
TEST_ASSERT_EQUAL_INT(0, rc);
repl_fini(&r);
printf("test_catchup_done_unlocks_wait_ack: PASS\n");
}
/* ── P2-6-F: Worker join path uses async, no sync block ──────── */
static void test_worker_join_uses_async(void)
{
/* Design invariant: HOOK_DETAIL_WORKER_JOIN → persist_wal_async()
* Verified by code inspection: handle_worker_join() holds workermgr WRITE lock,
* so it must never call persist_wal_sync_append(). */
printf("test_worker_join_uses_async: PASS (design-invariant test)\n");
}
int main(void)
{
UNITY_BEGIN();
RUN_TEST(test_broadcast_fan_out_all_standbys);
RUN_TEST(test_sync_repl_round_trip);
RUN_TEST(test_sync_repl_timeout_degraded_mode);
RUN_TEST(test_checkpoint_not_replicated);
RUN_TEST(test_catchup_done_unlocks_wait_ack);
RUN_TEST(test_worker_join_uses_async);
return UNITY_END();
}
```
### Commit
```bash
git add src/controller/testsuite/test_repl_integration.c \
src/controller/testsuite/test.c \
src/controller/meson.build
git commit -m "test: P2-6 replication integration tests — round-trip, timeout, catch-up, WAL trim"
```
---
## P2.5 Note (Separate Track)
P2.5 (libxce smart discovery) is a parallel workstream. The **client library code** (P2.5-1 through P2.5-6) depends only on P1 and can be developed concurrently with P2. However, **P2.5-7** (`RegisterToBackend()` in `src/frontend/apiserver/backend/clib.go`) should only ship alongside P2 — without P2 replication, a promoted standby has incomplete state and reconnecting the API server to it is misleading rather than helpful.
Do **not** include P2.5 work in P2 commits.
---
## Summary: Sequence and Green-Gate Checks
```
P2-1 repl.h + repl.c structures, trim_lsn → test_repl_unit.c
P2-2 apply_wal_entry, OP_replEntry dispatch → test_repl_apply.c
P2-3 repl_wait_ack + ACK deadlock prevention → test_repl_sync.c
P2-4 Wire broadcast into persist WAL paths → test_repl_wire.c
P2-5 Standby catch-up + snapshot atomic write → test_repl_catchup.c
P2-6 Integration: round-trip, timeout, catch-up → test_repl_integration.c
```
Each step's tests must be green (`make test`) before proceeding.
---
## Invariants Checklist (Cross-Cutting)
- [ ] `assert(type != WAL_CHECKPOINT)` at top of `persist_wal_async()` and `persist_wal_sync_append()` in debug builds
- [ ] `OP_replAck` handled in io_worker read callback — NOT dispatched via vtaskmgr
- [ ] `OP_replCatchUpDone` handled in io_worker read callback — NOT dispatched via vtaskmgr
- [ ] `repl_broadcast()` checks `am_i_primary(c)` before sending
- [ ] `repl_wait_ack()` excludes standbys with `in_catch_up == 1`
- [ ] Snapshot: `total_size <= MAX_SNAPSHOT_BYTES` checked BEFORE `malloc`
- [ ] Snapshot: temp file `snap-<lsn>.tmp` → `rename(2)` → `snap-<lsn>.capnp`
- [ ] `snap/` directory mode 0700 verified by `persist_init()` (both primary and standby)
- [ ] `HOOK_DETAIL_WORKER_JOIN` → `persist_wal_async()` only (never sync)
- [ ] Disconnect slot expiry: reset `last_disconnected_lsn=0` when `now >= last_disconnect_time + snap_interval_s`
- [ ] Decommissioned standby: `last_disconnected_lsn[i]` zeroed immediately in `replicator_remove_candidate()`
- [ ] `persist_start()` logs `WARN` if `sync_timeout_ms > 0` and `vcpus < 16`
- [ ] `ha_state_t.last_received_crc` reset to `WAL_CRC_NONE` in `ha_do_promote()` (COH-5)
- [ ] `io_worker` pool sized to at least `MAX_STANDBYS + 2` — log at `INFO` in `persist_start()`
Reference in a new issue View git blame Copy permalink