#define _GNU_SOURCE #include "walcompare.h" #include #include #include #include #include #include #include #include "persist.h" /* WAL_LEGACY_TERM */ #include "waldump.h" #define WALCMP_MAX_MEMBERS 16 /* One member's collected log, indexed by LSN. `slot[lsn]` is valid when * `present[lsn]` is set. LSNs are 1-based; index 0 is unused. */ typedef struct { uint64_t recovered_tip; /* highest CRC-intact LSN held (waldump tip) */ uint64_t committed_lsn; /* effective committed boundary for this member */ /* raft-ha-core-safety task 5.10a / ha-entity-snapshots task 7.1: the * highest checkpoint LSN this member has installed (0 = never installed * a snapshot -- expect raw WAL from LSN 1, the pre-existing behavior, * unchanged). A member that caught up via a full-checkpoint snapshot * legitimately holds NO raw WAL entries at or below its own floor_lsn -- * that range was restored via entity content, not WAL replay -- so it * must never be flagged MISSING there. Read once per member from its own * snap/ directory (scan_snapshot_floor); never uniform across members * (unlike committed_lsn, which the caller may force to one shared value). */ uint64_t floor_lsn; uint64_t cap; /* allocated index size (LSN range [0..cap-1]) */ uint64_t* term; /* raw on-disk term per LSN (0 = legacy / absent) */ uint32_t* pcrc; /* payload CRC per LSN */ uint8_t* present; /* 1 if this LSN is held */ } member_log_t; /* Visitor context: collect entries into a member_log_t. */ typedef struct { member_log_t* log; int oom; /* set on an allocation failure mid-scan */ } collect_ctx_t; /* Map a raw on-disk term to its matching term: legacy 0 -> WAL_LEGACY_TERM * (R1.8) so two pre-feature entries are not spuriously divergent. */ static uint64_t match_term(uint64_t raw) { return raw == 0 ? WAL_LEGACY_TERM : raw; } /* Grow the per-LSN index so `lsn` is addressable. Returns 0 or -ENOMEM. */ static int member_log_ensure(member_log_t* m, uint64_t lsn) { if (lsn < m->cap) return 0; uint64_t newcap = m->cap ? m->cap : 64; while (newcap <= lsn) newcap *= 2; uint64_t* nt = realloc(m->term, newcap * sizeof(*nt)); uint32_t* np = realloc(m->pcrc, newcap * sizeof(*np)); uint8_t* ne = realloc(m->present, newcap * sizeof(*ne)); if (!nt || !np || !ne) { /* realloc may have partially succeeded; keep whatever grew so the * eventual free() is safe, but report OOM. */ if (nt) m->term = nt; if (np) m->pcrc = np; if (ne) m->present = ne; return -ENOMEM; } /* Zero the newly added tail. */ memset(nt + m->cap, 0, (newcap - m->cap) * sizeof(*nt)); memset(np + m->cap, 0, (newcap - m->cap) * sizeof(*np)); memset(ne + m->cap, 0, (newcap - m->cap) * sizeof(*ne)); m->term = nt; m->pcrc = np; m->present = ne; m->cap = newcap; return 0; } static int collect_visit(void* ctx, const waldump_entry_t* e) { collect_ctx_t* cc = ctx; member_log_t* m = cc->log; if (e->lsn == 0) return 0; /* LSN 0 is never a real committed entry */ if (member_log_ensure(m, e->lsn) != 0) { cc->oom = 1; return 1; /* stop the scan */ } m->term[e->lsn] = e->term; m->pcrc[e->lsn] = e->pcrc; m->present[e->lsn] = 1; return 0; } static void member_log_free(member_log_t* m) { free(m->term); free(m->pcrc); free(m->present); m->term = NULL; m->pcrc = NULL; m->present = NULL; m->cap = 0; } /* Scan one directory for `snap-<16-hex-digit-lsn>.{capnp,bin}` files (the * on-disk naming write_snapshot()/legacy checkpoints use — persist.c) and * return the highest embedded LSN found, or 0 if the directory doesn't exist * or holds no matching file. A bare 16-hex-digit match is required (skips * `.tmp` in-progress files and anything else) so this never mis-parses an * unrelated entry as a checkpoint LSN. */ static uint64_t scan_snap_dir_floor(const char* dir) { DIR* d = opendir(dir); if (!d) return 0; /* no such dir (no snapshot ever installed here) — floor 0 */ uint64_t max_lsn = 0; struct dirent* de; while ((de = readdir(d)) != NULL) { unsigned long long lsn = 0; char ext[8]; /* "snap-XXXXXXXXXXXXXXXX.capnp" / ".bin" — %16llx would also accept * fewer than 16 digits, so additionally check the literal width via * strlen on the numeric run; simplest robust check: require the * filename to be exactly "snap-" + 16 hex chars + "." + ext. */ size_t len = strlen(de->d_name); if (len < 5 + 16 + 1 + 3 || strncmp(de->d_name, "snap-", 5) != 0) continue; int matched = sscanf(de->d_name + 5, "%16llx.%7s", &lsn, ext); if (matched != 2) continue; if (strcmp(ext, "capnp") != 0 && strcmp(ext, "bin") != 0) continue; /* Confirm the hex run is EXACTLY 16 chars (reject a short-digit name * like "snap-1.capnp" that sscanf's %16llx would still happily take). */ int hexlen = 0; while (hexlen < 16 && isxdigit((unsigned char)de->d_name[5 + hexlen])) hexlen++; if (hexlen != 16 || de->d_name[5 + 16] != '.') continue; if ((uint64_t)lsn > max_lsn) max_lsn = (uint64_t)lsn; } closedir(d); return max_lsn; } /* Find this member's own checkpoint floor (raft-ha-core-safety task 5.10a / * ha-entity-snapshots task 7.1). `src->path` may be either a "data" dir * containing sibling `wal/`+`snap/` subdirs (the soak harness's convention: * `$TOP/controller/ctrl-N/data`), or the WAL dir itself passed directly (seen * in unit-test call sites, e.g. `persist_t.wal_path`) — in the latter case the * sibling snap dir sits at the same parent level with `wal` replaced by * `snap`. Try both; a nonexistent candidate just yields floor 0 (harmless). */ static uint64_t member_snapshot_floor(const char* path) { if (!path || !*path) return 0; char candidate[4160]; uint64_t floor1 = 0, floor2 = 0; /* Candidate 1: /snap (path is a "data" parent dir). */ snprintf(candidate, sizeof(candidate), "%s/snap", path); floor1 = scan_snap_dir_floor(candidate); /* Candidate 2: sibling of with a trailing "/wal" (or "wal") * component replaced by "/snap" (path is the WAL dir itself). */ size_t plen = strlen(path); const char* base = strrchr(path, '/'); const char* leaf = base ? base + 1 : path; if (strcmp(leaf, "wal") == 0) { size_t prefix_len = (size_t)(leaf - path); if (prefix_len + 4 < sizeof(candidate)) { memcpy(candidate, path, prefix_len); snprintf(candidate + prefix_len, sizeof(candidate) - prefix_len, "snap"); floor2 = scan_snap_dir_floor(candidate); } } (void)plen; return floor1 > floor2 ? floor1 : floor2; } /* Load one member's WAL into `*m`. Returns 0 or a negative errno. */ static int load_member(const walcompare_member_t* src, member_log_t* m, FILE* out) { memset(m, 0, sizeof(*m)); waldump_stats_t st; memset(&st, 0, sizeof(st)); collect_ctx_t cc = { .log = m, .oom = 0 }; /* Send the human dump to the caller's `out` (or a discard sink). The * visitor harvests the (lsn -> term, pcrc) index from the same decode. */ FILE* dump_sink = out ? out : fopen("/dev/null", "w"); int own_sink = (out == NULL); if (!dump_sink) return -errno; int rc = waldump_path_visit(src->path, dump_sink, &st, collect_visit, &cc); if (own_sink) fclose(dump_sink); if (rc != 0) { member_log_free(m); return rc; /* hard load error (missing path / no segments) */ } if (cc.oom) { member_log_free(m); return -ENOMEM; } m->recovered_tip = st.recovered_tip; /* An explicit committed_lsn wins; 0 means "derive from the shared * boundary" (filled by the caller after all members are loaded). */ m->committed_lsn = src->committed_lsn; /* Always PER-MEMBER, never uniform/shared (unlike committed_lsn) — a * member's own checkpoint floor has nothing to do with any other * member's state. */ m->floor_lsn = member_snapshot_floor(src->path); return 0; } /* Loaded cluster state shared by walcompare() and walcompare_report(). */ typedef struct { member_log_t logs[WALCMP_MAX_MEMBERS]; int n; int loaded; /* members successfully loaded (for cleanup) */ uint64_t boundary; /* highest committed LSN compared (committed edge) */ uint64_t max_tip; /* highest recovered tip across members (tail edge) */ } wc_cluster_t; /* Stable status token for one member at one LSN. */ static const char* wc_status(const member_log_t* m, uint64_t lsn) { if ((lsn < m->cap) && m->present[lsn]) return "present"; if (lsn <= m->floor_lsn) return "compacted"; /* below this member's OWN checkpoint floor — * legitimately absent (entity-restored, not * WAL-held), never a divergence signal */ if (lsn <= m->committed_lsn) return "MISSING"; /* a committed entry this member SHOULD hold (R4.2) */ return "beyond"; /* legitimately not held (above its committed prefix) */ } /* Load all members and compute the committed boundary + tail edge. On a load * error, frees what loaded, reports to `out`, returns a negative errno. */ static int wc_load(const walcompare_member_t* members, int n, FILE* out, wc_cluster_t* cl) { memset(cl, 0, sizeof(*cl)); cl->n = n; for (int i = 0; i < n; i++) { int rc = load_member(&members[i], &cl->logs[i], out); if (rc != 0) { if (out) fprintf(out, "walcompare: cannot load member %d ('%s'): %s\n", i, members[i].path ? members[i].path : "(null)", strerror(-rc)); for (int j = 0; j < cl->loaded; j++) member_log_free(&cl->logs[j]); return rc; } cl->loaded = i + 1; } /* Shared committed boundary = the minimum recovered tip across members: * the largest committed prefix every member can be expected to hold. A * member with an explicit committed_lsn keeps its own. */ uint64_t shared = cl->logs[0].recovered_tip; for (int i = 1; i < n; i++) if (cl->logs[i].recovered_tip < shared) shared = cl->logs[i].recovered_tip; for (int i = 0; i < n; i++) { if (cl->logs[i].committed_lsn == 0) cl->logs[i].committed_lsn = shared; if (cl->logs[i].committed_lsn > cl->boundary) cl->boundary = cl->logs[i].committed_lsn; if (cl->logs[i].recovered_tip > cl->max_tip) cl->max_tip = cl->logs[i].recovered_tip; } return 0; } static void wc_unload(wc_cluster_t* cl) { for (int i = 0; i < cl->loaded; i++) member_log_free(&cl->logs[i]); } /* One machine line for member `i` at `lsn`. Fixed field order, stable tokens. */ static void wc_emit_member(const wc_cluster_t* cl, int i, uint64_t lsn, int committed, FILE* out) { const member_log_t* m = &cl->logs[i]; int holds = (lsn < m->cap) && m->present[lsn]; fprintf(out, "WALCMP member=%d lsn=%" PRIu64 " term=%" PRIu64 " pcrc=%08x status=%s committed=%d\n", i, lsn, holds ? match_term(m->term[lsn]) : 0, holds ? m->pcrc[lsn] : 0u, wc_status(m, lsn), committed); } /* Full per-member machine table over [1..max_tip]; committed=1 within the * committed boundary, 0 in the uncommitted tail. */ static void wc_emit_table(const wc_cluster_t* cl, FILE* out) { for (uint64_t lsn = 1; lsn <= cl->max_tip; lsn++) { int committed = (lsn <= cl->boundary) ? 1 : 0; for (int i = 0; i < cl->n; i++) wc_emit_member(cl, i, lsn, committed, out); } } /* Emit the classification + per-member identity for ONE diverging LSN. */ static void wc_emit_divergence(const wc_cluster_t* cl, uint64_t lsn, int committed, int compared, int present, FILE* out) { fprintf(out, "WALCMP_DIVERGENCE lsn=%" PRIu64 " class=%s compared=%d present=%d\n", lsn, committed ? "committed" : "uncommitted", compared, present); for (int i = 0; i < cl->n; i++) wc_emit_member(cl, i, lsn, committed, out); } /* Compare one committed LSN. Fills `div` (committed=1) and prints the human + * machine report when divergent. Returns 1 if divergent, else 0. */ static int wc_compare_committed(const wc_cluster_t* cl, const walcompare_member_t* members, uint64_t lsn, FILE* out, walcompare_divergence_t* div) { int have = 0, present = 0, have_ref = 0, divergent = 0; uint64_t ref_term = 0; uint32_t ref_pcrc = 0; for (int i = 0; i < cl->n; i++) { if (lsn > cl->logs[i].committed_lsn) continue; /* beyond member i's committed prefix */ if (lsn <= cl->logs[i].floor_lsn) continue; /* below member i's OWN checkpoint floor — legitimately * compacted (entity-restored), never a divergence */ have++; const member_log_t* m = &cl->logs[i]; int holds = (lsn < m->cap) && m->present[lsn]; if (!holds) { divergent = 1; /* member should hold this committed LSN but doesn't */ continue; } present++; uint64_t mt = match_term(m->term[lsn]); uint32_t mp = m->pcrc[lsn]; if (!have_ref) { ref_term = mt; ref_pcrc = mp; have_ref = 1; } else if (mt != ref_term || mp != ref_pcrc) { divergent = 1; } } if (have < 2 || !divergent) return 0; if (div) { memset(div, 0, sizeof(*div)); div->lsn = lsn; div->n_members = cl->n; div->committed = 1; for (int i = 0; i < cl->n && i < 16; i++) { const member_log_t* m = &cl->logs[i]; int holds = (lsn <= m->committed_lsn) && (lsn > m->floor_lsn) && (lsn < m->cap) && m->present[lsn]; div->member_present[i] = holds ? 1 : 0; div->member_term[i] = holds ? match_term(m->term[lsn]) : 0; div->member_pcrc[i] = holds ? m->pcrc[lsn] : 0; } } if (out) { fprintf(out, "*** DIVERGENCE at committed LSN %" PRIu64 " (compared %d, present %d) ***\n", lsn, have, present); for (int i = 0; i < cl->n; i++) { if (lsn > cl->logs[i].committed_lsn) { fprintf(out, " member %d ('%s'): LSN beyond its committed prefix" " (committed=%" PRIu64 ")\n", i, members[i].path, cl->logs[i].committed_lsn); continue; } if (lsn <= cl->logs[i].floor_lsn) { fprintf(out, " member %d ('%s'): LSN below its own checkpoint floor" " (floor=%" PRIu64 ") — legitimately compacted\n", i, members[i].path, cl->logs[i].floor_lsn); continue; } int holds = (lsn < cl->logs[i].cap) && cl->logs[i].present[lsn]; if (holds) fprintf(out, " member %d ('%s'): term=%" PRIu64 " pcrc=%08x\n", i, members[i].path, match_term(cl->logs[i].term[lsn]), cl->logs[i].pcrc[lsn]); else fprintf(out, " member %d ('%s'): MISSING committed entry\n", i, members[i].path); } wc_emit_divergence(cl, lsn, 1, have, present, out); } return 1; } /* Scan the uncommitted tail (boundary < lsn <= max_tip): emit a NON-fatal * class=uncommitted classification for any LSN where >=2 members hold the entry * but disagree on (term, pcrc). Never changes the verdict. */ static void wc_scan_uncommitted(const wc_cluster_t* cl, FILE* out) { for (uint64_t lsn = cl->boundary + 1; lsn <= cl->max_tip; lsn++) { int present = 0, have_ref = 0, divergent = 0; uint64_t ref_term = 0; uint32_t ref_pcrc = 0; for (int i = 0; i < cl->n; i++) { const member_log_t* m = &cl->logs[i]; int holds = (lsn < m->cap) && m->present[lsn]; if (!holds) continue; present++; uint64_t mt = match_term(m->term[lsn]); uint32_t mp = m->pcrc[lsn]; if (!have_ref) { ref_term = mt; ref_pcrc = mp; have_ref = 1; } else if (mt != ref_term || mp != ref_pcrc) { divergent = 1; } } if (present < 2 || !divergent) continue; wc_emit_divergence(cl, lsn, 0, present, present, out); } } /* Shared driver for walcompare() (full_report=0) and walcompare_report() * (full_report=1). */ static int wc_run(const walcompare_member_t* members, int n, FILE* out, walcompare_divergence_t* div, int full_report) { if (!members || n < 2) return -EINVAL; if (n > WALCMP_MAX_MEMBERS) return -E2BIG; wc_cluster_t cl; int rc = wc_load(members, n, out, &cl); if (rc != 0) return rc; if (out) fprintf(out, "== walcompare: %d members, committed boundary up to LSN %" PRIu64 " ==\n", n, cl.boundary); if (full_report && out) wc_emit_table(&cl, out); /* Walk every committed LSN. At each LSN, compare the (term, pcrc) of the * members whose own committed boundary covers it and that hold the entry. * A member that does not hold a committed LSN it SHOULD is a committed- * prefix divergence: a committed entry is missing (R4.2 / R6.4). */ rc = WALCMP_OK; for (uint64_t lsn = 1; lsn <= cl.boundary; lsn++) { if (wc_compare_committed(&cl, members, lsn, out, div)) { rc = WALCMP_DIVERGENT; /* first committed divergence is the verdict */ goto done; } } /* The committed prefix is identical. In full-report mode also classify any * uncommitted-tail divergence (informational, never fatal). */ if (full_report && out) wc_scan_uncommitted(&cl, out); if (out) fprintf(out, "walcompare: OK — committed prefixes identical\n"); done: wc_unload(&cl); return rc; } int walcompare(const walcompare_member_t* members, int n, FILE* out, walcompare_divergence_t* div) { return wc_run(members, n, out, div, 0); } int walcompare_report(const walcompare_member_t* members, int n, FILE* out, walcompare_divergence_t* div) { return wc_run(members, n, out, div, 1); }