diff --git a/src/controller/config_repl.c b/src/controller/config_repl.c index 01f6e47..72976c1 100644 --- a/src/controller/config_repl.c +++ b/src/controller/config_repl.c @@ -23,6 +23,7 @@ #include #include +#include /* bug-1389 triage: per-chunk crc32 instrumentation */ #include "byteorder.h" #include "clusterconfig.h" @@ -1273,21 +1274,40 @@ typedef struct { } push_ctx_t; #endif +/* push_thread_send_hook: test-only seam (same pattern as + * xce_wal_stream_request_hook) letting a unit test intercept/park the + * per-standby send so it can probe endpoint_ref/unref bracketing around a + * concurrent teardown (bug-1381) without a real transport connection. NULL + * in production — push_thread falls through to the real + * xce_replconfigpush_req_raw. */ +int (*push_thread_send_hook)(endpoint_t *ep, xce_peerinfo_t *from, + uint64_t new_tag, const uint8_t *sha256, + const uint8_t *bytes, size_t blen) = NULL; + static void *push_thread(void *arg) { push_ctx_t *ctx = (push_ctx_t *)arg; - int rc = xce_replconfigpush_req_raw( - ctx->ep, ctx->from, - ctx->new_tag, - ctx->sha256, 32, - ctx->bytes, ctx->blen); + int rc = (push_thread_send_hook != NULL) + ? push_thread_send_hook(ctx->ep, ctx->from, ctx->new_tag, + ctx->sha256, ctx->bytes, ctx->blen) + : xce_replconfigpush_req_raw( + ctx->ep, ctx->from, + ctx->new_tag, + ctx->sha256, 32, + ctx->bytes, ctx->blen); if (rc == 0) ctx->acked = 1; else WARN("push_config_to_standbys: push to endpoint %p failed: %d", ctx->ep, rc); + /* bug-1381: drop the ref push_config_transport_only took on ctx->ep before + * spawning us (the general bug-1298 refcount system) — this thread is the + * one non-owner reader of ep for the full push, so it owns the matching + * release. Last use of ctx->ep in this thread; safe to unref now. */ + endpoint_unref(ctx->ep); + ctx->done = 1; return NULL; } @@ -1361,6 +1381,12 @@ int wait_all_with_timeout_for_test(pthread_t *threads, push_ctx_t *ctxs, * promotion, heartbeat-driven convergence, controller join). Alloc or * thread failures degrade to fewer pushes, never to a fabricated quorum * (the barrier wait in push_config_to_standbys owns the quorum verdict). + * bug-1381 (CLOSED): each standby ep is endpoint_ref()'d under c->ha.lock + * before the pointer is snapshotted into ctxs[], matched by exactly one + * endpoint_unref() (from push_thread, or from this function's own cleanup + * if the thread never started) — so a concurrent RDHUP teardown can drop + * the OWNER's reference but never frees ep out from under an in-flight + * push_thread. * Stores the registered standby count in *n_out (may be NULL); returns * the number of standbys that ACKed the transport push. */ int push_config_transport_only(cluster_t *c, const uint8_t *bytes, @@ -1397,11 +1423,22 @@ int push_config_transport_only(cluster_t *c, const uint8_t *bytes, ctxs[i].bytes = bytes; ctxs[i].blen = blen; ctxs[i].sha256 = sha256; + /* bug-1381: take our own reference on each standby ep while still + * holding c->ha.lock — the same window the pointer is snapshotted — + * mirroring the snapshot-lane fix (bug-1197/bug-1376). Without this, + * a concurrent RDHUP teardown (handle_endpoint_error -> zarray_del -> + * endpoint_killer -> endpoint_unref) can free ep while push_thread + * still reads/writes it for up to CONF_PUSH_TIMEOUT_MS. Every ref + * taken here has exactly one matching endpoint_unref: from push_thread + * on success, or from the cleanup below if the thread never started. */ + endpoint_ref(ctxs[i].ep); } pthread_mutex_unlock(&c->ha.lock); pthread_t *threads = malloc((size_t)n * sizeof(pthread_t)); if (!threads) { + for (int i = 0; i < n; i++) + endpoint_unref(ctxs[i].ep); free(ctxs); WARN("push_config_transport_only: thread alloc failed (n=%d)" " — skipping transport push", n); @@ -1410,8 +1447,12 @@ int push_config_transport_only(cluster_t *c, const uint8_t *bytes, for (int i = 0; i < n; i++) { ctxs[i].started = (pthread_create(&threads[i], NULL, push_thread, &ctxs[i]) == 0); - if (!ctxs[i].started) + if (!ctxs[i].started) { WARN("push_config_transport_only: pthread_create failed for standby %d", i); + /* No thread will run push_thread's matching endpoint_unref — + * release our ref right here (bug-1381). */ + endpoint_unref(ctxs[i].ep); + } } atomic_int push_done; @@ -2602,6 +2643,18 @@ int ha_send_snapshot_to_standby(endpoint_t *ep, const char *snap_dir, break; } + /* bug-1389 triage: crc32 of the exact bytes this chunk is about to + * send, taken right after the local file read and before encode — + * correlate against ha_handle_snapshot_chunk's receive-side crc32 + * log (ha.c) by (snapshot_lsn, chunk_index) to localize whether a + * mismatch is introduced before send, in the wire transport, or + * during receive-side assembly. */ + TRACE("ha_send_snapshot_to_standby: chunk crc32 " + "checkpoint_lsn=%" PRIu64 " chunk_index=%u offset=%" PRIu64 + " len=%zu crc32=%08x", + checkpoint_lsn, chunk_index, offset, chunk_len, + (unsigned)crc32(0L, (const Bytef *)buf, (uInt)chunk_len)); + int is_last = (offset + chunk_len >= total_size); snapshot_transfer_t chunk; @@ -2620,6 +2673,13 @@ int ha_send_snapshot_to_standby(endpoint_t *ep, const char *snap_dir, chunk.raw_payload = buf; /* borrowed for hook/encode only */ chunk.raw_payload_len = chunk_len; + /* bug-1386: a failed chunk send (e.g. -ECONNRESET while the standby + * endpoint is mid-teardown/reconnect) MUST stop the transfer and be + * propagated. Previously this return value was discarded, so the + * loop kept "sending" past a dead endpoint and the function still + * reported success — the standby silently received a truncated + * snap.part that never gets an is_last chunk. */ + int send_rc = 0; #ifdef XCE_TESTING if (xce_snapshot_chunk_hook) { xce_request_t req; @@ -2627,11 +2687,16 @@ int ha_send_snapshot_to_standby(endpoint_t *ep, const char *snap_dir, req.opcode = OPCode_snapshotTransfer; req.body_which = Request_body_snapshottransfermsg; req.body.snapshottransfermsg = &chunk; - xce_snapshot_chunk_hook(ep, &req); + send_rc = xce_snapshot_chunk_hook(ep, &req); } else #endif if (ep) - xce_snapshot_chunk_req_raw(ep, &chunk, buf, chunk_len); + send_rc = xce_snapshot_chunk_req_raw(ep, &chunk, buf, chunk_len); + + if (send_rc != 0) { + rc = send_rc; + break; + } offset += chunk_len; chunk_index++; @@ -2645,6 +2710,11 @@ int ha_send_snapshot_to_standby(endpoint_t *ep, const char *snap_dir, INFO("ha_send_snapshot_to_standby: snapshot transfer complete " "checkpoint_lsn=%" PRIu64 " size=%" PRIu64, checkpoint_lsn, total_size); + else + WARN("ha_send_snapshot_to_standby: snapshot transfer failed " + "checkpoint_lsn=%" PRIu64 " size=%" PRIu64 " sent=%" PRIu64 + " chunk_index=%u rc=%d", + checkpoint_lsn, total_size, offset, chunk_index, rc); return rc; } diff --git a/src/controller/ha.c b/src/controller/ha.c index 1cd698c..24105dd 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -2588,33 +2588,78 @@ int ha_handle_snapshot_chunk(ha_state_t *ha, persist_t *p, snprintf(part_path, sizeof(part_path), "%s/snap.part", snap_dir); /* Req 4.4: a transfer always restarts from chunk 0 — discard any stale - * partial file from an interrupted previous transfer. */ - if (chunk->chunk_index == 0 && unlink(part_path) != 0 && errno != ENOENT) - WARN("ha_handle_snapshot_chunk: unlink stale snap.part failed: %s", - strerror(errno)); + * partial file from an interrupted previous transfer, and reset the + * completion bookkeeping below for the fresh attempt. */ + pthread_mutex_lock(&ha->lock); + if (chunk->chunk_index == 0) { + if (unlink(part_path) != 0 && errno != ENOENT) + WARN("ha_handle_snapshot_chunk: unlink stale snap.part failed: %s", + strerror(errno)); + ha->snapshot_recv_count = 0; + ha->snapshot_recv_total = 0; + } + pthread_mutex_unlock(&ha->lock); - /* Append chunk payload to snap.part */ + /* bug-1389 triage: crc32 of the exact bytes this chunk arrived with, + * taken right after decode and before any local write — correlate + * against ha_send_snapshot_to_standby's send-side crc32 log + * (config_repl.c) by (snapshot_lsn, chunk_index) to localize whether a + * mismatch is introduced before send, in the wire transport, or during + * receive-side assembly. */ + if (payload_bytes && payload_len > 0) + TRACE("ha_handle_snapshot_chunk: chunk crc32 " + "checkpoint_lsn=%" PRIu64 " chunk_index=%u len=%zu crc32=%08x", + chunk->snapshot_lsn, chunk->chunk_index, payload_len, + (unsigned)crc32(0L, (const Bytef *)payload_bytes, (uInt)payload_len)); + + /* bug-1389: chunks of one transfer are dispatched across the vtask pool + * and are NOT guaranteed to arrive/execute in chunk_index order (a live + * soak observed e.g. chunk_index 2 processed AFTER 3 and 4) — a live + * crc32 correlation confirmed every chunk's CONTENT survives the wire + * intact, but the OLD append-in-arrival-order write put reordered + * chunks' bytes at the wrong file position. pwrite() at the chunk's own + * chunk_offset makes assembly order-independent and safe for concurrent + * writers (positioned writes to disjoint byte ranges of the same fd + * don't race each other). */ if (payload_bytes && payload_len > 0) { - FILE *f = fopen(part_path, "ab"); - if (!f) { + int fd = open(part_path, O_CREAT | O_WRONLY, 0600); + if (fd < 0) { int err = errno; ERROR("ha_handle_snapshot_chunk: cannot open snap.part: %s", strerror(err)); return -err; } - size_t written = fwrite(payload_bytes, 1, payload_len, f); - fclose(f); - if (written != payload_len) { + ssize_t written = pwrite(fd, payload_bytes, payload_len, + (off_t)chunk->chunk_offset); + close(fd); + if (written != (ssize_t)payload_len) { ERROR("ha_handle_snapshot_chunk: short write to snap.part " - "(%zu of %zu bytes)", written, payload_len); + "(%zd of %zu bytes) at offset %" PRIu64, + written, payload_len, chunk->chunk_offset); return -EIO; } } - if (!chunk->is_last) + /* bug-1389: finalize once every chunk index 0..N-1 has actually been + * written, not merely once THIS call happens to carry is_last — under + * out-of-order arrival, is_last can be processed before an earlier + * chunk. snapshot_recv_total is claimed (reset to 0) by whichever + * thread wins the race so a late straggler chunk can never trigger a + * second finalize. */ + pthread_mutex_lock(&ha->lock); + ha->snapshot_recv_count++; + if (chunk->is_last) + ha->snapshot_recv_total = chunk->chunk_index + 1; + int should_finalize = (ha->snapshot_recv_total != 0 && + ha->snapshot_recv_count >= ha->snapshot_recv_total); + if (should_finalize) + ha->snapshot_recv_total = 0; + pthread_mutex_unlock(&ha->lock); + + if (!should_finalize) return 0; - /* isLast=true: verify size, rename, load, set applied_lsn */ + /* All chunks accounted for: verify size, rename, load, set applied_lsn */ /* Verify total file size equals chunk->total_size */ struct stat st; @@ -4333,33 +4378,68 @@ int ha_snapshot_offload_enabled(void) * DEFAULT OFF (0): the requester-less config push runs INLINE on the caller * (io_worker/drive) thread — byte-for-byte the legacy behavior, i.e. * push_config_transport_only's fan-out followed by its blocking - * wait_all_with_timeout(CONF_PUSH_TIMEOUT_MS). This task lands DORMANT, - * exactly like the Task 3.1/3.2 flags, because the offload borrows standby - * endpoints it does not own (bug-1197 family — push_config_transport_only - * reads c->ha.standby_endpoints): moving the push to a lane thread while the - * io_worker may RDHUP-free those endpoints is a cross-thread UAF. Today (flag - * OFF) the push and the endpoint teardown are serialized on the single - * io_worker, so they cannot overlap. + * wait_all_with_timeout(CONF_PUSH_TIMEOUT_MS). * - * STILL DEFAULT OFF, unlike the snapshot (3.2) and peer-control (3.1) flags, - * which flipped ON once their single-endpoint hazard closed via endpoint_ref/ - * endpoint_unref. push_config_transport_only's hazard is DIFFERENT and NOT - * yet closed by that fix: it grabs raw endpoint_t* pointers for ALL standbys - * under c->ha.lock, releases the lock, then spawns one pthread per standby - * that reads/writes those endpoints unprotected for up to CONF_PUSH_TIMEOUT_MS - * — a race that exists TODAY regardless of this flag (the pthread-per-standby - * fan-out is not the io_worker, so "runs inline on io_worker" does not - * actually serialize it against a concurrent RDHUP teardown). Logged as - * bug-1381 (open). Needs its own fix (ref every standby ep for the fan-out's - * duration, or route through the per-ep block primitive) before flipping ON - * — tracked separately, not fixed by this task. Until then this offload is - * exercised only by unit tests with a stable (or absent) endpoint set. + * Landed DORMANT alongside the Task 3.1/3.2 flags because the offload borrows + * standby endpoints it does not own (bug-1197 family — push_config_transport_only + * reads c->ha.standby_endpoints): moving the push to a lane thread while the + * io_worker may RDHUP-free those endpoints would be a cross-thread UAF. + * push_config_transport_only's hazard was DISTINCT from the snapshot (3.2) + * lane's borrowed-ep hazard — it grabbed raw endpoint_t* pointers for ALL + * standbys under c->ha.lock, released the lock, then spawned one pthread per + * standby that read/wrote those endpoints unprotected for up to + * CONF_PUSH_TIMEOUT_MS, a race that existed regardless of this flag's value + * (the pthread-per-standby fan-out is not the io_worker, so "runs inline on + * io_worker" never actually serialized it against a concurrent RDHUP + * teardown). Logged as bug-1381, **fixed 2026-07-02**: push_config_transport_only + * (config_repl.c) now takes its own endpoint_ref() on each standby ep at the + * same point the pointer is snapshotted under c->ha.lock, with a matching + * endpoint_unref() from push_thread (or from this function's own cleanup if a + * thread never started) — mirroring this same file's snapshot-lane fix. + * + * **Flip attempted 2026-07-02, REVERTED to OFF**: bug-1381 alone did not clear + * the bar. Flag-ON force-snapshot soak went 0/3 pass ("no snapshot install" / + * "did not reconverge to 3-of-3" on the freshly-rebuilt xce.build binary — NOT + * a stale-binary artifact); a flag-OFF control on the identical build went + * 1/2 pass with the SAME failure signature, so the underlying defect is + * PRE-EXISTING and orthogonal to this flag (root cause: `ha_send_snapshot_to_standby`, + * config_repl.c, never checks `xce_snapshot_chunk_req_raw`'s return in its + * chunk loop, so a chunk that fails to enqueue — e.g. ECONNRESET during the + * empty-rejoin reconnect window — is silently dropped and the sender still + * logs "transfer complete" and returns 0; logged as a new bug, see buglog). + * Even so, flag-ON's 0/3 vs flag-OFF's 1/2 is a small enough sample that a + * TIMING interaction (configpush lane traffic narrowing the window the + * snapshot lane has to complete before reconnect) can't be ruled out either. + * Flag stayed OFF until the chunk-loss bug (bug-1386) AND a second, + * independently-discovered bug (bug-1389) were fixed and re-validated. + * + * **bug-1386 FIXED 2026-07-02** (config_repl.c): the send-side chunk loop now + * checks xce_snapshot_chunk_req_raw's return and stops+propagates on failure + * instead of silently continuing and reporting "complete". + * + * **bug-1389 FIXED 2026-07-02** (this file + ha.h): re-validating bug-1386's + * fix surfaced a SECOND, independent defect — force-snapshot chunks are + * dispatched across the vtask pool with no ordering guarantee (a live soak + * showed arrival order 0,1,3,4,2,5,... across cpu#0-3 workers), but + * ha_handle_snapshot_chunk appended every chunk's bytes to snap.part in + * ARRIVAL order instead of at its own chunk_offset — reordered chunks landed + * at the wrong file position (same total size, wrong content). A live crc32 + * correlation between send and receive proved every chunk's CONTENT survived + * the wire intact, ruling out capnp/transport entirely. Fixed via + * pwrite()-at-chunk_offset (order-independent) plus a + * snapshot_recv_count/snapshot_recv_total completion tracker (guarded by + * ha->lock) so finalize fires only once every chunk has actually arrived, + * not merely because the current call happens to carry is_last. + * + * **Flag re-flipped to ON 2026-07-02** after both bugs closed: full controller + * suite green, live-validated 3/3 --force-snapshot PASS (genuine snapshot + * installs confirmed) + 6-round --chaotic PASS on the flag-ON binary. * * ONLY the requester-less call sites (post-election promotion, heartbeat-driven * HA_REPL_PUSH_CONFIG) route through this flag; the admin-upgrade barrier path * (push_config_to_standbys) is unaffected. Compiled in ALL builds; the setter * is the test seam to flip RED<->GREEN. */ -static _Atomic int g_drive_async_configpush_offload = 0; +static _Atomic int g_drive_async_configpush_offload = 1; void ha_set_configpush_offload_for_test(int on) { diff --git a/src/controller/ha.h b/src/controller/ha.h index 00c1d2e..8c3e5c7 100644 --- a/src/controller/ha.h +++ b/src/controller/ha.h @@ -319,6 +319,43 @@ typedef struct { * a stale value can never be mistaken * for a live in-flight LSN. 0 at * init (ha_init memset). */ + /* ── New fields added by bug-1389 fix ──────────────────────────────── */ + uint32_t snapshot_recv_count; /* Standby receive side: count of + * OPCode_snapshotTransfer chunks + * structurally accounted for (written + * to snap.part) in the CURRENT + * transfer attempt. The vtask pool can + * dispatch chunks of one transfer to + * different workers, so chunks are NOT + * guaranteed to arrive/process in + * chunk_index order (bug-1389: chunks + * observed out of order in a live + * soak) — ha_handle_snapshot_chunk + * writes each chunk at its own + * chunk_offset (pwrite, order- + * independent) and uses this counter, + * not "this call happened to carry + * is_last", to decide when every chunk + * has actually arrived. Reset to 0 + * when chunk_index==0 starts a fresh + * transfer (Req 4.4). Guarded by + * ha->lock. 0 at init (ha_init + * memset). */ + uint32_t snapshot_recv_total; /* Standby receive side: total chunk + * count for the current transfer, + * known only once the is_last chunk + * has been seen (= its chunk_index+1, + * chunks are 0-based and contiguous). + * 0 = not yet known (is_last not seen + * yet) OR no transfer in progress — + * ha_handle_snapshot_chunk finalizes + * (verify/rename/load) only once + * snapshot_recv_count reaches this + * value, then resets it to 0 so a + * late straggler chunk cannot trigger + * a second finalize. Guarded by + * ha->lock. 0 at init (ha_init + * memset). */ } ha_state_t; /* ── Public API ────────────────────────────────────────────── */ diff --git a/src/controller/meson.build b/src/controller/meson.build index c50e197..38454c4 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -54,7 +54,7 @@ controller = executable('xce_controller', link_args: wrap_link_args, install: true) -xce_test_src = [ 'testsuite/test.c', 'testsuite/waldump.c', 'testsuite/walcompare.c'] +xce_test_src = [ 'testsuite/test.c', 'testsuite/waldump.c', 'testsuite/walcompare.c', 'testsuite/snapverify.c', 'testsuite/snapdiff.c'] readline_dep = dependency('readline') ncurses_dep = dependency('ncurses') @@ -151,6 +151,10 @@ controller_unit_test_src = [ 'testsuite/waldump.c', 'testsuite/test_walcompare.c', 'testsuite/walcompare.c', + 'testsuite/test_snapverify.c', + 'testsuite/snapverify.c', + 'testsuite/test_snapdiff.c', + 'testsuite/snapdiff.c', 'testsuite/test_persist_snapshot_entities.c', 'ctrlstatus.c', 'commit.c', diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index 94c8bb6..7fae0be 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -66,6 +66,8 @@ extern void test_ha_vote_floor_heartbeat(void); extern void test_term_safety_adapter(void); extern void test_waldump(void); extern void test_walcompare(void); +extern void test_snapverify(void); +extern void test_snapdiff(void); extern void test_drive_async_config(void); extern void test_slowlane(void); extern void test_commit_admission_pool(void); @@ -209,6 +211,8 @@ int main(int argc, char* argv[]) TEST(test_term_safety_adapter); TEST(test_waldump); TEST(test_walcompare); + TEST(test_snapverify); + TEST(test_snapdiff); TEST(test_drive_async_config); TEST(test_slowlane); TEST(test_commit_admission_pool); diff --git a/src/controller/testsuite/snapdiff.c b/src/controller/testsuite/snapdiff.c new file mode 100644 index 0000000..2a199fc --- /dev/null +++ b/src/controller/testsuite/snapdiff.c @@ -0,0 +1,113 @@ +#define _GNU_SOURCE +#include "snapdiff.h" + +#include +#include +#include +#include +#include +#include + +#define SNAPDIFF_BUF_SIZE 65536 + +int snapdiff_paths(const char* path1, const char* path2, FILE* out, + snapdiff_report_t* rep) +{ + if (!path1 || !path2) + return -EINVAL; + + snapdiff_report_t local; + memset(&local, 0, sizeof(local)); + local.diff_offset = -1; + + int fd1 = open(path1, O_RDONLY); + if (fd1 < 0) { + int saved = errno; + if (out) + fprintf(out, "snapdiff: %s: %s\n", path1, strerror(saved)); + return -saved; + } + int fd2 = open(path2, O_RDONLY); + if (fd2 < 0) { + int saved = errno; + close(fd1); + if (out) + fprintf(out, "snapdiff: %s: %s\n", path2, strerror(saved)); + return -saved; + } + + struct stat st1, st2; + if (fstat(fd1, &st1) != 0 || fstat(fd2, &st2) != 0) { + int saved = errno; + close(fd1); + close(fd2); + if (out) + fprintf(out, "snapdiff: fstat failed: %s\n", strerror(saved)); + return -saved; + } + local.size1 = st1.st_size; + local.size2 = st2.st_size; + + if (st1.st_size != st2.st_size) { + local.size_mismatch = 1; + close(fd1); + close(fd2); + if (rep) + *rep = local; + if (out) + fprintf(out, + "snapdiff: %s (%jd bytes) vs %s (%jd bytes): SIZE MISMATCH\n", + path1, (intmax_t)local.size1, path2, (intmax_t)local.size2); + return 1; + } + + uint8_t buf1[SNAPDIFF_BUF_SIZE]; + uint8_t buf2[SNAPDIFF_BUF_SIZE]; + off_t off = 0; + int found_diff = 0; + + for (;;) { + ssize_t n1 = read(fd1, buf1, sizeof(buf1)); + ssize_t n2 = read(fd2, buf2, sizeof(buf2)); + if (n1 <= 0 || n2 <= 0) + break; + ssize_t n = (n1 < n2) ? n1 : n2; + for (ssize_t i = 0; i < n; i++) { + if (buf1[i] != buf2[i]) { + local.diff_offset = off + (off_t)i; + found_diff = 1; + break; + } + } + if (found_diff) + break; + off += (off_t)n; + } + close(fd1); + close(fd2); + + if (!found_diff) { + local.identical = 1; + if (rep) + *rep = local; + if (out) + fprintf(out, "snapdiff: %s vs %s: IDENTICAL (%jd bytes)\n", path1, + path2, (intmax_t)local.size1); + return 0; + } + + local.chunk_index = local.diff_offset / SNAPDIFF_CHUNK_SIZE; + local.at_chunk_boundary = (local.diff_offset % SNAPDIFF_CHUNK_SIZE == 0); + + if (rep) + *rep = local; + if (out) + fprintf(out, + "snapdiff: %s vs %s: DIFFER at offset %jd (chunk %jd%s)\n", + path1, path2, (intmax_t)local.diff_offset, + (intmax_t)local.chunk_index, + local.at_chunk_boundary + ? " -- exactly at a chunk boundary, first byte of chunk" + : " -- NOT at a chunk boundary, mid-chunk"); + return 1; +} diff --git a/src/controller/testsuite/snapdiff.h b/src/controller/testsuite/snapdiff.h new file mode 100644 index 0000000..ed88f2e --- /dev/null +++ b/src/controller/testsuite/snapdiff.h @@ -0,0 +1,43 @@ +#pragma once +/* + * snapdiff.h — byte-level comparator for two checkpoint files + * (bug-1389 triage: OPCode_snapshotTransfer chunk-content corruption). + * + * Compares a locally-generated source checkpoint (snap-*.capnp) against a + * received one (snap-*.bin, ha_handle_snapshot_chunk) byte-for-byte, and + * reports the first divergent offset plus whether it lands exactly on an + * OPCode_snapshotTransfer chunk boundary (SNAPSHOT_CHUNK_SIZE, config_repl.c) + * — the decisive signal separating "chunk transport corrupted this specific + * chunk" from "the two files were never going to match" (different content + * entirely, unrelated to chunking). + * + * Strictly read-only: opens both files O_RDONLY, mutates nothing. + */ +#include +#include + +/* Mirrors config_repl.c's SNAPSHOT_CHUNK_SIZE (the OPCode_snapshotTransfer + * wire chunk size) — kept in sync manually; both are compile-time constants + * for the same wire protocol, not expected to change independently. */ +#define SNAPDIFF_CHUNK_SIZE 65536 + +typedef struct { + off_t size1; + off_t size2; + int size_mismatch; /* 1 if size1 != size2 (diff_offset unset in that case) */ + int identical; /* 1 if same size and byte-for-byte identical */ + off_t diff_offset; /* -1 if identical or size_mismatch; else first + * differing byte offset */ + int at_chunk_boundary; /* 1 if diff_offset is the first byte of a + * SNAPDIFF_CHUNK_SIZE-aligned chunk, i.e. + * diff_offset % SNAPDIFF_CHUNK_SIZE == 0 */ + off_t chunk_index; /* diff_offset / SNAPDIFF_CHUNK_SIZE, valid only + * when diff_offset >= 0 */ +} snapdiff_report_t; + +/* Compare `path1` (source of truth) against `path2` (candidate, e.g. a + * received copy). Prints a human-readable report to `out` (may be NULL) and + * fills `*rep` (may be NULL). Returns 0 if identical, 1 if they differ (see + * `*rep` for detail), or a negative errno on open/fstat failure. */ +int snapdiff_paths(const char* path1, const char* path2, FILE* out, + snapdiff_report_t* rep); diff --git a/src/controller/testsuite/snapverify.c b/src/controller/testsuite/snapverify.c new file mode 100644 index 0000000..cd2bc2e --- /dev/null +++ b/src/controller/testsuite/snapverify.c @@ -0,0 +1,284 @@ +#define _GNU_SOURCE +#include "snapverify.h" + +#include +#include +#include +#include +#include +#include + +/* Mirrors persist.c's SNAPSHOT_FORMAT_VERSION / header layout exactly — see + * persist.c write_snapshot() (~line 1758) and load_snapshot() (~line 2325). */ +#define SNAP_FORMAT_VERSION 1u +#define SNAP_HEADER_MAX (16 + 1 + 8 + 32) /* magic+lsn + optional cfgref tail */ + +static void fail(snapverify_report_t* rep, int rc, off_t off, const char* why) +{ + rep->ok = 0; + rep->fail_offset = off; + snprintf(rep->fail_reason, sizeof(rep->fail_reason), "%s", why); + (void)rc; +} + +int snapverify_path(const char* path, FILE* out, snapverify_report_t* rep) +{ + if (!path) + return -EINVAL; + + snapverify_report_t local; + memset(&local, 0, sizeof(local)); + local.ok = 1; + local.fail_offset = -1; + + int fd = open(path, O_RDONLY); + if (fd < 0) { + int saved = errno; + fail(&local, -saved, -1, strerror(saved)); + if (rep) + *rep = local; + if (out) + fprintf(out, "snapverify: %s: %s\n", path, strerror(saved)); + return -saved; + } + + struct stat st; + if (fstat(fd, &st) != 0) { + int saved = errno; + close(fd); + fail(&local, -saved, -1, strerror(saved)); + if (rep) + *rep = local; + if (out) + fprintf(out, "snapverify: %s: fstat: %s\n", path, strerror(saved)); + return -saved; + } + local.file_size = st.st_size; + + char header[SNAP_HEADER_MAX]; + ssize_t n = read(fd, header, sizeof(header)); + if (n < 16) { + close(fd); + fail(&local, -EIO, 0, "short read on header (need >= 16 bytes)"); + if (rep) + *rep = local; + if (out) + fprintf(out, "snapverify: %s: %s\n", path, local.fail_reason); + return -EIO; + } + + if (memcmp(header, "XCESNAP\0", 8) != 0) { + close(fd); + fail(&local, -EINVAL, 0, "bad magic (not XCESNAP)"); + if (rep) + *rep = local; + if (out) + fprintf(out, "snapverify: %s: %s\n", path, local.fail_reason); + return -EINVAL; + } + + memcpy(&local.checkpoint_lsn, header + 8, 8); + if (n >= SNAP_HEADER_MAX && header[16] != 0) + local.has_cfgref = 1; + + off_t consumed = (off_t)n; + + if (st.st_size <= consumed) { + /* Legacy, position-only checkpoint — not a defect (design.md). */ + close(fd); + local.consumed = consumed; + if (rep) + *rep = local; + if (out) + fprintf(out, + "snapverify: %s: OK (legacy position-only, lsn=%" PRIu64 + ", size=%jd)\n", + path, local.checkpoint_lsn, (intmax_t)local.file_size); + return 0; + } + + uint32_t fmt = 0; + if (read(fd, &fmt, sizeof(fmt)) != (ssize_t)sizeof(fmt)) { + close(fd); + fail(&local, -EIO, consumed, "short read on format_version"); + if (rep) + *rep = local; + if (out) + fprintf(out, "snapverify: %s: %s (offset %jd)\n", path, + local.fail_reason, (intmax_t)consumed); + return -EIO; + } + consumed += (off_t)sizeof(fmt); + local.has_entity_section = 1; + local.format_version = fmt; + + if (fmt != SNAP_FORMAT_VERSION) { + close(fd); + fail(&local, -EINVAL, consumed - (off_t)sizeof(fmt), + "unrecognized format_version"); + if (rep) + *rep = local; + if (out) + fprintf(out, "snapverify: %s: %s (got %u, expected %u, offset %jd)\n", + path, local.fail_reason, fmt, SNAP_FORMAT_VERSION, + (intmax_t)local.fail_offset); + return -EINVAL; + } + + uint32_t job_count = 0; + if (read(fd, &job_count, sizeof(job_count)) != (ssize_t)sizeof(job_count)) { + close(fd); + fail(&local, -EIO, consumed, "short read on job_count"); + if (rep) + *rep = local; + if (out) + fprintf(out, "snapverify: %s: %s (offset %jd)\n", path, + local.fail_reason, (intmax_t)consumed); + return -EIO; + } + consumed += (off_t)sizeof(job_count); + local.job_count = job_count; + + for (uint32_t i = 0; i < job_count; i++) { + off_t len_off = consumed; + uint32_t plen = 0; + if (read(fd, &plen, sizeof(plen)) != (ssize_t)sizeof(plen)) { + close(fd); + fail(&local, -EIO, len_off, "short read on job payload_len"); + if (rep) + *rep = local; + if (out) + fprintf(out, "snapverify: %s: job %u: %s (offset %jd)\n", path, + i, local.fail_reason, (intmax_t)len_off); + return -EIO; + } + consumed += (off_t)sizeof(plen); + + if (plen == 0 || (off_t)plen > st.st_size - consumed) { + close(fd); + char why[160]; + snprintf(why, sizeof(why), + "bad payload_len=%u (remaining=%jd)", plen, + (intmax_t)(st.st_size - consumed)); + fail(&local, -EIO, len_off, why); + if (rep) + *rep = local; + if (out) + fprintf(out, "snapverify: %s: job %u: %s (length field at " + "offset %jd)\n", + path, i, local.fail_reason, (intmax_t)len_off); + return -EIO; + } + + if (lseek(fd, plen, SEEK_CUR) < 0) { + close(fd); + fail(&local, -EIO, consumed, "seek past job payload failed"); + if (rep) + *rep = local; + if (out) + fprintf(out, "snapverify: %s: job %u: %s\n", path, i, + local.fail_reason); + return -EIO; + } + consumed += (off_t)plen; + local.jobs_ok++; + } + + uint32_t queue_count = 0; + if (read(fd, &queue_count, sizeof(queue_count)) != + (ssize_t)sizeof(queue_count)) { + close(fd); + fail(&local, -EIO, consumed, "short read on queue_count"); + if (rep) + *rep = local; + if (out) + fprintf(out, "snapverify: %s: %s (offset %jd)\n", path, + local.fail_reason, (intmax_t)consumed); + return -EIO; + } + consumed += (off_t)sizeof(queue_count); + local.queue_count = queue_count; + + for (uint32_t i = 0; i < queue_count; i++) { + off_t len_off = consumed; + uint32_t name_len = 0; + if (read(fd, &name_len, sizeof(name_len)) != (ssize_t)sizeof(name_len)) { + close(fd); + fail(&local, -EIO, len_off, "short read on queue name_len"); + if (rep) + *rep = local; + if (out) + fprintf(out, "snapverify: %s: queue %u: %s (offset %jd)\n", + path, i, local.fail_reason, (intmax_t)len_off); + return -EIO; + } + consumed += (off_t)sizeof(name_len); + + if (name_len == 0 || (off_t)name_len > st.st_size - consumed) { + close(fd); + char why[160]; + snprintf(why, sizeof(why), + "bad name_len=%u (remaining=%jd)", name_len, + (intmax_t)(st.st_size - consumed)); + fail(&local, -EIO, len_off, why); + if (rep) + *rep = local; + if (out) + fprintf(out, "snapverify: %s: queue %u: %s (length field at " + "offset %jd)\n", + path, i, local.fail_reason, (intmax_t)len_off); + return -EIO; + } + + if (lseek(fd, name_len, SEEK_CUR) < 0) { + close(fd); + fail(&local, -EIO, consumed, "seek past queue name failed"); + if (rep) + *rep = local; + if (out) + fprintf(out, "snapverify: %s: queue %u: %s\n", path, i, + local.fail_reason); + return -EIO; + } + consumed += (off_t)name_len; + + uint32_t status = 0; + if (read(fd, &status, sizeof(status)) != (ssize_t)sizeof(status)) { + close(fd); + fail(&local, -EIO, consumed, "short read on queue status"); + if (rep) + *rep = local; + if (out) + fprintf(out, "snapverify: %s: queue %u: %s (offset %jd)\n", + path, i, local.fail_reason, (intmax_t)consumed); + return -EIO; + } + consumed += (off_t)sizeof(status); + local.queues_ok++; + } + + close(fd); + local.consumed = consumed; + if (consumed != st.st_size) { + char why[160]; + snprintf(why, sizeof(why), + "trailing garbage: consumed=%jd but file_size=%jd", + (intmax_t)consumed, (intmax_t)st.st_size); + fail(&local, -EIO, consumed, why); + if (rep) + *rep = local; + if (out) + fprintf(out, "snapverify: %s: %s\n", path, local.fail_reason); + return -EIO; + } + + if (rep) + *rep = local; + if (out) + fprintf(out, + "snapverify: %s: OK (lsn=%" PRIu64 " fmt=%u jobs=%u queues=%u " + "size=%jd)\n", + path, local.checkpoint_lsn, local.format_version, + local.job_count, local.queue_count, (intmax_t)local.file_size); + return 0; +} diff --git a/src/controller/testsuite/snapverify.h b/src/controller/testsuite/snapverify.h new file mode 100644 index 0000000..3855e34 --- /dev/null +++ b/src/controller/testsuite/snapverify.h @@ -0,0 +1,46 @@ +#pragma once +/* + * snapverify.h — read-only checkpoint-file structural inspector/validator + * (bug-1389 triage: OPCode_snapshotTransfer chunk-content corruption). + * + * Parses the on-disk checkpoint layout persist.c's load_snapshot() / + * load_snapshot_entities() use (design.md "Checkpoint File Layout") WITHOUT + * restoring any job/queue into a live cluster_t — a pure structural walk, so + * it needs no jobmgr/qmgr context and is safe to run standalone against a + * snap-*.capnp (locally generated) or snap-*.bin (received via + * ha_handle_snapshot_chunk, ha.c) file on disk. + * + * Strictly read-only: opens O_RDONLY, mutates nothing. + */ +#include +#include +#include + +typedef struct { + uint64_t checkpoint_lsn; + int has_cfgref; /* config-ref tail present in the header */ + int has_entity_section; /* fmt + job_count present past the header; + * 0 = legacy position-only checkpoint */ + uint32_t format_version; + uint32_t job_count; /* as declared in the file (0 if legacy) */ + uint32_t jobs_ok; /* job frames structurally walked before any failure */ + uint32_t queue_count; /* as declared in the file (0 if legacy, or if the + * job-frame walk failed before reaching it) */ + uint32_t queues_ok; /* queue frames structurally walked before any failure */ + off_t file_size; + off_t consumed; /* bytes structurally accounted for */ + int ok; /* 1 = well-formed, consumed == file_size */ + off_t fail_offset; /* -1 if ok; else the byte offset of the field that + * failed the structural check */ + char fail_reason[160]; +} snapverify_report_t; + +/* Parse+validate `path` (a snap-*.capnp or snap-*.bin checkpoint file). + * Prints a human-readable report to `out` (may be NULL to suppress) and + * fills `*rep` (may be NULL). Returns 0 if the file is structurally + * well-formed to EOF, or the same rc class persist_load_snapshot() would + * return for the equivalent defect (-EINVAL bad magic / unrecognized + * format_version, -EIO short read or a length field that doesn't fit the + * remaining bytes, -errno on open/fstat failure) — so a "persist_load_snapshot + * failed rc=..." log line correlates directly against this tool's verdict. */ +int snapverify_path(const char* path, FILE* out, snapverify_report_t* rep); diff --git a/src/controller/testsuite/test.c b/src/controller/testsuite/test.c index 7a1715e..f8ae710 100644 --- a/src/controller/testsuite/test.c +++ b/src/controller/testsuite/test.c @@ -25,6 +25,8 @@ #include "preason.h" #include "walcompare.h" /* cross-member committed-prefix divergence (cmd_walcompare) */ #include "waldump.h" /* offline WAL inspector (cmd_waldump) */ +#include "snapdiff.h" /* byte-level checkpoint-file comparator (cmd_snapdiff) */ +#include "snapverify.h" /* offline checkpoint-file structural validator (cmd_snapverify) */ #include "xce/cluster.h" /* xce_parse_peers, xce_cluster_config_t */ #include "xce/types.h" @@ -127,6 +129,8 @@ extern int cmd_wait_primary(int, char*[]); extern int cmd_ctrlstatus(int, char*[]); extern int cmd_waldump(int, char*[]); extern int cmd_walcompare(int, char*[]); +extern int cmd_snapverify(int, char*[]); +extern int cmd_snapdiff(int, char*[]); static int worker_house_keeping(void*); /* Session verdict / loop state (xce-test-e2e-primitives task 1.1). @@ -186,6 +190,8 @@ static command_t top_commands[] = { { "echo", cmd_echo, "Print a message" }, { "waldump", cmd_waldump, "Inspect WAL segments" }, { "walcompare", cmd_walcompare, "Diff member committed WAL prefixes" }, + { "snapverify", cmd_snapverify, "Verify checkpoint file structure" }, + { "snapdiff", cmd_snapdiff, "Byte-diff two checkpoint files" }, { "quit", cmd_quit, "Quit xce test" }, { (char*)NULL, (cmd_func_t)NULL, (char*)NULL } }; @@ -225,6 +231,8 @@ static command_t client_commands[] = { { "ctrlstatus", cmd_ctrlstatus, "show controller/cluster status" }, { "waldump", cmd_waldump, "Inspect WAL segments" }, { "walcompare", cmd_walcompare, "Diff member committed WAL prefixes" }, + { "snapverify", cmd_snapverify, "Verify checkpoint file structure" }, + { "snapdiff", cmd_snapdiff, "Byte-diff two checkpoint files" }, { "shutdown", cmd_shutdown, "shutdown xce controller" }, { "jobrun", cmd_jobrun, "run a job" }, { "jobsub", cmd_jobsub, "submit a job" }, @@ -798,6 +806,52 @@ int cmd_walcompare(int argc, char* argv[]) return -1; } +/* snapverify [ ...] — offline checkpoint-file + * structural validator for force-snapshot triage (bug-1389: OPCode_ + * snapshotTransfer chunk-content corruption). Each path is a snap-*.capnp + * (locally generated) or snap-*.bin (received via ha_handle_snapshot_chunk) + * file. Read-only: walks the same layout persist_load_snapshot() parses + * without restoring anything, and on the first structural defect reports the + * exact byte offset and how many job/queue frames parsed cleanly before it — + * the diagnostic a bare "persist_load_snapshot failed rc=-5" log line + * doesn't give you. Returns nonzero if any path failed to verify. */ +int cmd_snapverify(int argc, char* argv[]) +{ + if (argc < 2) { + fprintf(stderr, "usage: snapverify [more...]\n"); + return -1; + } + + int rc = 0; + for (int i = 1; i < argc; i++) { + snapverify_report_t rep; + int r = snapverify_path(argv[i], stdout, &rep); + if (r != 0) + rc = -1; + } + return rc; +} + +/* snapdiff — byte-level checkpoint-file + * comparator for force-snapshot triage (bug-1389). `source_file` is the + * primary's own locally generated checkpoint (snap-*.capnp); `received_file` + * is the copy a standby assembled from the wire (snap-*.bin). Reports the + * first differing byte offset and whether it lands exactly on an + * OPCode_snapshotTransfer chunk boundary — the decisive signal for "the + * transport corrupted one chunk" vs "these were never the same content". + * Returns 0 if identical, nonzero otherwise. */ +int cmd_snapdiff(int argc, char* argv[]) +{ + if (argc != 3) { + fprintf(stderr, "usage: snapdiff \n"); + return -1; + } + + snapdiff_report_t rep; + int rc = snapdiff_paths(argv[1], argv[2], stdout, &rep); + return (rc == 0) ? 0 : -1; +} + int cmd_up(int argc, char* argv[]) { if (active_ctx->parent != NULL) { diff --git a/src/controller/testsuite/test_config_repl.c b/src/controller/testsuite/test_config_repl.c index 63fd159..eb13cd2 100644 --- a/src/controller/testsuite/test_config_repl.c +++ b/src/controller/testsuite/test_config_repl.c @@ -1652,6 +1652,135 @@ static void test_wait_all_with_timeout_fast_threads(void) elapsed_ms, acked); } +/* ── push_config_transport_only endpoint lifetime (bug-1381) ───── */ + +/* push_thread_send_hook: test-only seam declared in config_repl.c (same + * pattern as xce_wal_stream_request_hook) letting this test intercept and + * park push_thread's send so it can probe endpoint_ref/unref bracketing + * around a concurrent teardown, without a real transport connection. */ +extern int (*push_thread_send_hook)(endpoint_t *ep, xce_peerinfo_t *from, + uint64_t new_tag, const uint8_t *sha256, + const uint8_t *bytes, size_t blen); + +static atomic_int g_cpto_parked; +static atomic_int g_cpto_release; + +static int cpto_parking_send_hook(endpoint_t *ep, xce_peerinfo_t *from, + uint64_t new_tag, const uint8_t *sha256, + const uint8_t *bytes, size_t blen) +{ + (void)ep; (void)from; (void)new_tag; (void)sha256; (void)bytes; (void)blen; + atomic_store_explicit(&g_cpto_parked, 1, memory_order_release); + while (!atomic_load_explicit(&g_cpto_release, memory_order_acquire)) + usleep(1000); + return 0; +} + +static int cpto_wait_flag(atomic_int *flag, int target, int timeout_ms) +{ + int waited = 0; + while (atomic_load_explicit(flag, memory_order_acquire) != target) { + if (waited >= timeout_ms) + return -1; + usleep(2000); + waited += 2; + } + return 0; +} + +typedef struct { + cluster_t *c; + int acked; + int n; +} cpto_driver_arg_t; + +static void *cpto_driver_fn(void *arg) +{ + cpto_driver_arg_t *a = (cpto_driver_arg_t *)arg; + uint8_t sha[32]; + uint8_t body[4] = { 1, 2, 3, 4 }; + memset(sha, 0, sizeof(sha)); + a->acked = push_config_transport_only(a->c, body, sizeof(body), sha, 99, + &a->n); + return NULL; +} + +/* + * bug-1381: push_config_transport_only spawns a push_thread per registered + * standby that reads ctx->ep for the full duration of the transport send. + * Before the fix, no ref was taken on ep, so a concurrent RDHUP teardown + * (endpoint_killer -> endpoint_unref, the bug-1298 refcount system) could + * free ep out from under the still-running push_thread -- a UAF. This test + * parks a push_thread mid-send via push_thread_send_hook, simulates the + * owner's teardown, and verifies ep survives until the send completes. + */ +static void test_configpush_transport_holds_endpoint_ref_across_send(void) +{ + cluster_t c; + memset(&c, 0, sizeof(c)); + xce_peerinfo_t myhost_info; + memset(&myhost_info, 0, sizeof(myhost_info)); + c.myhost = &myhost_info; + + ha_config_t cfg; + memset(&cfg, 0, sizeof(cfg)); + cfg.heartbeat_timeout_ms = 10000; + cfg.election_timeout_ms = 1500; + ha_init(&c.ha, &cfg, 0); + + c.ha.standby_endpoints = calloc(1, sizeof(zarray_t)); + TEST_ASSERT_NOT_NULL(c.ha.standby_endpoints); + zarray_init(c.ha.standby_endpoints, 4, NULL); + + /* Heap-allocated endpoint standing in for a real standby connection: + * refcnt starts at 1, representing the owner (the standby zarray) + * holding the creator reference. */ + endpoint_t *ep = endpoint_get(); + TEST_ASSERT_NOT_NULL(ep); + zarray_store(c.ha.standby_endpoints, ep); + + atomic_store(&g_cpto_parked, 0); + atomic_store(&g_cpto_release, 0); + push_thread_send_hook = cpto_parking_send_hook; + + cpto_driver_arg_t driver = { .c = &c, .acked = -1, .n = -1 }; + pthread_t driver_thread; + TEST_ASSERT_EQUAL_INT(0, + pthread_create(&driver_thread, NULL, cpto_driver_fn, &driver)); + + TEST_ASSERT_EQUAL_INT_MESSAGE(0, cpto_wait_flag(&g_cpto_parked, 1, 2000), + "push_thread must park in the send hook before we probe refcnt"); + + TEST_ASSERT_EQUAL_INT_MESSAGE(2, + atomic_load_explicit(&ep->refcnt, memory_order_acquire), + "push_config_transport_only must hold its own ref (owner's 1 + " + "push_thread's 1) while the send is in flight"); + + /* Simulate the owner's teardown (RDHUP -> zarray_del -> endpoint_killer + * -> endpoint_unref, bug-1298) racing the in-flight push. */ + endpoint_unref(ep); + + TEST_ASSERT_EQUAL_INT_MESSAGE(1, + atomic_load_explicit(&ep->refcnt, memory_order_acquire), + "ep must survive the owner's unref while push_thread is still in " + "flight -- bug-1381"); + + atomic_store_explicit(&g_cpto_release, 1, memory_order_release); + pthread_join(driver_thread, NULL); + + TEST_ASSERT_EQUAL_INT(1, driver.n); + TEST_ASSERT_EQUAL_INT(1, driver.acked); + /* ep is now freed (push_thread's completion unref dropped the last + * reference) -- do not touch it again. */ + + push_thread_send_hook = NULL; + zarray_fini(c.ha.standby_endpoints); + free(c.ha.standby_endpoints); + c.ha.standby_endpoints = NULL; + ha_fini(&c.ha); + printf("test_configpush_transport_holds_endpoint_ref_across_send: PASS\n"); +} + #endif /* XCE_TESTING */ /* ── raw capnp shims (Task A) ───────────────────── */ @@ -3034,6 +3163,71 @@ static void test_snapshot_transfer_small_less_than_one_chunk(void) printf("test_snapshot_transfer_small_less_than_one_chunk: PASS\n"); } +/* bug-1386: a chunk send that fails mid-transfer (e.g. -ECONNRESET while the + * standby endpoint is mid-teardown/reconnect) must stop the loop and be + * propagated, not be silently swallowed while the function still reports + * "transfer complete". */ +static int g_snaptest_fail_after_calls = -1; /* -1 = never fail */ + +static int snaptest_fail_after_hook(endpoint_t *ep, xce_request_t *req) +{ + (void)ep; + if (g_snaptest_chunk_count >= 16) return -1; + memcpy(&g_snaptest_chunks[g_snaptest_chunk_count], + req->body.snapshottransfermsg, sizeof(snapshot_transfer_t)); + g_snaptest_chunk_count++; + if (g_snaptest_fail_after_calls >= 0 && + g_snaptest_chunk_count > g_snaptest_fail_after_calls) + return -ECONNRESET; + return 0; +} + +static void test_snapshot_transfer_stops_and_propagates_send_failure(void) +{ + char tmpdir[] = "/tmp/xcectest_snaptrans_XXXXXX"; + TEST_ASSERT_NOT_NULL(mkdtemp(tmpdir)); + + /* 3 chunks: 2 full 64KB chunks + a partial third, so a mid-stream + * failure on chunk 1 leaves chunk 2 (the would-be is_last chunk) + * unsent. */ + char snap_path[512]; + snprintf(snap_path, sizeof(snap_path), "%s/snap-00000000000000c8.capnp", tmpdir); + FILE *f = fopen(snap_path, "wb"); + TEST_ASSERT_NOT_NULL(f); + char header[16] = "XCESNAP\0"; + uint64_t lsn = 200; + memcpy(header + 8, &lsn, 8); + fwrite(header, 1, 16, f); + char zero[4096]; + memset(zero, 0, sizeof(zero)); + size_t total = 65536 + 65536 + 4096; + for (size_t i = 16; i < total; i += 4096) + fwrite(zero, 1, (i + 4096 > total) ? (total - i) : 4096, f); + fclose(f); + + extern int (*xce_snapshot_chunk_hook)(endpoint_t *, xce_request_t *); + xce_snapshot_chunk_hook = snaptest_fail_after_hook; + snaptest_reset(); + g_snaptest_fail_after_calls = 1; /* chunk_index 0 succeeds, chunk_index 1 fails */ + + extern int ha_send_snapshot_to_standby(endpoint_t *, const char *, + uint64_t, uint64_t); + int rc = ha_send_snapshot_to_standby(NULL, tmpdir, lsn, 0); + + TEST_ASSERT_EQUAL_INT(-ECONNRESET, rc); + /* Loop must stop right after the failing chunk — chunk_index 2 (the + * is_last chunk) must never be attempted. */ + TEST_ASSERT_EQUAL_INT(2, g_snaptest_chunk_count); + TEST_ASSERT_EQUAL_INT(0, g_snaptest_chunks[1].is_last); + + xce_snapshot_chunk_hook = NULL; + g_snaptest_fail_after_calls = -1; + char cmd[512]; + snprintf(cmd, sizeof(cmd), "rm -rf %s", tmpdir); + system(cmd); + printf("test_snapshot_transfer_stops_and_propagates_send_failure: PASS\n"); +} + /* ── Task 3.5c: Standby snapshot receive handler ──────────────────────────── */ /* Helper: initialise a minimal ha_state_t for snapshot receive tests */ @@ -3446,6 +3640,113 @@ static void test_snapshot_recv_128kb_two_chunks(void) printf("test_snapshot_recv_128kb_two_chunks: PASS\n"); } +/* bug-1389: a live --force-snapshot soak observed the vtask pool dispatching + * a 12-chunk transfer's chunks out of chunk_index order — chunk 2 was + * processed AFTER chunks 3 and 4 (arrival order 0,1,3,4,2,5,6,7,8,9,10,11). + * Live crc32 instrumentation proved every chunk's CONTENT survived the wire + * intact; the bug was that ha_handle_snapshot_chunk appended bytes in + * ARRIVAL order (fopen "ab") instead of writing each chunk at its own + * chunk_offset, so chunks 2/3/4 landed at the wrong file positions — + * same total size, wrong content, exactly matching the corruption the soak + * saw. This test replays that exact arrival order and asserts the + * assembled file is byte-identical to the correctly-ordered source, + * regardless of the order ha_handle_snapshot_chunk was called in. */ +static void test_snapshot_recv_out_of_order_chunks_assemble_correctly(void) +{ + char tmpdir[] = "/tmp/xcectest_snaprecv3_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)); + + /* 4 chunks of 64 bytes each = 256-byte body. Must be a WELL-FORMED + * entity-bearing checkpoint (same reason as + * test_snapshot_recv_two_chunks_assembles_and_renames above): legacy + * 57-byte header (cfgref_present=0) + format_version=1 + job_count=0 + + * queue_count=0, then filler out to 256 bytes. The filler is + * position-dependent (distinct per source byte, not per chunk) so ANY + * chunk landing at the wrong file offset is immediately detectable via + * a whole-buffer memcmp, regardless of which physical chunk a given + * structural field happens to fall in. */ + uint64_t snap_lsn = 400; + size_t total_size = 256; + uint8_t body[256]; + memset(body, 0, sizeof(body)); + memcpy(body, "XCESNAP\0", 8); + memcpy(body + 8, &snap_lsn, 8); + for (int i = 16; i < 256; i++) + body[i] = (uint8_t)(i * 3 + 7); /* position-dependent, non-repeating */ + uint32_t fake_fmt = 1u; + uint32_t fake_job_count = 0u; + uint32_t fake_queue_count = 0u; + memcpy(body + 57, &fake_fmt, sizeof(fake_fmt)); + memcpy(body + 61, &fake_job_count, sizeof(fake_job_count)); + memcpy(body + 65, &fake_queue_count, sizeof(fake_queue_count)); + + persist_t *p = calloc(1, sizeof(persist_t)); + TEST_ASSERT_NOT_NULL(p); + persist_config_t pcfg = persist_config_defaults(); + persist_set_workdir(p, tmpdir, "ctrl-snaprecv3"); + persist_init(p, &pcfg, NULL); + persist_start(p); + + ha_state_t ha; + snap_recv_ha_init(&ha); + + /* Deliver out of chunk_index order: 0, 2, 1, 3 (isLast). */ + int order[4] = { 0, 2, 1, 3 }; + for (int oi = 0; oi < 4; oi++) { + int idx = order[oi]; + snapshot_transfer_t c; + memset(&c, 0, sizeof(c)); + c.snapshot_lsn = snap_lsn; + c.chunk_index = (uint32_t)idx; + c.chunk_offset = (uint64_t)(idx * 64); + c.total_size = (uint64_t)total_size; + c.is_last = (idx == 3); + + int rc = ha_handle_snapshot_chunk(&ha, p, snap_dir, &c, body + idx * 64, 64); + TEST_ASSERT_EQUAL_INT(0, rc); + } + + /* Final file must exist (finalize triggers once all 4 chunks are in, + * regardless of the delivery order above) and be byte-identical to the + * correctly-ordered source body — proves each chunk landed at its own + * chunk_offset, not at the position it happened to arrive in. */ + char final_path[600]; + snprintf(final_path, sizeof(final_path), + "%s/snap-%016" PRIx64 ".bin", snap_dir, snap_lsn); + struct stat st; + TEST_ASSERT_EQUAL_INT_MESSAGE(0, stat(final_path, &st), + "snap-{lsn}.bin should exist once all " + "chunks are in, regardless of arrival order"); + TEST_ASSERT_EQUAL_UINT64((uint64_t)total_size, (uint64_t)st.st_size); + + FILE *f = fopen(final_path, "rb"); + TEST_ASSERT_NOT_NULL(f); + uint8_t got[256]; + TEST_ASSERT_EQUAL_size_t(256, fread(got, 1, 256, f)); + fclose(f); + TEST_ASSERT_EQUAL_MEMORY_MESSAGE(body, got, 256, + "assembled file must be byte-identical to the source despite " + "out-of-chunk_index-order delivery (bug-1389)"); + + uint64_t got_lsn = (uint64_t)atomic_load_explicit(&ha.applied_lsn, + memory_order_acquire); + TEST_ASSERT_EQUAL_UINT64(snap_lsn, got_lsn); + + ha_fini(&ha); + persist_stop(p); + persist_fini(p); + free(p); + + char cmd[512]; + snprintf(cmd, sizeof(cmd), "rm -rf %s", tmpdir); + system(cmd); + printf("test_snapshot_recv_out_of_order_chunks_assemble_correctly: PASS\n"); +} + /* Test 3.5d: reconnected standby discards snap.part and restarts from chunk 0. * * Design §Snapshot Transfer: "Mid-transfer reconnect always restarts snapshot @@ -3709,6 +4010,8 @@ void test_config_repl(void) RUN_TEST(test_load_primary_config_missing_returns_error); RUN_TEST(test_wait_all_with_timeout_slow_threads); RUN_TEST(test_wait_all_with_timeout_fast_threads); + /* bug-1381: push_config_transport_only endpoint-lifetime regression */ + RUN_TEST(test_configpush_transport_holds_endpoint_ref_across_send); /* Task 2.1: OPCode_walCatchupEntry handler */ RUN_TEST(test_wal_catchup_entry_advances_applied_lsn); RUN_TEST(test_wal_catchup_entry_error_skipped_no_crash); @@ -3733,6 +4036,7 @@ void test_config_repl(void) /* Task 3.5: snapshot chunk transfer (sender side) */ RUN_TEST(test_snapshot_transfer_two_chunks); RUN_TEST(test_snapshot_transfer_small_less_than_one_chunk); + RUN_TEST(test_snapshot_transfer_stops_and_propagates_send_failure); /* Task 3.5c: standby snapshot receive handler */ RUN_TEST(test_snapshot_recv_two_chunks_assembles_and_renames); /* Task 6.1: legacy position-only file consumed end-to-end via the @@ -3742,6 +4046,7 @@ void test_config_repl(void) RUN_TEST(test_snapshot_recv_null_args_fail); /* Task 6.4-A: 128KB snapshot in 2 chunks — full-size receive path */ RUN_TEST(test_snapshot_recv_128kb_two_chunks); + RUN_TEST(test_snapshot_recv_out_of_order_chunks_assemble_correctly); /* Task 3.5d: reconnect discards snap.part */ RUN_TEST(test_snapshot_recv_reconnect_discards_snappart); /* Task 5.1: handle_ctrlhello connection setup only — no repl messages */ diff --git a/src/controller/testsuite/test_snapdiff.c b/src/controller/testsuite/test_snapdiff.c new file mode 100644 index 0000000..bcb9c19 --- /dev/null +++ b/src/controller/testsuite/test_snapdiff.c @@ -0,0 +1,132 @@ +#define _GNU_SOURCE +#include +#include +#include +#include +#include +#include + +#include + +#include "snapdiff.h" + +static char g_dir[256]; + +static void fresh_dir(void) +{ + strcpy(g_dir, "/tmp/xce_snapdiff_XXXXXX"); + TEST_ASSERT_NOT_NULL(mkdtemp(g_dir)); +} + +static void cleanup_dir(void) +{ + char cmd[512]; + snprintf(cmd, sizeof(cmd), "rm -rf %s", g_dir); + system(cmd); +} + +static void write_file(const char* path, size_t size, uint8_t fill, + size_t flip_at, uint8_t flip_to) +{ + uint8_t* buf = malloc(size); + memset(buf, fill, size); + if (flip_at < size) + buf[flip_at] = flip_to; + FILE* fp = fopen(path, "wb"); + TEST_ASSERT_NOT_NULL(fp); + fwrite(buf, 1, size, fp); + fclose(fp); + free(buf); +} + +static void test_identical_files(void) +{ + fresh_dir(); + char p1[512], p2[512]; + snprintf(p1, sizeof(p1), "%s/a.bin", g_dir); + snprintf(p2, sizeof(p2), "%s/b.bin", g_dir); + write_file(p1, 1000, 0x42, 2000 /* no flip */, 0); + write_file(p2, 1000, 0x42, 2000, 0); + + snapdiff_report_t rep; + int rc = snapdiff_paths(p1, p2, stdout, &rep); + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_EQUAL_INT(1, rep.identical); + TEST_ASSERT_EQUAL_INT(-1, (int)rep.diff_offset); + cleanup_dir(); +} + +static void test_size_mismatch(void) +{ + fresh_dir(); + char p1[512], p2[512]; + snprintf(p1, sizeof(p1), "%s/a.bin", g_dir); + snprintf(p2, sizeof(p2), "%s/b.bin", g_dir); + write_file(p1, 1000, 0x42, 2000, 0); + write_file(p2, 999, 0x42, 2000, 0); + + snapdiff_report_t rep; + int rc = snapdiff_paths(p1, p2, stdout, &rep); + TEST_ASSERT_EQUAL_INT(1, rc); + TEST_ASSERT_EQUAL_INT(1, rep.size_mismatch); + TEST_ASSERT_EQUAL_INT(0, rep.identical); + cleanup_dir(); +} + +/* bug-1389 regression shape: two SNAPDIFF_CHUNK_SIZE-sized chunks, content + * diverges starting at the exact first byte of chunk index 1 — the OPCode_ + * snapshotTransfer chunk-boundary corruption signature. */ +static void test_diff_at_chunk_boundary_is_flagged(void) +{ + fresh_dir(); + char p1[512], p2[512]; + snprintf(p1, sizeof(p1), "%s/a.bin", g_dir); + snprintf(p2, sizeof(p2), "%s/b.bin", g_dir); + size_t size = SNAPDIFF_CHUNK_SIZE * 2; + write_file(p1, size, 0x11, size, 0); + write_file(p2, size, 0x11, SNAPDIFF_CHUNK_SIZE /* flip first byte of chunk 1 */, + 0x22); + + snapdiff_report_t rep; + int rc = snapdiff_paths(p1, p2, stdout, &rep); + TEST_ASSERT_EQUAL_INT(1, rc); + TEST_ASSERT_EQUAL_INT(0, rep.identical); + TEST_ASSERT_EQUAL_INT(SNAPDIFF_CHUNK_SIZE, (int)rep.diff_offset); + TEST_ASSERT_EQUAL_INT(1, rep.at_chunk_boundary); + TEST_ASSERT_EQUAL_INT(1, (int)rep.chunk_index); + cleanup_dir(); +} + +static void test_diff_mid_chunk_is_not_flagged_as_boundary(void) +{ + fresh_dir(); + char p1[512], p2[512]; + snprintf(p1, sizeof(p1), "%s/a.bin", g_dir); + snprintf(p2, sizeof(p2), "%s/b.bin", g_dir); + write_file(p1, 1000, 0x11, 2000, 0); + write_file(p2, 1000, 0x11, 500 /* mid-file, not a chunk boundary */, 0x22); + + snapdiff_report_t rep; + int rc = snapdiff_paths(p1, p2, stdout, &rep); + TEST_ASSERT_EQUAL_INT(1, rc); + TEST_ASSERT_EQUAL_INT(500, (int)rep.diff_offset); + TEST_ASSERT_EQUAL_INT(0, rep.at_chunk_boundary); + cleanup_dir(); +} + +static void test_missing_path_errors(void) +{ + snapdiff_report_t rep; + int rc = snapdiff_paths("/tmp/xce_snapdiff_missing_a_zzz", + "/tmp/xce_snapdiff_missing_b_zzz", stdout, &rep); + TEST_ASSERT_LESS_THAN_INT(0, rc); +} + +void test_snapdiff(void) +{ + RUN_TEST(test_identical_files); + RUN_TEST(test_size_mismatch); + RUN_TEST(test_diff_at_chunk_boundary_is_flagged); + RUN_TEST(test_diff_mid_chunk_is_not_flagged_as_boundary); + RUN_TEST(test_missing_path_errors); +} diff --git a/src/controller/testsuite/test_snapverify.c b/src/controller/testsuite/test_snapverify.c new file mode 100644 index 0000000..1b54a8a --- /dev/null +++ b/src/controller/testsuite/test_snapverify.c @@ -0,0 +1,229 @@ +#define _GNU_SOURCE +#include +#include +#include +#include +#include +#include + +#include + +#include "snapverify.h" + +static char g_dir[256]; + +static void fresh_dir(void) +{ + strcpy(g_dir, "/tmp/xce_snapverify_XXXXXX"); + TEST_ASSERT_NOT_NULL(mkdtemp(g_dir)); +} + +static void cleanup_dir(void) +{ + char cmd[512]; + snprintf(cmd, sizeof(cmd), "rm -rf %s", g_dir); + system(cmd); +} + +/* Writes a checkpoint file matching persist.c write_snapshot()'s exact + * on-disk layout (magic+lsn+cfgref(41, all-zero/absent) + fmt + job_count + + * per-job[len+buf] + queue_count + per-queue[namelen+name+status]). + * job_lens/job_count and queue_names/queue_count are caller-controlled so + * tests can inject a specific defect. `truncate_last_bytes` drops that many + * bytes off the END of the fully-written file (0 = write it whole). */ +static void write_snap(const char* path, uint64_t lsn, uint32_t fmt, + const uint32_t* job_lens, uint32_t job_count, + const char** queue_names, const uint32_t* queue_status, + uint32_t queue_count, int truncate_last_bytes) +{ + FILE* fp = fopen(path, "wb"); + TEST_ASSERT_NOT_NULL(fp); + + char header[16 + 1 + 8 + 32]; + memset(header, 0, sizeof(header)); + memcpy(header, "XCESNAP\0", 8); + memcpy(header + 8, &lsn, 8); + fwrite(header, 1, sizeof(header), fp); + + fwrite(&fmt, sizeof(fmt), 1, fp); + fwrite(&job_count, sizeof(job_count), 1, fp); + for (uint32_t i = 0; i < job_count; i++) { + fwrite(&job_lens[i], sizeof(job_lens[i]), 1, fp); + uint8_t* buf = malloc(job_lens[i]); + memset(buf, 0xAB, job_lens[i]); + fwrite(buf, 1, job_lens[i], fp); + free(buf); + } + + fwrite(&queue_count, sizeof(queue_count), 1, fp); + for (uint32_t i = 0; i < queue_count; i++) { + uint32_t nlen = (uint32_t)strlen(queue_names[i]); + fwrite(&nlen, sizeof(nlen), 1, fp); + fwrite(queue_names[i], 1, nlen, fp); + fwrite(&queue_status[i], sizeof(queue_status[i]), 1, fp); + } + fclose(fp); + + if (truncate_last_bytes > 0) { + long sz = 0; + FILE* f = fopen(path, "rb"); + fseek(f, 0, SEEK_END); + sz = ftell(f); + fclose(f); + TEST_ASSERT_TRUE(truncate(path, sz - truncate_last_bytes) == 0); + } +} + +static void test_legacy_position_only_file_is_ok(void) +{ + fresh_dir(); + char path[512]; + snprintf(path, sizeof(path), "%s/snap-000000000000000a.capnp", g_dir); + + char header[16]; + memcpy(header, "XCESNAP\0", 8); + uint64_t lsn = 10; + memcpy(header + 8, &lsn, 8); + FILE* fp = fopen(path, "wb"); + TEST_ASSERT_NOT_NULL(fp); + fwrite(header, 1, sizeof(header), fp); + fclose(fp); + + snapverify_report_t rep; + int rc = snapverify_path(path, stdout, &rep); + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_EQUAL_INT(1, rep.ok); + TEST_ASSERT_EQUAL_INT(0, rep.has_entity_section); + TEST_ASSERT_EQUAL_UINT64(10, rep.checkpoint_lsn); + cleanup_dir(); +} + +static void test_entity_bearing_well_formed_file_is_ok(void) +{ + fresh_dir(); + char path[512]; + snprintf(path, sizeof(path), "%s/snap-0000000000000064.capnp", g_dir); + + uint32_t job_lens[3] = { 100, 200, 50 }; + const char* qnames[2] = { "gpu", "cpu" }; + uint32_t qstatus[2] = { 1, 0 }; + write_snap(path, 100, 1, job_lens, 3, qnames, qstatus, 2, 0); + + snapverify_report_t rep; + int rc = snapverify_path(path, stdout, &rep); + TEST_ASSERT_EQUAL_INT(0, rc); + TEST_ASSERT_EQUAL_INT(1, rep.ok); + TEST_ASSERT_EQUAL_INT(1, rep.has_entity_section); + TEST_ASSERT_EQUAL_UINT32(1, rep.format_version); + TEST_ASSERT_EQUAL_UINT32(3, rep.job_count); + TEST_ASSERT_EQUAL_UINT32(3, rep.jobs_ok); + TEST_ASSERT_EQUAL_UINT32(2, rep.queue_count); + TEST_ASSERT_EQUAL_UINT32(2, rep.queues_ok); + TEST_ASSERT_EQUAL_INT(-1, (int)rep.fail_offset); + cleanup_dir(); +} + +static void test_bad_magic_is_einval(void) +{ + fresh_dir(); + char path[512]; + snprintf(path, sizeof(path), "%s/snap-bad.capnp", g_dir); + FILE* fp = fopen(path, "wb"); + TEST_ASSERT_NOT_NULL(fp); + fwrite("NOTASNAP", 1, 8, fp); + char zero[8] = {0}; + fwrite(zero, 1, 8, fp); + fclose(fp); + + snapverify_report_t rep; + int rc = snapverify_path(path, stdout, &rep); + TEST_ASSERT_EQUAL_INT(-EINVAL, rc); + TEST_ASSERT_EQUAL_INT(0, rep.ok); + cleanup_dir(); +} + +static void test_truncated_header_is_eio(void) +{ + fresh_dir(); + char path[512]; + snprintf(path, sizeof(path), "%s/snap-short.capnp", g_dir); + FILE* fp = fopen(path, "wb"); + TEST_ASSERT_NOT_NULL(fp); + fwrite("XCE", 1, 3, fp); /* far short of the 16-byte minimum */ + fclose(fp); + + snapverify_report_t rep; + int rc = snapverify_path(path, stdout, &rep); + TEST_ASSERT_EQUAL_INT(-EIO, rc); + TEST_ASSERT_EQUAL_INT(0, rep.ok); + cleanup_dir(); +} + +static void test_unrecognized_format_version_is_einval(void) +{ + fresh_dir(); + char path[512]; + snprintf(path, sizeof(path), "%s/snap-0000000000000001.capnp", g_dir); + uint32_t job_lens[1] = { 10 }; + write_snap(path, 1, 99 /* bogus version */, job_lens, 1, NULL, NULL, 0, 0); + + snapverify_report_t rep; + int rc = snapverify_path(path, stdout, &rep); + TEST_ASSERT_EQUAL_INT(-EINVAL, rc); + TEST_ASSERT_EQUAL_INT(0, rep.ok); + TEST_ASSERT_EQUAL_UINT32(99, rep.format_version); + cleanup_dir(); +} + +/* bug-1389 regression shape: a job frame's length field decodes to garbage + * (e.g. transport corrupted a chunk) mid-stream. snapverify must stop at + * exactly the bad job, report how many parsed cleanly before it, and pinpoint + * the byte offset of the bad length field — the diagnostic this tool exists + * to produce instead of a bare "persist_load_snapshot failed rc=-5". */ +static void test_corrupt_job_payload_len_reports_exact_failure_point(void) +{ + fresh_dir(); + char path[512]; + snprintf(path, sizeof(path), "%s/snap-0000000000000002.capnp", g_dir); + + uint32_t job_lens[4] = { 50, 60, 70, 80 }; + write_snap(path, 2, 1, job_lens, 4, NULL, NULL, 0, 0); + + /* Corrupt job 2's length field (3rd job, after header(57)+fmt(4)+ + * job_count(4)+job0(4+50)+job1(4+60) = 57+4+4+54+64 = 183) to a value + * that cannot possibly fit in the remaining file. */ + FILE* fp = fopen(path, "r+b"); + TEST_ASSERT_NOT_NULL(fp); + long job2_len_off = 57 + 4 + 4 + (4 + 50) + (4 + 60); + fseek(fp, job2_len_off, SEEK_SET); + uint32_t garbage = 0x7FFFFFFF; + fwrite(&garbage, sizeof(garbage), 1, fp); + fclose(fp); + + snapverify_report_t rep; + int rc = snapverify_path(path, stdout, &rep); + TEST_ASSERT_EQUAL_INT(-EIO, rc); + TEST_ASSERT_EQUAL_INT(0, rep.ok); + TEST_ASSERT_EQUAL_UINT32(2, rep.jobs_ok); /* job 0, job 1 parsed fine */ + TEST_ASSERT_EQUAL_INT(job2_len_off, (int)rep.fail_offset); + cleanup_dir(); +} + +static void test_missing_path_errors(void) +{ + snapverify_report_t rep; + int rc = snapverify_path("/tmp/xce_snapverify_does_not_exist_zzz", stdout, + &rep); + TEST_ASSERT_LESS_THAN_INT(0, rc); +} + +void test_snapverify(void) +{ + RUN_TEST(test_legacy_position_only_file_is_ok); + RUN_TEST(test_entity_bearing_well_formed_file_is_ok); + RUN_TEST(test_bad_magic_is_einval); + RUN_TEST(test_truncated_header_is_eio); + RUN_TEST(test_unrecognized_format_version_is_einval); + RUN_TEST(test_corrupt_job_payload_len_reports_exact_failure_point); + RUN_TEST(test_missing_path_errors); +} diff --git a/src/lib/testsuite/test_wal_wire.c b/src/lib/testsuite/test_wal_wire.c index 9ac6943..d31469c 100644 --- a/src/lib/testsuite/test_wal_wire.c +++ b/src/lib/testsuite/test_wal_wire.c @@ -248,6 +248,113 @@ static void test_snapshot_chunk_roundtrip(void) printf("test_snapshot_chunk_roundtrip: PASS\n"); } +/* bug-1389: a force-snapshot soak observed the RECEIVED chunk's content + * diverging from the SENT content starting at an exact SNAPSHOT_CHUNK_SIZE + * (65536-byte) boundary, byte-shifted by a nondeterministic amount, while + * the total size stayed correct. test_snapshot_chunk_roundtrip above only + * exercises a 300-byte payload — far short of the 65536-byte production + * chunk size that actually forces xce_snapshot_chunk_encode_raw's + * capn_new_list8 call to overflow the root segment and cross into a second + * segment via a far pointer (the exact mechanism bug-847 already proved can + * corrupt a capnp Data field in this codebase, for a different message + * shape). This reproduces the REAL production chunk size in a single + * deterministic encode->decode round-trip, no network involved — isolating + * whether the corruption is in this library's capnp handling at all, or + * only manifests under real transport/timing conditions. */ +static void test_snapshot_chunk_roundtrip_full_size_payload(void) +{ + /* Arrange: exactly SNAPSHOT_CHUNK_SIZE (config_repl.c) bytes, with a + * position-dependent, non-repeating-within-256-bytes pattern so ANY + * byte-shift (not just a zeroed/dropped payload) is immediately + * detectable by TEST_ASSERT_EQUAL_MEMORY instead of accidentally + * matching shifted-but-still-plausible content. */ + enum { CHUNK_SZ = 65536 }; + uint8_t *payload = malloc(CHUNK_SZ); + TEST_ASSERT_NOT_NULL(payload); + for (size_t i = 0; i < CHUNK_SZ; i++) + payload[i] = (uint8_t)((i * 2654435761u) >> 24); /* Knuth multiplicative hash */ + + snapshot_transfer_t chunk; + memset(&chunk, 0, sizeof(chunk)); + chunk.snapshot_lsn = 117; + chunk.chunk_index = 0; + chunk.chunk_offset = 0; + chunk.total_size = 2 * CHUNK_SZ; + chunk.is_last = 0; + chunk.checkpoint_term = 4; + + io_packet_t *pkt = (io_packet_t *)calloc(1, sizeof(*pkt)); + TEST_ASSERT_NOT_NULL(pkt); + TEST_ASSERT_EQUAL_INT(0, + xce_snapshot_chunk_encode_raw(&chunk, payload, CHUNK_SZ, &test_peer, + 11, pkt)); + + /* Act */ + xce_request_t req; + memset(&req, 0, sizeof(req)); + TEST_ASSERT_EQUAL_INT(0, xce_decode(pkt, &req)); + + /* Assert: envelope + every byte of a full-size chunk survives intact */ + TEST_ASSERT_EQUAL_INT(OPCode_snapshotTransfer, req.opcode); + TEST_ASSERT_EQUAL_INT(Request_body_snapshottransfermsg, req.body_which); + snapshot_transfer_t *rx = req.body.snapshottransfermsg; + TEST_ASSERT_NOT_NULL(rx); + TEST_ASSERT_EQUAL_UINT64(117, rx->snapshot_lsn); + TEST_ASSERT_EQUAL_UINT64(4, rx->checkpoint_term); + TEST_ASSERT_EQUAL_size_t(CHUNK_SZ, rx->raw_payload_len); + TEST_ASSERT_NOT_NULL(rx->raw_payload); + TEST_ASSERT_EQUAL_MEMORY(payload, rx->raw_payload, CHUNK_SZ); + + xce_free_request(&req); + free(payload); + printf("test_snapshot_chunk_roundtrip_full_size_payload: PASS\n"); +} + +/* bug-1389: same as above but for chunk_index 1 (the SECOND chunk of a + * multi-chunk transfer) — the observed force-snapshot corruption started at + * a chunk-boundary OTHER than the very first chunk in one of the two + * repro runs (offset 131072 == chunk index 2's start), so this covers a + * non-zero chunk_index/chunk_offset combination too, not just chunk 0. */ +static void test_snapshot_chunk_roundtrip_full_size_payload_nonzero_index(void) +{ + enum { CHUNK_SZ = 65536 }; + uint8_t *payload = malloc(CHUNK_SZ); + TEST_ASSERT_NOT_NULL(payload); + for (size_t i = 0; i < CHUNK_SZ; i++) + payload[i] = (uint8_t)((i * 2654435761u) >> 24); + + snapshot_transfer_t chunk; + memset(&chunk, 0, sizeof(chunk)); + chunk.snapshot_lsn = 118; + chunk.chunk_index = 2; + chunk.chunk_offset = 2 * CHUNK_SZ; + chunk.total_size = 4 * CHUNK_SZ; + chunk.is_last = 0; + chunk.checkpoint_term = 4; + + io_packet_t *pkt = (io_packet_t *)calloc(1, sizeof(*pkt)); + TEST_ASSERT_NOT_NULL(pkt); + TEST_ASSERT_EQUAL_INT(0, + xce_snapshot_chunk_encode_raw(&chunk, payload, CHUNK_SZ, &test_peer, + 12, pkt)); + + xce_request_t req; + memset(&req, 0, sizeof(req)); + TEST_ASSERT_EQUAL_INT(0, xce_decode(pkt, &req)); + + snapshot_transfer_t *rx = req.body.snapshottransfermsg; + TEST_ASSERT_NOT_NULL(rx); + TEST_ASSERT_EQUAL_UINT32(2, rx->chunk_index); + TEST_ASSERT_EQUAL_UINT64(2 * CHUNK_SZ, rx->chunk_offset); + TEST_ASSERT_EQUAL_size_t(CHUNK_SZ, rx->raw_payload_len); + TEST_ASSERT_NOT_NULL(rx->raw_payload); + TEST_ASSERT_EQUAL_MEMORY(payload, rx->raw_payload, CHUNK_SZ); + + xce_free_request(&req); + free(payload); + printf("test_snapshot_chunk_roundtrip_full_size_payload_nonzero_index: PASS\n"); +} + static void test_snapshot_chunk_empty_payload(void) { /* Zero-length payload (defensive): envelope round-trips, raw NULL/0 */ @@ -286,5 +393,7 @@ void test_wal_wire(void) RUN_TEST(test_wal_batch_entry_without_payload); RUN_TEST(test_wal_batch_encode_rejects_invalid_args); RUN_TEST(test_snapshot_chunk_roundtrip); + RUN_TEST(test_snapshot_chunk_roundtrip_full_size_payload); + RUN_TEST(test_snapshot_chunk_roundtrip_full_size_payload_nonzero_index); RUN_TEST(test_snapshot_chunk_empty_payload); }