1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
// SPDX-License-Identifier: Apache-2.0
//! Write-ahead log: append-only durability for records not yet flushed to an SSTable.
//! `UACE-FR-1.1` — a record `fdatasync`'d here survives an unclean process restart.
use std::path::{Path, PathBuf};
use glommio::io::{BufferedFile, OpenOptions};
use quatzal_schema::{UaceError, UaceResult};
use crate::framing::{for_each_framed, frame};
use crate::pax::PaxRecord;
pub struct Wal {
file: BufferedFile,
offset: u64,
path: PathBuf,
}
impl Wal {
/// Open (creating if absent) the WAL file at `path`, positioned for appending after
/// whatever is already in it.
///
/// `UACE-FR-1.11`: opened read+write via `OpenOptions`, never `BufferedFile::open` --
/// glommio's `open` is **read-only**, and using it here meant every reopen over an
/// existing WAL produced a writer whose first append failed with `EBADF` (write on a
/// read-only fd). The engine literally could not accept writes after restarting onto
/// existing data; five waves of durability tests missed it because they all read after
/// reopen and never wrote. No `truncate`: the existing tail is the durability record.
pub async fn open_for_append(path: &Path) -> UaceResult<Self> {
let file = OpenOptions::new()
.read(true)
.write(true)
.create(true)
.buffered_open(path)
.await
.map_err(|e| UaceError::Io(format!("wal open {path:?}: {e}")))?;
let offset = file
.file_size()
.await
.map_err(|e| UaceError::Io(format!("wal size {path:?}: {e}")))?;
Ok(Wal {
file,
offset,
path: path.to_path_buf(),
})
}
/// Append one record and fsync before returning — this fsync is what makes
/// `UACE-FR-1.1`'s durability guarantee true.
pub async fn append(&mut self, record: &PaxRecord) -> UaceResult<()> {
let bytes = record.to_bytes();
let framed = frame(&bytes);
let len = framed.len() as u64;
self.file
.write_at(framed, self.offset)
.await
.map_err(|e| UaceError::Io(format!("wal write {:?}: {e}", self.path)))?;
self.file
.fdatasync()
.await
.map_err(|e| UaceError::Io(format!("wal fsync {:?}: {e}", self.path)))?;
self.offset += len;
Ok(())
}
/// `UACE-FR-1.9`: append many records with **one** write and **one** `fdatasync` --
/// group commit, the fix `CLAUDE.md` has named as deferred since Phase 1. It became
/// the binding constraint on the product surface once the vector tier stopped being
/// the bottleneck: `UACE-NFR-33` measured ~540 rows/sec end-to-end and traced it here,
/// where per-row `append` pays ~1.85ms of fsync per row.
///
/// Durability is all-or-nothing at the batch boundary in the sense that matters: the
/// caller is told "durable" only after the single fsync covering every record. A crash
/// mid-write leaves a torn tail, which replay rejects exactly as it does for a torn
/// single append (`Documents/RECOVERY.md`).
pub async fn append_batch(&mut self, records: &[PaxRecord]) -> UaceResult<()> {
if records.is_empty() {
return Ok(());
}
let mut framed = Vec::new();
for record in records {
framed.extend_from_slice(&frame(&record.to_bytes()));
}
let len = framed.len() as u64;
self.file
.write_at(framed, self.offset)
.await
.map_err(|e| UaceError::Io(format!("wal batch write {:?}: {e}", self.path)))?;
self.file
.fdatasync()
.await
.map_err(|e| UaceError::Io(format!("wal batch fsync {:?}: {e}", self.path)))?;
self.offset += len;
Ok(())
}
/// Truncate the WAL back to empty. Called right after a successful MemTable flush,
/// since everything the WAL held is now durable in an SSTable instead.
pub async fn reset(path: &Path) -> UaceResult<Self> {
let file = BufferedFile::create(path)
.await
.map_err(|e| UaceError::Io(format!("wal reset {path:?}: {e}")))?;
Ok(Wal {
file,
offset: 0,
path: path.to_path_buf(),
})
}
/// Replay every record currently in the WAL file at `path`, in write order. Returns an
/// empty vec if the file doesn't exist yet (fresh shard).
pub async fn replay(path: &Path) -> UaceResult<Vec<PaxRecord>> {
if !path.exists() {
return Ok(Vec::new());
}
let file = BufferedFile::open(path)
.await
.map_err(|e| UaceError::Io(format!("wal replay open {path:?}: {e}")))?;
let size = file
.file_size()
.await
.map_err(|e| UaceError::Io(format!("wal replay size {path:?}: {e}")))?;
let result = file
.read_at(0, size as usize)
.await
.map_err(|e| UaceError::Io(format!("wal replay read {path:?}: {e}")))?;
let bytes: &[u8] = &result;
let mut records = Vec::new();
for_each_framed(bytes, |payload| {
let (record, used) = PaxRecord::from_bytes(payload)?;
if used != payload.len() {
return Err(UaceError::Codec("wal record framing mismatch".into()));
}
records.push(record);
Ok(())
})?;
// Explicit close, matching the sstable convention -- glommio files should not be
// torn down by drop.
file.close()
.await
.map_err(|e| UaceError::Io(format!("wal replay close {path:?}: {e}")))?;
Ok(records)
}
}