diff --git a/src/controller/ha.c b/src/controller/ha.c index c092511c..1b17687b 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -1030,6 +1030,27 @@ static void ha_advance_applied_lsn(ha_state_t *ha, uint64_t new_lsn) * Returns 0 on success, non-zero on error (caller logs and skips). */ static int ha_apply_single_entry(persist_t *p, wal_catchup_entry_t *e) { + /* Persist-before-apply-before-ACK: durably write the replicated entry to + * this standby's own WAL BEFORE applying it in memory (and before the + * caller ACKs). The ACK feeds the primary's commit majority, so it must + * imply durability — a standby must not ACK an entry it could lose on + * restart (Raft log[i] includes standbys; ha-async-io.md). + * + * persist_wal_write_replicated does NOT assign a new lsn (uses e->lsn) and + * does NOT re-broadcast; it fsyncs before returning and is idempotent on + * re-stream. WAL_CHECKPOINT is never streamed, but it would be rejected + * by both this primitive and apply_wal_entry. */ + int prc = persist_wal_write_replicated(p, e->lsn, (uint32_t)e->entry_type, + e->raw_payload, + (uint32_t)e->raw_payload_len); + if (prc != 0) { + ERROR("ha_apply_single_entry: persist_wal_write_replicated failed for " + "lsn=%" PRIu64 " type=%u rc=%d — not applying/acking", + e->lsn, (unsigned)e->entry_type, prc); + /* No apply, no ACK on persist failure (mirrors no-ACK-on-apply-fail). */ + return prc; + } + wal_entry_hdr_t hdr; memset(&hdr, 0, sizeof(hdr)); hdr.lsn = e->lsn; diff --git a/src/controller/persist.c b/src/controller/persist.c index 0e5d657d..6f660fc1 100644 --- a/src/controller/persist.c +++ b/src/controller/persist.c @@ -397,6 +397,110 @@ int persist_wal_async(persist_t* p, uint32_t type, return 0; } +int persist_wal_write_replicated(persist_t* p, uint64_t lsn, uint32_t type, + const void* payload, uint32_t len) +{ + if (!p) + return -EINVAL; + + /* WAL_CHECKPOINT is a local-only marker, never streamed to standbys. */ + if (type == WAL_CHECKPOINT) + return -EINVAL; + + /* Reject oversized payloads BEFORE touching the segment. */ + if (len > MAX_WAL_PAYLOAD_BYTES) + return -EMSGSIZE; + + pthread_mutex_lock(&p->lock); + + /* Idempotency / dedup guard: if this lsn is already persisted (current_lsn + * already advanced past it) treat the re-stream as a no-op. This keeps the + * write idempotent and avoids a duplicate entry / CRC-chain corruption when + * the primary re-sends an lsn the standby already wrote. */ + if (lsn + 1 <= p->current_lsn) { + pthread_mutex_unlock(&p->lock); + return 0; + } + + /* Open first segment lazily. open_new_segment names the file from + * current_lsn — for a fresh standby receiving lsn 1 first, current_lsn is + * 1 (reserve-lsn-0), so the segment is seg-...001.wal. */ + if (p->wal_fd < 0) { + if (open_new_segment(p) != 0) { + pthread_mutex_unlock(&p->lock); + return -EIO; + } + } + + /* Build entry header. Unlike persist_wal_async, lsn is GIVEN (not + * current_lsn++). term isn't streamed, so use our best-effort p->term. + * The CRC chain is SELF-computed (the streamed entry carries no crc): + * recovery validates an intra-segment chain (prev_crc == last_crc), so the + * standby's own writes must chain to each other — cross-node CRC equality + * is neither required nor possible. */ + wal_entry_hdr_t hdr; + memset(&hdr, 0, sizeof(hdr)); + hdr.term = p->term; + hdr.lsn = lsn; + hdr.timestamp = now_us(); + hdr.type = type; + hdr.payload_len = len; + hdr.prev_crc = (p->seg_offset == 0) ? WAL_CRC_NONE : p->last_crc; + + uint32_t c = crc32(0L, (const Bytef*)&hdr, WAL_HDR_CRC_SIZE); + if (payload && len > 0) + c = crc32(c, (const Bytef*)payload, len); + hdr.crc = c; + + /* Write header */ + ssize_t hw = pwrite(p->wal_fd, &hdr, sizeof(hdr), (off_t)p->seg_offset); + if (hw != (ssize_t)sizeof(hdr)) { + pthread_mutex_unlock(&p->lock); + return -EIO; + } + + /* Write payload (if any) */ + if (payload && len > 0) { + ssize_t pw = pwrite(p->wal_fd, payload, len, + (off_t)(p->seg_offset + sizeof(hdr))); + if (pw != (ssize_t)len) { + pthread_mutex_unlock(&p->lock); + return -EIO; + } + } + + uint32_t entry_size = (uint32_t)(sizeof(hdr) + len); + p->seg_offset += entry_size; + p->total_wal_bytes += entry_size; + p->last_crc = hdr.crc; + + /* Advance current_lsn to one past the given lsn (max-idempotent — mirrors + * apply_wal_entry's monotonic advance). */ + if (lsn + 1 > p->current_lsn) + p->current_lsn = lsn + 1; + + /* Durability: fsync the segment BEFORE returning so the caller's ACK + * implies the entry is on stable storage (persist-before-ACK). + * persist_wal_async defers fsync to the background thread / persist_flush; + * here we sync inline because the entry feeds the primary's commit + * majority and must survive a standby restart. fdatasync is sufficient + * (entry bytes + segment metadata are pre-allocated by open_new_segment). */ + fdatasync(p->wal_fd); + + /* Rotate segment when full (same threshold as the primary append). */ + if (p->seg_offset >= WAL_SEGMENT_SIZE_BYTES) { + if (rotate_segment(p) != 0) { + pthread_mutex_unlock(&p->lock); + return -EIO; + } + } + + /* NOTE: no repl_broadcast — standbys never re-replicate. */ + + pthread_mutex_unlock(&p->lock); + return 0; +} + int persist_flush(persist_t* p) { pthread_mutex_lock(&p->lock); diff --git a/src/controller/testsuite/test_repl_apply.c b/src/controller/testsuite/test_repl_apply.c index 30c2f823..3ea945e2 100644 --- a/src/controller/testsuite/test_repl_apply.c +++ b/src/controller/testsuite/test_repl_apply.c @@ -8,9 +8,11 @@ * - Oversized payloads are rejected (rc != 0) * - Unknown types (> WAL_ENTRY_TYPE_MAX) are skipped with rc=0 */ +#include #include #include #include +#include #include @@ -83,6 +85,85 @@ static persist_t* make_test_persist_for_apply(void) return p; } +/* Like make_test_persist_for_apply but reports the chosen tmpdir so a test + * can re-open the same WAL via a FRESH persist_t (restart/recover proof). */ +static persist_t* make_test_persist_in(char* tmpdir_out, size_t cap) +{ + persist_t* p = calloc(1, sizeof(persist_t)); + if (!p) { fprintf(stderr, "calloc failed\n"); exit(1); } + + persist_config_t cfg = persist_config_defaults(); + cfg.wal_sync_ms = 5000; + + snprintf(tmpdir_out, cap, "/tmp/xce_replwr_XXXXXX"); + if (!mkdtemp(tmpdir_out)) { + fprintf(stderr, "mkdtemp failed\n"); + free(p); + exit(1); + } + + persist_set_workdir(p, tmpdir_out, "ctrl-replwr"); + persist_init(p, &cfg, NULL); + persist_start(p); + return p; +} + +/* Re-open an existing workdir as a fresh persist_t (simulates a restart). */ +static persist_t* reopen_test_persist(const char* tmpdir) +{ + persist_t* p = calloc(1, sizeof(persist_t)); + if (!p) { fprintf(stderr, "calloc failed\n"); exit(1); } + + persist_config_t cfg = persist_config_defaults(); + cfg.wal_sync_ms = 5000; + + persist_set_workdir(p, tmpdir, "ctrl-replwr"); + persist_init(p, &cfg, NULL); + persist_start(p); + return p; +} + +/* Count *.wal files under /ctrl-replwr/data/wal */ +static int count_replwr_wal_files(const char* tmpdir) +{ + char wal_path[512]; + snprintf(wal_path, sizeof(wal_path), "%s/ctrl-replwr/data/wal", tmpdir); + DIR* dir = opendir(wal_path); + if (!dir) return 0; + int count = 0; + struct dirent* ent; + while ((ent = readdir(dir)) != NULL) { + size_t len = strlen(ent->d_name); + if (len > 4 && strcmp(ent->d_name + len - 4, ".wal") == 0) + count++; + } + closedir(dir); + return count; +} + +/* Total bytes across all *.wal segment files (proxy for "entry on disk"). */ +static off_t total_replwr_wal_bytes(const char* tmpdir) +{ + char wal_path[512]; + snprintf(wal_path, sizeof(wal_path), "%s/ctrl-replwr/data/wal", tmpdir); + DIR* dir = opendir(wal_path); + if (!dir) return 0; + off_t total = 0; + struct dirent* ent; + while ((ent = readdir(dir)) != NULL) { + size_t len = strlen(ent->d_name); + if (len <= 4 || strcmp(ent->d_name + len - 4, ".wal") != 0) + continue; + char full[1024]; + snprintf(full, sizeof(full), "%s/%s", wal_path, ent->d_name); + struct stat st; + if (stat(full, &st) == 0) + total += st.st_size; + } + closedir(dir); + return total; +} + static void destroy_test_persist(persist_t* p) { persist_stop(p); @@ -197,6 +278,138 @@ static void test_apply_wal_entry_null_guard(void) printf("test_apply_wal_entry_null_guard: PASS\n"); } +/* ================================================================== */ +/* persist_wal_write_replicated — standby persists replicated entries */ +/* (durability: persist-before-ACK). RED before the fix. */ +/* ================================================================== */ + +/* A streamed entry written to the standby's WAL must survive a restart: + * a FRESH persist_recover over the same WAL replays it (crc-valid) and + * advances current_lsn. Before the fix the segment is empty / nothing + * is recovered. */ +static void test_repl_write_persists_and_recovers(void) +{ + char tmpdir[256]; + persist_t* p = make_test_persist_in(tmpdir, sizeof(tmpdir)); + + /* Fresh standby: first replicated entry is lsn 1 (1-based; reserve-lsn-0). */ + const char payload[] = "replicated_job_update_payload"; + int rc = persist_wal_write_replicated(p, /*lsn=*/1, WAL_JOB_UPDATE, + payload, (uint32_t)strlen(payload)); + TEST_ASSERT_EQUAL_INT(0, rc); + + /* current_lsn advanced to lsn+1 */ + TEST_ASSERT_EQUAL_UINT64(2, p->current_lsn); + + /* A .wal segment must now exist on disk. */ + TEST_ASSERT_TRUE(count_replwr_wal_files(tmpdir) >= 1); + TEST_ASSERT_TRUE(total_replwr_wal_bytes(tmpdir) > 0); + + /* Simulate a restart: drop this persist_t, re-open the same workdir, + * and run recovery. The replicated entry must replay (crc-chain valid) + * and current_lsn must advance past it. */ + destroy_test_persist(p); + + persist_t* p2 = reopen_test_persist(tmpdir); + int rec = persist_recover(p2); + TEST_ASSERT_TRUE(rec >= 0); + + /* Recovery must have advanced current_lsn to 2 (replayed lsn 1). */ + TEST_ASSERT_EQUAL_UINT64(2, p2->current_lsn); + + destroy_test_persist(p2); + printf("test_repl_write_persists_and_recovers: PASS\n"); +} + +/* Multiple replicated entries in sequence form a self-consistent crc chain + * that the standby's OWN recovery validates with no chain break (all replay). */ +static void test_repl_write_crc_chain_recovers(void) +{ + char tmpdir[256]; + persist_t* p = make_test_persist_in(tmpdir, sizeof(tmpdir)); + + const char* payloads[] = { "entry-one", "entry-two", "entry-three" }; + for (int i = 0; i < 3; i++) { + uint64_t lsn = (uint64_t)(i + 1); /* 1,2,3 */ + int rc = persist_wal_write_replicated(p, lsn, WAL_JOB_UPDATE, + payloads[i], + (uint32_t)strlen(payloads[i])); + TEST_ASSERT_EQUAL_INT(0, rc); + } + TEST_ASSERT_EQUAL_UINT64(4, p->current_lsn); + + destroy_test_persist(p); + + /* Fresh recovery must replay ALL three (no crc-chain break) → current_lsn=4. */ + persist_t* p2 = reopen_test_persist(tmpdir); + int rec = persist_recover(p2); + TEST_ASSERT_TRUE(rec >= 0); + TEST_ASSERT_EQUAL_UINT64(4, p2->current_lsn); + + destroy_test_persist(p2); + printf("test_repl_write_crc_chain_recovers: PASS\n"); +} + +/* Re-streaming the same lsn must be idempotent: the dedup guard skips the + * second write (no double-write, no chain corruption). After a dup write, + * recovery still yields exactly the single entry chain. */ +static void test_repl_write_dedup_idempotent(void) +{ + char tmpdir[256]; + persist_t* p = make_test_persist_in(tmpdir, sizeof(tmpdir)); + + const char payload[] = "dup-entry"; + int rc1 = persist_wal_write_replicated(p, 1, WAL_JOB_UPDATE, + payload, (uint32_t)strlen(payload)); + TEST_ASSERT_EQUAL_INT(0, rc1); + uint64_t lsn_after_first = p->current_lsn; /* expect 2 */ + off_t bytes_after_first = total_replwr_wal_bytes(tmpdir); + + /* Re-stream lsn 1 (already persisted): must be a no-op success. */ + int rc2 = persist_wal_write_replicated(p, 1, WAL_JOB_UPDATE, + payload, (uint32_t)strlen(payload)); + TEST_ASSERT_EQUAL_INT(0, rc2); + + /* current_lsn must NOT advance further; no extra bytes written. */ + TEST_ASSERT_EQUAL_UINT64(lsn_after_first, p->current_lsn); + TEST_ASSERT_EQUAL_INT64((long long)bytes_after_first, + (long long)total_replwr_wal_bytes(tmpdir)); + + /* Now legitimately stream lsn 2 — should append fine after the dup. */ + int rc3 = persist_wal_write_replicated(p, 2, WAL_JOB_UPDATE, + payload, (uint32_t)strlen(payload)); + TEST_ASSERT_EQUAL_INT(0, rc3); + TEST_ASSERT_EQUAL_UINT64(3, p->current_lsn); + + destroy_test_persist(p); + + /* Recovery must replay lsn 1 and lsn 2 (no duplicate / no break). */ + persist_t* p2 = reopen_test_persist(tmpdir); + int rec = persist_recover(p2); + TEST_ASSERT_TRUE(rec >= 0); + TEST_ASSERT_EQUAL_UINT64(3, p2->current_lsn); + + destroy_test_persist(p2); + printf("test_repl_write_dedup_idempotent: PASS\n"); +} + +/* WAL_CHECKPOINT must never be persisted via the replicated-write path + * (standbys never receive it; it must be rejected, not written). */ +static void test_repl_write_rejects_checkpoint(void) +{ + char tmpdir[256]; + persist_t* p = make_test_persist_in(tmpdir, sizeof(tmpdir)); + + int rc = persist_wal_write_replicated(p, 1, WAL_CHECKPOINT, NULL, 0); + TEST_ASSERT_NOT_EQUAL(0, rc); + + /* current_lsn must not advance for a rejected checkpoint. */ + TEST_ASSERT_EQUAL_UINT64(1, p->current_lsn); + + destroy_test_persist(p); + printf("test_repl_write_rejects_checkpoint: PASS\n"); +} + /* ------------------------------------------------------------------ */ /* Suite entry point (registered in controller_test.c) */ /* ------------------------------------------------------------------ */ @@ -208,4 +421,10 @@ void test_repl_apply(void) RUN_TEST(test_apply_wal_entry_rejects_oversized); RUN_TEST(test_apply_wal_entry_skips_unknown_type); RUN_TEST(test_apply_wal_entry_null_guard); + + /* persist-before-ACK durability (standby persists replicated entries) */ + RUN_TEST(test_repl_write_persists_and_recovers); + RUN_TEST(test_repl_write_crc_chain_recovers); + RUN_TEST(test_repl_write_dedup_idempotent); + RUN_TEST(test_repl_write_rejects_checkpoint); } diff --git a/src/include/persist.h b/src/include/persist.h index 252c1ad0..f6d7d50a 100644 --- a/src/include/persist.h +++ b/src/include/persist.h @@ -116,6 +116,21 @@ int persist_wal_sync_append(persist_t* p, uint32_t type, const void* payload, uint32_t len, int* sync_rc_out); +/* Write a REPLICATED entry (received from the primary) into the local WAL + * at its GIVEN lsn — used by standbys to durably persist streamed entries + * BEFORE acking them (persist-before-ACK; an ACK must imply durability). + * + * Unlike persist_wal_async it does NOT assign a new lsn (the lsn is supplied + * by the primary) and does NOT repl_broadcast (standbys never re-replicate). + * It builds a SELF-consistent CRC chain (the streamed entry carries no crc), + * fsyncs the segment before returning, and is idempotent on re-stream: if the + * lsn is already persisted (lsn+1 <= current_lsn) it is a no-op returning 0. + * + * WAL_CHECKPOINT is rejected with -EINVAL (never streamed to standbys). + * Returns 0 on success (incl. dedup no-op), negative errno on failure. */ +int persist_wal_write_replicated(persist_t* p, uint64_t lsn, uint32_t type, + const void* payload, uint32_t len); + /* ============================================================ * Write-path commit classification (ha-commit-barrier-rollout) * ============================================================ */