feat(ha-config-bootstrap): implement persist_truncate_wal_above and committed_lsn (task 7.5)

This commit is contained in:
Rongsong Shen 2026-05-28 14:13:29 +00:00
parent e7da075315
commit 85c8565957
3 changed files with 151 additions and 1 deletions

View file

@ -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;
}