From 3563d7387b079cf16e7f819a428f47ab3771e02f Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Thu, 7 May 2026 17:54:48 +0000 Subject: [PATCH] feat: P2-4 wire repl_broadcast into persist_wal_async/sync_append; last_replicated_crc MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Move repl.h from controller/ to include/ (shared header) to break persist.h ↔ repl.h circular include - Extract wal_entry_hdr_t + wal_entry_type into include/wal_types.h so repl.h can include only what it needs without pulling in all of persist.h - Add replicator_t repl field to persist_t (embedded by value, P2-4) - Wire repl_init/repl_fini into persist_init/persist_fini lifecycle - Wire repl_broadcast into persist_wal_async after each WAL write; last_replicated_crc updated here (not in persist_write_checkpoint_marker) - Wire repl_wait_ack into persist_wal_sync_append; returns 0 on local WAL success; sync_rc_out carries -ENODEV/-ETIMEDOUT for degraded mode - Add 4 TDD tests in test_persist_repl.c: structural repl field check, async no-standbys succeeds, sync degraded mode rc, checkpoint CRC invariant --- src/controller/meson.build | 1 + src/controller/persist.c | 34 ++++- src/controller/testsuite/controller_test.c | 2 + src/controller/testsuite/test_persist_repl.c | 145 +++++++++++++++++++ src/include/persist.h | 39 +---- src/{controller => include}/repl.h | 7 +- src/include/wal_types.h | 41 ++++++ 7 files changed, 228 insertions(+), 41 deletions(-) create mode 100644 src/controller/testsuite/test_persist_repl.c rename src/{controller => include}/repl.h (89%) create mode 100644 src/include/wal_types.h diff --git a/src/controller/meson.build b/src/controller/meson.build index ff16d9a..e27333d 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -89,6 +89,7 @@ controller_unit_test_src = [ 'testsuite/test_p1_integration.c', 'testsuite/test_repl_unit.c', 'testsuite/test_repl_apply.c', + 'testsuite/test_persist_repl.c', 'persist.c', 'ha.c', 'repl.c', diff --git a/src/controller/persist.c b/src/controller/persist.c index 52ebf2c..c97de0e 100644 --- a/src/controller/persist.c +++ b/src/controller/persist.c @@ -168,6 +168,7 @@ int persist_init(persist_t* p, persist_config_t* cfg, cluster_t* c) pthread_mutex_init(&p->lock, NULL); pthread_mutex_init(&p->checkpoint_mutex, NULL); pthread_cond_init(&p->checkpoint_done, NULL); + repl_init(&p->repl); /* P2-4: fan-out replicator lifecycle */ fprintf(stderr, "INFO: WAL/snapshot stored unencrypted in %s — " @@ -214,6 +215,7 @@ void persist_stop(persist_t* p) void persist_fini(persist_t* p) { + repl_fini(&p->repl); /* P2-4: fan-out replicator lifecycle */ pthread_mutex_destroy(&p->lock); pthread_mutex_destroy(&p->checkpoint_mutex); pthread_cond_destroy(&p->checkpoint_done); @@ -274,8 +276,15 @@ int persist_wal_async(persist_t* p, uint32_t type, p->seg_offset += entry_size; p->total_wal_bytes += entry_size; p->last_crc = hdr.crc; - /* For single-controller P0 there is no replication; track same value */ - p->last_replicated_crc = hdr.crc; + + /* Replicate to standbys — never replicate WAL_CHECKPOINT entries. + * last_replicated_crc is distinct from last_crc: + * last_crc includes checkpoint markers; last_replicated_crc skips them. + * This maintains a continuous CRC chain for standbys (design doc Part 2.3). + * Note: WAL_CHECKPOINT is blocked by the assert at function entry, so + * every entry that reaches here is a replicable entry. */ + p->last_replicated_crc = hdr.crc; + repl_broadcast(&p->repl, &hdr, payload, len); /* Rotate segment when full */ if (p->seg_offset >= WAL_SEGMENT_SIZE_BYTES) { @@ -845,9 +854,26 @@ int persist_wal_sync_append(persist_t* p, uint32_t type, int* sync_rc_out) { int rc = persist_wal_async(p, type, payload, len); + if (rc != 0) { + if (sync_rc_out) + *sync_rc_out = rc; + return rc; + } + + /* For sync replication (WAL_JOB_SUBMIT): wait for at least one standby ACK. + * We use current_lsn - 1 because persist_wal_async already incremented it. */ + uint64_t written_lsn; + pthread_mutex_lock(&p->lock); + written_lsn = p->current_lsn - 1; + pthread_mutex_unlock(&p->lock); + + int sync_err = repl_wait_ack(&p->repl, written_lsn, + p->config.sync_timeout_ms); if (sync_rc_out) - *sync_rc_out = rc; - return rc; + *sync_rc_out = sync_err; + /* -ETIMEDOUT = degraded mode (local-only durability); -ENODEV = no standbys. + * Both are acceptable — primary continues operating. */ + return 0; } void persist_log_entry_metadata(const wal_entry_hdr_t* hdr) diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index 318e46e..8b22bc4 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -25,6 +25,7 @@ extern void test_ha_redirect(void); extern void test_p1_integration(void); extern void test_repl_unit(void); extern void test_repl_apply(void); +extern void test_persist_repl(void); void setUp(void) { return; } @@ -119,6 +120,7 @@ int main(int argc, char* argv[]) TEST(test_p1_integration); TEST(test_repl_unit); TEST(test_repl_apply); + TEST(test_persist_repl); return UNITY_END(); } diff --git a/src/controller/testsuite/test_persist_repl.c b/src/controller/testsuite/test_persist_repl.c new file mode 100644 index 0000000..d872ffa --- /dev/null +++ b/src/controller/testsuite/test_persist_repl.c @@ -0,0 +1,145 @@ +/* + * test_persist_repl.c — P2-4: Tests for repl_broadcast wired into WAL write paths. + * + * TDD RED phase: these tests must FAIL before implementation. + * After implementation they must all PASS. + * + * Tests verify: + * 1. persist_t has repl field (structural/compile-time check) + * 2. persist_wal_async succeeds with zero standbys + * 3. persist_wal_sync_append returns 0 (WAL success) but sets sync_rc < 0 (degraded) + * 4. last_replicated_crc NOT updated by WAL_CHECKPOINT (checkpoint marker) + */ +#include +#include +#include +#include +#include +#include + +#include +#include "persist.h" +#include "repl.h" + +/* ------------------------------------------------------------------ */ +/* Helpers */ +/* ------------------------------------------------------------------ */ + +static persist_t* make_repl_persist(const char* suffix) +{ + persist_t* p = calloc(1, sizeof(persist_t)); + persist_config_t cfg = persist_config_defaults(); + cfg.wal_sync_ms = 5000; + cfg.sync_timeout_ms = 50; /* short timeout so sync tests don't hang */ + + char tmpdir[256]; + snprintf(tmpdir, sizeof(tmpdir), "/tmp/xce_repl_%s_XXXXXX", suffix); + if (!mkdtemp(tmpdir)) { + free(p); + return NULL; + } + + persist_set_workdir(p, tmpdir, "ctrl"); + persist_init(p, &cfg, NULL); + persist_start(p); + return p; +} + +static void teardown_repl_persist(persist_t* p) +{ + if (!p) return; + persist_stop(p); + persist_fini(p); + free(p); +} + +/* ------------------------------------------------------------------ */ +/* Tests */ +/* ------------------------------------------------------------------ */ + +/* + * Structural: persist_t must contain a repl field of type replicator_t. + * This is a compile-time check — accessing p.repl.nstandbys must compile. + */ +static void test_persist_has_repl_field(void) +{ + persist_t p; + memset(&p, 0, sizeof(p)); + /* Accessing repl.nstandbys is the compile-time guard */ + TEST_ASSERT_EQUAL_INT(0, p.repl.nstandbys); + printf("test_persist_has_repl_field: PASS\n"); +} + +/* + * WAL async must succeed when there are zero standbys. + * repl_broadcast is a stub; no replication occurs, but write must not fail. + */ +static void test_wal_async_no_standbys_succeeds(void) +{ + persist_t* p = make_repl_persist("async"); + TEST_ASSERT_NOT_NULL(p); + + int rc = persist_wal_async(p, WAL_JOB_UPDATE, "payload", 7); + TEST_ASSERT_EQUAL_INT(0, rc); + + teardown_repl_persist(p); + printf("test_wal_async_no_standbys_succeeds: PASS\n"); +} + +/* + * persist_wal_sync_append must return 0 (WAL write succeeded locally) even + * when there are no standbys (degraded mode). The sync_rc_out parameter + * must be set to a negative value to signal degraded mode + * (-ENODEV = no standbys, or -ETIMEDOUT = timeout). + */ +static void test_wal_sync_no_standbys_returns_enodev_in_sync_rc(void) +{ + persist_t* p = make_repl_persist("sync"); + TEST_ASSERT_NOT_NULL(p); + + int sync_rc = 0; + int rc = persist_wal_sync_append(p, WAL_JOB_SUBMIT, "job_payload", 11, + &sync_rc); + /* WAL write itself must succeed */ + TEST_ASSERT_EQUAL_INT(0, rc); + /* sync_rc: -ENODEV (no standbys) or -ETIMEDOUT — either indicates degraded */ + TEST_ASSERT_TRUE(sync_rc < 0); + + teardown_repl_persist(p); + printf("test_wal_sync_no_standbys_returns_enodev_in_sync_rc: PASS\n"); +} + +/* + * WAL_CHECKPOINT (via persist_checkpoint) must NOT update last_replicated_crc. + * Checkpoint markers are local-only; the replication CRC chain must remain + * unbroken for standbys (design doc Part 2.3). + */ +static void test_last_replicated_crc_not_updated_for_checkpoint(void) +{ + persist_t* p = make_repl_persist("crc"); + TEST_ASSERT_NOT_NULL(p); + + uint32_t crc_before = p->last_replicated_crc; + + /* persist_checkpoint writes a WAL_CHECKPOINT marker then triggers a snap. + * Neither the marker write nor snap must update last_replicated_crc. */ + persist_checkpoint(p); + + /* last_replicated_crc must be unchanged from before the checkpoint */ + TEST_ASSERT_EQUAL_UINT32(crc_before, p->last_replicated_crc); + + teardown_repl_persist(p); + printf("test_last_replicated_crc_not_updated_for_checkpoint: PASS\n"); +} + +/* ------------------------------------------------------------------ */ +/* Entry point */ +/* ------------------------------------------------------------------ */ + +void test_persist_repl(void) +{ + RUN_TEST(test_persist_has_repl_field); + RUN_TEST(test_wal_async_no_standbys_succeeds); + RUN_TEST(test_wal_sync_no_standbys_returns_enodev_in_sync_rc); + RUN_TEST(test_last_replicated_crc_not_updated_for_checkpoint); +} diff --git a/src/include/persist.h b/src/include/persist.h index 9285b9b..1ba0a8c 100644 --- a/src/include/persist.h +++ b/src/include/persist.h @@ -6,44 +6,12 @@ #include #include "config.h" +#include "wal_types.h" /* wal_entry_hdr_t, wal_entry_type — P2-4: shared with repl.h */ +#include "repl.h" /* replicator_t — P2-4: embedded in persist_t */ /* Forward declaration */ typedef struct cluster_s cluster_t; -/* ============================================================ - * WAL entry header — 40 bytes, 8-byte aligned (Part 2.2) - * ============================================================ */ -typedef struct { - uint64_t term; /* Election term (incremented on failover) */ - uint64_t lsn; /* Log sequence number within this term */ - uint64_t timestamp; /* Epoch microseconds */ - uint32_t type; /* wal_entry_type enum value */ - uint32_t payload_len; /* Bytes of payload following this header */ - uint32_t crc; /* CRC32(header[0..WAL_HDR_CRC_SIZE-1] + payload) */ - uint32_t prev_crc; /* CRC of previous record; WAL_CRC_NONE for first in segment */ -} wal_entry_hdr_t; -_Static_assert(sizeof(wal_entry_hdr_t) == 40, - "wal_entry_hdr_t must be 40 bytes for on-disk format stability"); - -/* ============================================================ - * WAL entry types (Part 2.3) - * ============================================================ */ -typedef enum { - WAL_JOB_SUBMIT = 1, /* SYNC replication */ - WAL_JOB_START = 2, - WAL_JOB_FINISH = 3, - WAL_JOB_FAIL = 4, - WAL_JOB_STOP = 5, - WAL_JOB_RESUME = 6, - WAL_JOB_UPDATE = 7, - WAL_JOB_RUSAGE = 8, - WAL_JOB_CKPT = 9, - WAL_JOB_PREASON = 10, - WAL_QUEUE_CTRL = 20, - WAL_WORKER_CTRL = 30, - WAL_CHECKPOINT = 100, /* Local WAL marker only — never replicated */ -} wal_entry_type; - /* ============================================================ * Persistence configuration (Part 2.6) * ============================================================ */ @@ -120,6 +88,9 @@ struct persist_s { pthread_cond_t checkpoint_done; pthread_mutex_t checkpoint_mutex; pthread_t snap_thread; + + /* Fan-out replicator — P2-4 */ + replicator_t repl; }; typedef struct persist_s persist_t; diff --git a/src/controller/repl.h b/src/include/repl.h similarity index 89% rename from src/controller/repl.h rename to src/include/repl.h index e135a0d..62ac652 100644 --- a/src/controller/repl.h +++ b/src/include/repl.h @@ -10,7 +10,7 @@ #include #include "config.h" #include "mylock.h" -#include "persist.h" /* wal_entry_hdr_t */ +#include "wal_types.h" /* wal_entry_hdr_t — P2-4: use wal_types.h to break circular include */ /* Endpoint is an opaque pointer — full type in netio.h (avoid heavy include) */ typedef void* repl_ep_t; @@ -27,8 +27,9 @@ typedef struct { time_t last_disconnect_time; } standby_conn_t; -/* Replicator — embedded in persist_t or cluster_t (P2-4) */ -typedef struct { +/* Replicator — embedded in persist_t or cluster_t (P2-4). + * Named tag replicator_s matches the forward declaration in persist.h. */ +typedef struct replicator_s { standby_conn_t standbys[MAX_STANDBYS]; int nstandbys; /* Active standby count */ mylock_t lock; /* Protects standbys[], nstandbys */ diff --git a/src/include/wal_types.h b/src/include/wal_types.h new file mode 100644 index 0000000..2597f70 --- /dev/null +++ b/src/include/wal_types.h @@ -0,0 +1,41 @@ +#pragma once +/* + * wal_types.h — WAL on-disk structures shared between persist.h and repl.h. + * Extracted in P2-4 to break the persist.h ↔ repl.h circular include. + */ +#include +#include "config.h" + +/* ============================================================ + * WAL entry header — 40 bytes, 8-byte aligned (Part 2.2) + * ============================================================ */ +typedef struct { + uint64_t term; /* Election term (incremented on failover) */ + uint64_t lsn; /* Log sequence number within this term */ + uint64_t timestamp; /* Epoch microseconds */ + uint32_t type; /* wal_entry_type enum value */ + uint32_t payload_len; /* Bytes of payload following this header */ + uint32_t crc; /* CRC32(header[0..WAL_HDR_CRC_SIZE-1] + payload) */ + uint32_t prev_crc; /* CRC of previous record; WAL_CRC_NONE for first in segment */ +} wal_entry_hdr_t; +_Static_assert(sizeof(wal_entry_hdr_t) == 40, + "wal_entry_hdr_t must be 40 bytes for on-disk format stability"); + +/* ============================================================ + * WAL entry types (Part 2.3) + * ============================================================ */ +typedef enum { + WAL_JOB_SUBMIT = 1, /* SYNC replication */ + WAL_JOB_START = 2, + WAL_JOB_FINISH = 3, + WAL_JOB_FAIL = 4, + WAL_JOB_STOP = 5, + WAL_JOB_RESUME = 6, + WAL_JOB_UPDATE = 7, + WAL_JOB_RUSAGE = 8, + WAL_JOB_CKPT = 9, + WAL_JOB_PREASON = 10, + WAL_QUEUE_CTRL = 20, + WAL_WORKER_CTRL = 30, + WAL_CHECKPOINT = 100, /* Local WAL marker only — never replicated */ +} wal_entry_type;