#define _GNU_SOURCE #include "waldump.h" #include #include #include #include #include #include #include #include #include #include "config.h" /* WAL_HDR_CRC_SIZE, WAL_CRC_NONE, MAX_WAL_PAYLOAD_BYTES */ #include "controller.capnp.h" /* Job decode for WAL_JOB_SUBMIT */ #include "utils.h" /* fmt_jobid — human-readable JobId formatting */ #include "wal_types.h" #define WAL_SEG_FMT "%s/seg-%016" PRIx64 ".wal" #define MAX_SEGS 4096 /* ------------------------------------------------------------------ */ /* Small helpers */ /* ------------------------------------------------------------------ */ static const char* wal_type_name(uint32_t t) { switch (t) { case WAL_JOB_SUBMIT: return "JOB_SUBMIT"; case WAL_JOB_ARRAY_SUBMIT: return "JOB_ARRAY_SUBMIT"; /* job-array-single- * wal-record task 6.1: name the type so the HA soak's WAL-record-count * assertion (exactly one JOB_ARRAY_SUBMIT per array submission, * regardless of array size) can grep an exact type name instead of * the generic "UNKNOWN" bucket, which would also catch any future, * genuinely-unrecognized type. */ case WAL_JOB_START: return "JOB_START"; case WAL_JOB_FINISH: return "JOB_FINISH"; case WAL_JOB_FAIL: return "JOB_FAIL"; case WAL_JOB_STOP: return "JOB_STOP"; case WAL_JOB_RESUME: return "JOB_RESUME"; case WAL_JOB_UPDATE: return "JOB_UPDATE"; case WAL_JOB_RUSAGE: return "JOB_RUSAGE"; case WAL_JOB_CKPT: return "JOB_CKPT"; case WAL_JOB_PREASON: return "JOB_PREASON"; case WAL_QUEUE_CTRL: return "QUEUE_CTRL"; case WAL_WORKER_CTRL: return "WORKER_CTRL"; case WAL_NOOP: return "NOOP"; case WAL_CHECKPOINT: return "CHECKPOINT"; default: return "UNKNOWN"; } } /* Decoded fields of a WAL_JOB_SUBMIT Cap'n Proto payload (debug view). */ typedef struct { uint64_t jobid; uint64_t subtime; /* job_submit stamps this with time(NULL) per submit */ uint32_t state; } submit_info_t; /* Decode a WAL_JOB_SUBMIT payload. Mirrors restore_job() (job.c) but reads a * few fields only, zero-alloc. Returns 0 on success, -1 if undecodable. * subtime is the decisive bug-845 signal: two entries of the SAME job written * in one submit share a subtime (a literal double-append); two separate * submits (e.g. a failover retry) would differ. */ static int decode_submit_info(const uint8_t* payload, uint32_t len, submit_info_t* out) { if (!payload || len == 0) return -1; struct capn cctx; if (capn_init_mem(&cctx, payload, len, 0) != 0) return -1; Job_ptr jp; jp.p = capn_getp(capn_root(&cctx), 0, 1); if (jp.p.type == CAPN_NULL) { capn_free(&cctx); return -1; } struct Job cj; memset(&cj, 0, sizeof(cj)); read_Job(&cj, jp); out->jobid = cj.jobid; out->subtime = cj.subtime; out->state = cj.state; capn_free(&cctx); return 0; } /* True for the WAL types whose payload is the flat, non-capnp * wal_jobstate_t {jobid, state, _pad} (wal_types.h) — every worker-reported * job-state fact (START/FINISH/FAIL/STOP/RESUME/UPDATE/RUSAGE/CKPT/PREASON). * Added for raft-apply-decoupling task 7.2: the kill-mid-fact-stream live * regression needs to verify, per jobid, which fact WAL entries actually * committed on each node — before this, waldump only decoded jobid for * WAL_JOB_SUBMIT, leaving every fact entry printed as "jobid=?". */ static int wal_type_is_jobstate(uint32_t type) { switch (type) { case WAL_JOB_START: case WAL_JOB_FINISH: case WAL_JOB_FAIL: case WAL_JOB_STOP: case WAL_JOB_RESUME: case WAL_JOB_UPDATE: case WAL_JOB_RUSAGE: case WAL_JOB_CKPT: case WAL_JOB_PREASON: return 1; default: return 0; } } /* Decode a wal_jobstate_t payload. Native-endian flat struct (single-node WAL * — see wal_types.h), so this is a bounds-checked copy, not a parser. Returns * 0 on success, -1 if the payload is too short to hold the struct. */ static int decode_jobstate_info(const uint8_t* payload, uint32_t len, wal_jobstate_t* out) { if (!payload || len < sizeof(*out)) return -1; memcpy(out, payload, sizeof(*out)); return 0; } /* True if `dir` directly contains at least one seg-*.wal file. */ static int dir_has_segments(const char* dir) { DIR* d = opendir(dir); if (!d) return 0; int found = 0; struct dirent* e; while ((e = readdir(d)) != NULL) { size_t n = strlen(e->d_name); if (n > 4 && strncmp(e->d_name, "seg-", 4) == 0 && strcmp(e->d_name + n - 4, ".wal") == 0) { found = 1; break; } } closedir(d); return found; } /* Resolve `path` (seg file / wal dir / node-or-data dir) to the wal dir that * holds the seg-*.wal files. Writes it to `out`. Returns 0, or -errno. When * `path` is itself a single seg file, sets *single_seg to that file path. */ static int resolve_wal_dir(const char* path, char* out, size_t outsz, char* single_seg, size_t segsz) { struct stat st; if (stat(path, &st) != 0) return -errno; single_seg[0] = '\0'; if (S_ISREG(st.st_mode)) { /* Only a seg-*.wal file is a WAL segment. Reject anything else (e.g. a * config blob like conf/cluster.bin) instead of parsing it as a WAL and * silently reporting 0 entries. */ const char* base = strrchr(path, '/'); base = base ? base + 1 : path; size_t bl = strlen(base); if (strncmp(base, "seg-", 4) != 0 || bl < 5 || strcmp(base + bl - 4, ".wal") != 0) return -EINVAL; snprintf(single_seg, segsz, "%s", path); /* wal dir = dirname(path); only used for display */ snprintf(out, outsz, "%s", path); char* slash = strrchr(out, '/'); if (slash) *slash = '\0'; return 0; } if (!S_ISDIR(st.st_mode)) return -ENOTDIR; /* A directory: try it directly, then data/wal, then wal/. */ if (dir_has_segments(path)) { snprintf(out, outsz, "%s", path); return 0; } char cand[4096]; snprintf(cand, sizeof(cand), "%s/data/wal", path); if (dir_has_segments(cand)) { snprintf(out, outsz, "%s", cand); return 0; } snprintf(cand, sizeof(cand), "%s/wal", path); if (dir_has_segments(cand)) { snprintf(out, outsz, "%s", cand); return 0; } return -ENOENT; /* no segments anywhere under path */ } /* Parse the start LSN out of a seg file basename ("seg-.wal"). */ static int parse_seg_start(const char* name, uint64_t* lsn) { return sscanf(name, "seg-%" SCNx64 ".wal", lsn) == 1 ? 0 : -1; } /* Collect + sort segment start LSNs under wal_dir. Returns count, or -errno. */ static int collect_segments(const char* wal_dir, uint64_t* seg_lsns, int max) { DIR* d = opendir(wal_dir); if (!d) return -errno; int n = 0; struct dirent* e; while ((e = readdir(d)) != NULL && n < max) { uint64_t s; if (parse_seg_start(e->d_name, &s) == 0) seg_lsns[n++] = s; } closedir(d); for (int i = 1; i < n; i++) { /* insertion sort — segment count is tiny */ uint64_t key = seg_lsns[i]; int j = i - 1; while (j >= 0 && seg_lsns[j] > key) { seg_lsns[j + 1] = seg_lsns[j]; j--; } seg_lsns[j + 1] = key; } return n; } /* ------------------------------------------------------------------ */ /* Per-segment scan: prints entries, validates the CRC chain, detects */ /* LSN gaps. Carries the chain CRC and prev LSN across segment calls. */ /* ------------------------------------------------------------------ */ static void scan_segment(const char* seg_path, FILE* out, uint32_t* last_crc, uint64_t* prev_lsn, uint64_t* prev_submit_jobid, uint32_t* prev_submit_pcrc, waldump_stats_t* st, waldump_visit_fn visit, void* visit_ctx, int* visit_stop) { int fd = open(seg_path, O_RDONLY); if (fd < 0) { fprintf(out, " !! cannot open %s: %s\n", seg_path, strerror(errno)); return; } uint32_t off = 0; for (;;) { wal_entry_hdr_t hdr; ssize_t hn = pread(fd, &hdr, sizeof(hdr), (off_t)off); if (hn == 0) break; /* clean EOF */ if (hn < (ssize_t)sizeof(hdr)) break; /* truncated / pre-allocated zeros */ if (hdr.lsn == 0 && hdr.type == 0 && hdr.payload_len == 0) break; /* pre-allocated zero region = clean end */ /* Bound payload before trusting it (corruption guard). */ if (hdr.payload_len > MAX_WAL_PAYLOAD_BYTES) { fprintf(out, " !! LSN %" PRIu64 " payload_len %u exceeds max — stop\n", hdr.lsn, hdr.payload_len); st->crc_broken = 1; break; } /* CRC chain check (identical rule to persist recovery). */ int chain_ok; if (off == 0) chain_ok = (hdr.prev_crc == WAL_CRC_NONE || *last_crc == WAL_CRC_NONE || hdr.prev_crc == *last_crc); else chain_ok = (hdr.prev_crc == *last_crc); if (!chain_ok) { fprintf(out, " !! LSN %" PRIu64 " prev_crc chain break — stop\n", hdr.lsn); st->crc_broken = 1; break; } /* Verify entry CRC over header + payload. */ uint8_t* payload = NULL; uint32_t want_crc = crc32(0L, (const Bytef*)&hdr, WAL_HDR_CRC_SIZE); if (hdr.payload_len > 0) { payload = malloc(hdr.payload_len); if (!payload) break; ssize_t pn = pread(fd, payload, hdr.payload_len, (off_t)(off + sizeof(hdr))); if (pn != (ssize_t)hdr.payload_len) { free(payload); break; /* short read — clean end */ } want_crc = crc32(want_crc, (const Bytef*)payload, hdr.payload_len); } if (hdr.crc != want_crc) { fprintf(out, " !! LSN %" PRIu64 " CRC mismatch — stop\n", hdr.lsn); st->crc_broken = 1; free(payload); break; } /* Gap detection against the running LSN sequence. */ if (*prev_lsn == 0) { if (hdr.lsn > 1) { fprintf(out, " *** GAP: missing prefix LSN 1..%" PRIu64 " (%" PRIu64 " entries absent) ***\n", hdr.lsn - 1, hdr.lsn - 1); st->n_gaps++; } } else if (hdr.lsn > *prev_lsn + 1) { fprintf(out, " *** GAP: missing LSN %" PRIu64 "..%" PRIu64 " ***\n", *prev_lsn + 1, hdr.lsn - 1); st->n_gaps++; } /* payload-only CRC — isolates the blob from the header (LSN/term/ts), * so two entries carrying the SAME job blob share a pcrc regardless of * their differing LSNs. Decisive for bug-845: equal pcrc on an adjacent * same-jobid pair ⇒ a literal byte-for-byte double-append. */ uint32_t pcrc = (hdr.payload_len > 0) ? crc32(0L, (const Bytef*)payload, hdr.payload_len) : 0; /* Surface the CRC-valid entry to the visitor so a caller can index the * log by (lsn -> term, pcrc) — the single canonical decoder, no second * parser. */ if (visit) { waldump_entry_t e = { .lsn = hdr.lsn, .term = hdr.term, .type = hdr.type, .payload_len = hdr.payload_len, .pcrc = pcrc, }; if (visit(visit_ctx, &e) != 0) { if (visit_stop) *visit_stop = 1; free(payload); break; /* visitor asked to stop */ } } /* Per-entry line, with jobid/subtime for submits. Sized to hold the * formatted JobId (fmt_jobid, up to ~40 chars) + subtime + state. */ char info[192] = ""; int is_dup = 0; int dup_identical = 0; if (hdr.type == WAL_JOB_SUBMIT) { submit_info_t si; uint64_t jobid = 0; if (decode_submit_info(payload, hdr.payload_len, &si) == 0) { jobid = si.jobid; /* Render the JobId in the same human-readable form the client * (joblist / jobsub) prints — fmt_jobid (utils.c) — so a waldump * jobid can be eyeballed against an `expect-job ` / joblist * entry without converting the raw uint64. */ char jidStr[128]; if (fmt_jobid(jidStr, sizeof(jidStr), si.jobid) < 0) snprintf(jidStr, sizeof(jidStr), "%" PRIu64, si.jobid); snprintf(info, sizeof(info), "jobid=%s subtime=%" PRIu64 " state=%u", jidStr, si.subtime, si.state); } else { snprintf(info, sizeof(info), "jobid=?"); } st->n_submits++; /* bug-845: adjacent duplicate submit — same jobid as the previous * entry at lsn = prev_lsn + 1. Capture identity BEFORE updating the * carried pcrc. */ if (jobid != 0 && jobid == *prev_submit_jobid && hdr.lsn == *prev_lsn + 1) { is_dup = 1; dup_identical = (pcrc == *prev_submit_pcrc); st->n_dup_submits++; if (dup_identical) st->n_dup_identical++; } *prev_submit_jobid = jobid; *prev_submit_pcrc = pcrc; } else if (wal_type_is_jobstate(hdr.type)) { /* Worker-reported job-state fact (task 7.2): decode the flat * {jobid, state, _pad} payload so a per-jobid committed-fact check * (e.g. "did WAL_JOB_FINISH for jobid X actually commit on this * node") can grep waldump output directly, the same way the * existing WAL_JOB_SUBMIT branch above already supports * expect-job/joblist cross-checks. */ wal_jobstate_t js; if (decode_jobstate_info(payload, hdr.payload_len, &js) == 0) { char jidStr[128]; if (fmt_jobid(jidStr, sizeof(jidStr), js.jobid) < 0) snprintf(jidStr, sizeof(jidStr), "%" PRIu64, js.jobid); snprintf(info, sizeof(info), "jobid=%s state=0x%x pad=0x%x", jidStr, js.state, js._pad); } else { snprintf(info, sizeof(info), "jobid=?"); } /* a non-submit entry breaks the adjacency for dup detection */ *prev_submit_jobid = 0; } else { /* a non-submit entry breaks the adjacency for dup detection */ *prev_submit_jobid = 0; } fprintf(out, " %8" PRIu64 " | %5" PRIu64 " | %-11s | %5u | %08x | %s\n", hdr.lsn, hdr.term, wal_type_name(hdr.type), hdr.payload_len, pcrc, info); if (is_dup) fprintf(out, " ^^ DUP submit: payload %s the prior entry\n", dup_identical ? "BYTE-IDENTICAL to" : "DIFFERS from"); free(payload); /* Advance watermarks. */ if (st->first_lsn == 0 || hdr.lsn < st->first_lsn) st->first_lsn = hdr.lsn; if (hdr.lsn > st->last_lsn) st->last_lsn = hdr.lsn; if (hdr.lsn > st->recovered_tip) st->recovered_tip = hdr.lsn; st->n_entries++; *prev_lsn = hdr.lsn; *last_crc = hdr.crc; off += (uint32_t)(sizeof(hdr) + hdr.payload_len); } close(fd); } /* ------------------------------------------------------------------ */ /* Public entry point */ /* ------------------------------------------------------------------ */ int waldump_path(const char* path, FILE* out, waldump_stats_t* stats) { return waldump_path_visit(path, out, stats, NULL, NULL); } int waldump_path_visit(const char* path, FILE* out, waldump_stats_t* stats, waldump_visit_fn visit, void* ctx) { if (!path || !out) return -EINVAL; char wal_dir[4096]; char single_seg[4096]; int rc = resolve_wal_dir(path, wal_dir, sizeof(wal_dir), single_seg, sizeof(single_seg)); if (rc != 0) { /* Clear, actionable error instead of silently scanning a non-WAL path * (e.g. conf/cluster.bin) and reporting "0 entries". */ fprintf(out, "waldump: no WAL segments at '%s' (%s)\n" " point at a node's WAL dir, e.g. " "/controller/ctrl-N/data/wal\n" " (or a specific seg-*.wal file inside it)\n", path, strerror(-rc)); return rc; } waldump_stats_t st; memset(&st, 0, sizeof(st)); uint64_t seg_lsns[MAX_SEGS]; int nseg; /* wider than wal_dir[4096] so the "/seg-.wal" suffix never truncates */ char seg_path[4096 + 64]; if (single_seg[0] != '\0') { /* A single seg file: parse its start for ordering display. */ const char* base = strrchr(single_seg, '/'); base = base ? base + 1 : single_seg; uint64_t start = 0; parse_seg_start(base, &start); seg_lsns[0] = start; nseg = 1; snprintf(seg_path, sizeof(seg_path), "%s", single_seg); /* If the dir holds OTHER segments, this single-file dump shows only a * slice of the log. A node that recovered then caught up keeps its * prefix in seg-...001.wal and the suffix in seg-....wal, * so a single-file dump looks (wrongly) like missing data. Steer the * user to the whole dir. */ uint64_t others[MAX_SEGS]; int total = collect_segments(wal_dir, others, MAX_SEGS); if (total > 1) fprintf(out, "NOTE: %s holds %d segments — this dumps only %s.\n" " Run 'waldump %s' (the directory) for the full " "stitched 1..N log.\n", wal_dir, total, base, wal_dir); } else { nseg = collect_segments(wal_dir, seg_lsns, MAX_SEGS); if (nseg < 0) return nseg; if (nseg == 0) return -ENOENT; } st.n_segments = (uint32_t)nseg; fprintf(out, "== WAL: %s (%d segment%s) ==\n", wal_dir, nseg, nseg == 1 ? "" : "s"); fprintf(out, " %8s | %5s | %-11s | %5s | %8s | %s\n", "LSN", "TERM", "TYPE", "PLEN", "PCRC", "INFO"); uint32_t last_crc = WAL_CRC_NONE; uint64_t prev_lsn = 0; uint64_t prev_submit_jobid = 0; uint32_t prev_submit_pcrc = 0; int visit_stop = 0; for (int i = 0; i < nseg && !visit_stop; i++) { if (single_seg[0] == '\0') snprintf(seg_path, sizeof(seg_path), WAL_SEG_FMT, wal_dir, seg_lsns[i]); scan_segment(seg_path, out, &last_crc, &prev_lsn, &prev_submit_jobid, &prev_submit_pcrc, &st, visit, ctx, &visit_stop); } fprintf(out, "-- summary --\n"); fprintf(out, " segments: %u entries: %u submits: %u\n", st.n_segments, st.n_entries, st.n_submits); fprintf(out, " lsn range: %" PRIu64 "..%" PRIu64 " recovered tip: %" PRIu64 "\n", st.first_lsn, st.last_lsn, st.recovered_tip); fprintf(out, " gaps: %u%s crc chain: %s\n", st.n_gaps, (st.n_gaps && st.first_lsn > 1) ? " (committed prefix missing!)" : "", st.crc_broken ? "BROKEN" : "intact"); if (st.n_dup_submits > 0) fprintf(out, " duplicate submits: %u (byte-identical: %u) — bug-845\n", st.n_dup_submits, st.n_dup_identical); if (stats) *stats = st; return 0; }