diff --git a/.kiro/specs/ha-config-bootstrap/tasks.md b/.kiro/specs/ha-config-bootstrap/tasks.md index c6d46670..31dd3c52 100644 --- a/.kiro/specs/ha-config-bootstrap/tasks.md +++ b/.kiro/specs/ha-config-bootstrap/tasks.md @@ -173,7 +173,7 @@ Dependency order: startup validation (D) unblocks vtaskmgr before election (A); - `make test` fails - _Requirements: 5.6, 5.7_ -- [ ] 7.5 Implement committed_lsn in persist_t and persist_truncate_wal_above() +- [x] 7.5 Implement committed_lsn in persist_t and persist_truncate_wal_above() - Add `_Atomic uint64_t committed_lsn` to `persist_t` (invariant: `committed_lsn <= applied_lsn`) - `persist_truncate_wal_above(persist_t *p, uint64_t cutoff_lsn)`: scan segment files named `seg-{LSN_hex}.wal`; find segment containing `cutoff_lsn`; sequential scan of entries until LSN > cutoff_lsn; truncate at that byte offset; delete later segments; atomically update `p->applied_lsn = cutoff_lsn`; idempotent - Update WAL trim guard: use `committed_lsn` not `applied_lsn` as threshold diff --git a/src/controller/persist.c b/src/controller/persist.c index 3f21668f..bacedd1d 100644 --- a/src/controller/persist.c +++ b/src/controller/persist.c @@ -1269,3 +1269,148 @@ bool worker_should_retain(const xce_worker_t* w, const persist_config_t* cfg) return true; } + +/* ============================================================ + * WAL truncation (task 7.5) + * + * Remove all WAL entries with LSN > cutoff_lsn. + * Idempotent: calling twice with same cutoff_lsn returns 0 both times. + * ============================================================ */ + +int persist_truncate_wal_above(persist_t* p, uint64_t cutoff_lsn) +{ + pthread_mutex_lock(&p->lock); + + /* No-op: nothing written yet, or cutoff covers all written entries */ + if (p->current_lsn == 0 || cutoff_lsn + 1 >= p->current_lsn) { + pthread_mutex_unlock(&p->lock); + return 0; + } + + /* --- Collect segment filenames and sort by start LSN --- */ + uint64_t seg_lsns[4096]; + int nseg = 0; + + DIR* d = opendir(p->wal_path); + if (d) { + struct dirent* e; + while ((e = readdir(d)) != NULL && + nseg < (int)(sizeof(seg_lsns) / sizeof(seg_lsns[0]))) { + if (strncmp(e->d_name, "seg-", 4) != 0) + continue; + uint64_t sl = 0; + if (sscanf(e->d_name, "seg-%" SCNx64 ".wal", &sl) != 1) + continue; + seg_lsns[nseg++] = sl; + } + closedir(d); + } + + if (nseg == 0) { + pthread_mutex_unlock(&p->lock); + return 0; + } + + /* Insertion sort ascending by start LSN */ + for (int i = 1; i < nseg; i++) { + 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; + } + + /* --- Find segment containing cutoff_lsn --- */ + int cutoff_seg_idx = -1; + for (int i = 0; i < nseg; i++) { + if (seg_lsns[i] <= cutoff_lsn) + cutoff_seg_idx = i; + else + break; + } + + if (cutoff_seg_idx < 0) { + pthread_mutex_unlock(&p->lock); + return 0; + } + + /* --- Scan segment to find byte offset of first entry with lsn > cutoff --- */ + char cutoff_seg_path[4096]; + snprintf(cutoff_seg_path, sizeof(cutoff_seg_path), + WAL_SEG_FMT, p->wal_path, seg_lsns[cutoff_seg_idx]); + + int scan_fd = open(cutoff_seg_path, O_RDONLY); + if (scan_fd < 0) { + pthread_mutex_unlock(&p->lock); + return -errno; + } + + uint32_t scan_off = 0; + uint32_t trunc_off = 0; + uint32_t new_last_crc = WAL_CRC_NONE; + bool found_cut = false; + + for (;;) { + wal_entry_hdr_t hdr; + ssize_t n = pread(scan_fd, &hdr, sizeof(hdr), (off_t)scan_off); + if (n < (ssize_t)sizeof(hdr)) + break; /* short read or EOF */ + /* Pre-allocated zero region: type==0 is not a valid WAL type (min=1) */ + if (hdr.lsn == 0 && hdr.type == 0 && hdr.payload_len == 0) + break; + if (hdr.lsn > cutoff_lsn) { + trunc_off = scan_off; + found_cut = true; + break; + } + new_last_crc = hdr.crc; + scan_off += (uint32_t)(sizeof(hdr) + hdr.payload_len); + } + close(scan_fd); + + if (!found_cut) + trunc_off = scan_off; /* end of valid data in segment */ + + /* --- Close the current wal_fd --- */ + if (p->wal_fd >= 0) { + fdatasync(p->wal_fd); + close(p->wal_fd); + p->wal_fd = -1; + } + + /* --- Truncate the cutoff segment at the computed offset --- */ + int new_fd = open(cutoff_seg_path, O_RDWR); + if (new_fd < 0) { + pthread_mutex_unlock(&p->lock); + return -errno; + } + if (ftruncate(new_fd, (off_t)trunc_off) != 0) { + int saved = errno; + close(new_fd); + pthread_mutex_unlock(&p->lock); + return -saved; + } + fdatasync(new_fd); + + /* --- Delete all segments after the cutoff segment --- */ + for (int i = cutoff_seg_idx + 1; i < nseg; i++) { + char del_path[4096]; + snprintf(del_path, sizeof(del_path), + WAL_SEG_FMT, p->wal_path, seg_lsns[i]); + unlink(del_path); + } + + /* --- Update persist_t state under lock --- */ + p->wal_fd = new_fd; + p->seg_offset = trunc_off; + snprintf(p->seg_path, sizeof(p->seg_path), + WAL_SEG_FMT, p->wal_path, seg_lsns[cutoff_seg_idx]); + p->current_lsn = cutoff_lsn + 1; + p->last_crc = new_last_crc; + atomic_store_explicit(&p->committed_lsn, cutoff_lsn, memory_order_release); + + pthread_mutex_unlock(&p->lock); + return 0; +} diff --git a/src/include/persist.h b/src/include/persist.h index d62dc5af..2efadaa5 100644 --- a/src/include/persist.h +++ b/src/include/persist.h @@ -68,6 +68,7 @@ struct persist_s { uint64_t term; uint32_t seg_offset; uint64_t checkpoint_lsn; + _Atomic uint64_t committed_lsn; /* highest LSN known committed; invariant: <= current_lsn - 1 */ uint64_t total_wal_bytes; char seg_path[4096]; @@ -121,6 +122,10 @@ int persist_flush(persist_t* p); /* Checkpoint */ int persist_checkpoint(persist_t* p); +/* WAL truncation — remove all entries with LSN > cutoff_lsn. + * Idempotent; returns 0 on success, negative errno on I/O error. */ +int persist_truncate_wal_above(persist_t* p, uint64_t cutoff_lsn); + /* Logging helper — never logs payload content (security invariant) */ void persist_log_entry_metadata(const wal_entry_hdr_t* hdr);