Skip to main content

reddb_server/storage/wal/
writer.rs

1use super::record::WalRecord;
2use reddb_file::{OpenMode, StdVfs, Vfs, VfsFile};
3use std::io::{self, SeekFrom};
4use std::path::Path;
5use std::sync::Arc;
6
7/// User-space buffer size for the WAL writer.
8///
9/// Chosen so that ~5 000 small records (Begin/Commit ≈ 21 bytes,
10/// small PageWrite ≈ 34 bytes) coalesce into a single `write` syscall
11/// before the next `sync()` drains the buffer. Tunable; reflects the
12/// postgres XLOG block size (8 KiB) scaled up because we batch
13/// record-level rather than page-level.
14pub const WAL_BUFFER_BYTES: usize = 64 * 1024;
15
16#[derive(Debug, Clone, Copy, PartialEq, Eq)]
17enum WalSyncMethod {
18    Data,
19    All,
20}
21
22pub(crate) struct WalGroupSync<F: VfsFile> {
23    target_lsn: u64,
24    sync_handle: Arc<F>,
25    method: WalSyncMethod,
26}
27
28impl<F: VfsFile> WalGroupSync<F> {
29    pub(crate) fn target_lsn(&self) -> u64 {
30        self.target_lsn
31    }
32
33    pub(crate) fn sync(&self) -> io::Result<()> {
34        match self.method {
35            WalSyncMethod::Data => self.sync_handle.sync_data(),
36            WalSyncMethod::All => self.sync_handle.sync_all(),
37        }
38    }
39}
40
41struct VfsBufWriter<F: VfsFile> {
42    file: F,
43    buffer: Vec<u8>,
44}
45
46impl<F: VfsFile> VfsBufWriter<F> {
47    fn with_capacity(capacity: usize, file: F) -> Self {
48        Self {
49            file,
50            buffer: Vec::with_capacity(capacity),
51        }
52    }
53
54    fn write_all(&mut self, bytes: &[u8]) -> io::Result<()> {
55        if self.buffer.len() + bytes.len() > self.buffer.capacity() {
56            self.flush()?;
57        }
58        if bytes.len() >= self.buffer.capacity() {
59            self.file.write_all(bytes)
60        } else {
61            self.buffer.extend_from_slice(bytes);
62            Ok(())
63        }
64    }
65
66    fn flush(&mut self) -> io::Result<()> {
67        if !self.buffer.is_empty() {
68            self.file.write_all(&self.buffer)?;
69            self.buffer.clear();
70        }
71        Ok(())
72    }
73
74    fn get_ref(&self) -> &F {
75        &self.file
76    }
77
78    fn get_mut(&mut self) -> &mut F {
79        &mut self.file
80    }
81}
82
83impl<F: VfsFile> Drop for VfsBufWriter<F> {
84    fn drop(&mut self) {
85        let _ = self.flush();
86    }
87}
88
89/// Whether a `fallocate` failure means "this filesystem can't preallocate"
90/// (tmpfs, overlayfs, many network filesystems) rather than a real I/O error.
91/// Those are soft failures that flip the feature off; anything else is left to
92/// the normal write path to surface (e.g. a genuine `ENOSPC`).
93fn fallocate_unsupported(err: &io::Error) -> bool {
94    if err.kind() == io::ErrorKind::Unsupported {
95        return true;
96    }
97    #[cfg(target_os = "linux")]
98    {
99        matches!(
100            err.raw_os_error(),
101            Some(libc::EOPNOTSUPP) | Some(libc::ENOSYS) | Some(libc::EINVAL)
102        )
103    }
104    #[cfg(not(target_os = "linux"))]
105    {
106        false
107    }
108}
109
110/// Writer for the Write-Ahead Log
111///
112/// Wraps the underlying file in a [`VfsBufWriter`] so each `append` does
113/// not pay a write syscall — bytes accumulate in a 64 KiB user-space
114/// buffer until `sync()` (or `flush_until()`) drains them and then
115/// calls `sync_data()`/`sync_all()` on the raw file. This is how postgres turns
116/// per-record append cost from ~500 ns down to ~5 ns; reddb's previous
117/// per-append `write_all` directly to the file paid the syscall on
118/// every record.
119///
120/// **Critical contract:** every code path that syncs the underlying
121/// file *must* drain the [`VfsBufWriter`] first via
122/// `VfsBufWriter::flush()`. Otherwise the bytes in user-space never reach
123/// the kernel before fsync, and durability is silently broken.
124pub struct WalWriter<V: Vfs = StdVfs> {
125    file: VfsBufWriter<V::File>,
126    /// Cloned file descriptor for `sync_all()` outside the writer
127    /// mutex. Both this and `file`'s inner `File` point at the same
128    /// kernel inode; calling `sync_all()` on either flushes ALL
129    /// pending bytes for that inode. This is the trick that lets
130    /// the group-commit leader release the WAL writer lock during
131    /// the expensive fsync — see [`WalWriter::drain_for_group_sync`].
132    ///
133    /// Without this clone, a leader holding the writer mutex during
134    /// `sync_all()` blocks every other writer from appending,
135    /// defeating the entire purpose of group commit.
136    sync_handle: Arc<V::File>,
137    /// Log Sequence Number — byte offset of the next record. Advances
138    /// every `append`; survives across restarts via `seek(End)`.
139    current_lsn: u64,
140    /// Highest LSN that has been `sync_all()`'d to disk. The WAL-first
141    /// flush invariant relies on this: a page with `header.lsn = L` may
142    /// only be written to its data file once `durable_lsn >= L`.
143    /// See `src/storage/cache/README.md` § Invariant 2 and the Target 3
144    /// section of `PLAN.md`.
145    durable_lsn: u64,
146    /// WAL byte frontier covered by the last full file sync. Appends that stay
147    /// inside this synced preallocation range can use `sync_data()`; crossing
148    /// it, or syncing after fresh preallocation metadata, falls back to
149    /// `sync_all()`.
150    last_synced_size: u64,
151    /// Exclusive byte offset up to which disk blocks are pre-reserved via
152    /// `fallocate(FALLOC_FL_KEEP_SIZE)`. Advances one
153    /// [`reddb_file::MAIN_WAL_SEGMENT_BYTES`]
154    /// segment at a time as `current_lsn` approaches it (issue #893). Reset to
155    /// `0` on [`truncate`](Self::truncate) — which frees the blocks — and
156    /// immediately re-extended (the checkpoint re-extend path).
157    preallocated_to: u64,
158    /// Cleared the first time `fallocate` reports the backing filesystem can't
159    /// preallocate (tmpfs/overlay/NFS → `EOPNOTSUPP`/`ENOSYS`, or any non-Linux
160    /// target) so we stop issuing syscalls that will always fail. Preallocation
161    /// is a best-effort optimization; clearing this never affects correctness.
162    prealloc_supported: bool,
163    /// Set when `fallocate(FALLOC_FL_KEEP_SIZE)` successfully reserved a new
164    /// range and that allocation metadata has not yet been covered by a full
165    /// sync.
166    prealloc_metadata_dirty: bool,
167    #[cfg(test)]
168    last_sync_method: Option<WalSyncMethod>,
169}
170
171impl WalWriter<StdVfs> {
172    /// Open a WAL file for writing. Creates it if it doesn't exist.
173    pub fn open<P: AsRef<Path>>(path: P) -> io::Result<Self> {
174        Self::open_with_vfs(StdVfs, path)
175    }
176}
177
178impl<V: Vfs> WalWriter<V> {
179    /// Open a WAL file through `vfs`. Creates it if it doesn't exist.
180    pub fn open_with_vfs<P: AsRef<Path>>(vfs: V, path: P) -> io::Result<Self> {
181        // We do all initial bookkeeping (write header, seek to EOF) on
182        // the raw VFS file BEFORE wrapping in a VfsBufWriter so we don't
183        // have to worry about flush ordering during construction.
184        let mut raw = vfs.open(path.as_ref(), OpenMode::create_keep())?;
185        let existing_len = raw.file_len()?;
186
187        let current_lsn = if existing_len == 0 {
188            raw.write_all(&reddb_file::encode_wal_file_header())?;
189            raw.sync_all()?;
190            reddb_file::WAL_FILE_HEADER_BYTES as u64
191        } else {
192            // Existing file, set LSN to current end. Append-mode files
193            // ignore this seek for *writes*, but we use the returned
194            // position as our LSN counter.
195            raw.seek(SeekFrom::End(0))?
196        };
197
198        // Clone the file handle BEFORE wrapping in BufWriter. The
199        // clone shares the same kernel file description, so
200        // sync_all() on either descriptor flushes the whole inode.
201        // The VfsBufWriter owns the original; the Arc handle is shared
202        // with the group-commit leader.
203        let sync_handle = Arc::new(raw.try_clone()?);
204        let file = VfsBufWriter::with_capacity(WAL_BUFFER_BYTES, raw);
205
206        // On open, every byte already on disk is by definition durable
207        // (any pre-crash unflushed tail was lost when the OS dropped
208        // page cache). Initialise `durable_lsn` to `current_lsn`.
209        let mut writer = Self {
210            file,
211            sync_handle,
212            current_lsn,
213            durable_lsn: current_lsn,
214            last_synced_size: current_lsn,
215            preallocated_to: 0,
216            prealloc_supported: true,
217            prealloc_metadata_dirty: false,
218            #[cfg(test)]
219            last_sync_method: None,
220        };
221        // Reserve the first segment up front so the very first appends land in
222        // contiguous extents rather than growing the file page-by-page.
223        writer.ensure_preallocated()?;
224        Ok(writer)
225    }
226
227    /// Ensure disk blocks are reserved at least up to the next segment
228    /// boundary above the current write frontier (`current_lsn`).
229    ///
230    /// Cheap (pure arithmetic) until the frontier crosses a
231    /// [`reddb_file::MAIN_WAL_SEGMENT_BYTES`] boundary, at which point it issues a single
232    /// `fallocate`. Best-effort: a filesystem that can't preallocate disables
233    /// the feature; a transient error is swallowed so a write never fails
234    /// because preallocation hiccuped (the write path surfaces a genuine
235    /// `ENOSPC` on its own). Never grows the file's logical length, so it is
236    /// invisible to crash recovery.
237    fn ensure_preallocated(&mut self) -> io::Result<()> {
238        if !self.prealloc_supported {
239            return Ok(());
240        }
241        let target = reddb_file::next_main_wal_segment_boundary(self.current_lsn);
242        if target <= self.preallocated_to {
243            return Ok(());
244        }
245        let from = self.preallocated_to;
246        match self.file.get_ref().reserve(from, target - from) {
247            Ok(()) => {
248                self.preallocated_to = target;
249                self.prealloc_metadata_dirty = true;
250            }
251            Err(ref e) if fallocate_unsupported(e) => self.prealloc_supported = false,
252            Err(_) => {
253                // Best-effort: leave `preallocated_to` as-is and retry at the
254                // next boundary. Never propagate.
255            }
256        }
257        Ok(())
258    }
259
260    /// Append a record to the WAL.
261    ///
262    /// Bytes go into the BufWriter — they are NOT durable on disk
263    /// after this call returns. Callers that need durability must
264    /// follow up with [`WalWriter::sync`] or
265    /// [`WalWriter::flush_until`].
266    ///
267    /// Returns the LSN (Log Sequence Number) of the record.
268    pub fn append(&mut self, record: &WalRecord) -> io::Result<u64> {
269        let bytes = record.encode();
270        self.file.write_all(&bytes)?;
271
272        let record_lsn = self.current_lsn;
273        self.current_lsn += bytes.len() as u64;
274
275        self.ensure_preallocated()?;
276        Ok(record_lsn)
277    }
278
279    /// Write already-encoded bytes and advance the LSN counter to
280    /// match. Used by the lock-free append path: writers encode +
281    /// atomically reserve an LSN range outside this writer, the
282    /// group-commit coordinator drains the pending queue in LSN
283    /// order, then calls `append_bytes` for each batch.
284    ///
285    /// The bytes MUST be a valid `WalRecord::encode()` payload (or a
286    /// concatenation of such) — no structural validation happens
287    /// here. The caller is responsible for keeping the on-disk
288    /// byte offset synchronised with the externally-tracked LSN
289    /// counter; this method just appends and advances.
290    pub fn append_bytes(&mut self, bytes: &[u8]) -> io::Result<u64> {
291        self.file.write_all(bytes)?;
292        let record_lsn = self.current_lsn;
293        self.current_lsn += bytes.len() as u64;
294        self.ensure_preallocated()?;
295        Ok(record_lsn)
296    }
297
298    /// Rewind the writer's LSN counter to a specific value. Used
299    /// by the lock-free append path to resync the writer with the
300    /// externally-tracked `next_lsn` after a drain batch; the
301    /// coordinator knows the exact byte offset it just wrote to
302    /// and needs `current_lsn` to match so subsequent direct
303    /// callers of `append` stay consistent.
304    pub fn set_current_lsn(&mut self, lsn: u64) {
305        self.current_lsn = lsn;
306    }
307
308    /// Force sync to disk.
309    ///
310    /// Drains the user-space [`BufWriter`] first, then calls
311    /// `sync_all()` on the underlying file so every byte appended
312    /// since the last sync is durable. Updates `durable_lsn` so
313    /// subsequent `flush_until` calls become no-ops up to
314    /// `current_lsn`.
315    pub fn sync(&mut self) -> io::Result<()> {
316        self.file.flush()?;
317        self.sync_flushed_file()?;
318        self.durable_lsn = self.current_lsn;
319        Ok(())
320    }
321
322    /// Ensure the WAL is durable on disk at least up to byte offset
323    /// `target`. No-op when `target <= durable_lsn`.
324    ///
325    /// This is the postgres `XLogFlush(LSN)` analogue. Pager flush
326    /// paths call this with `max(dirty.header.lsn)` before writing
327    /// any data page so the WAL record describing the change is
328    /// guaranteed to be on disk before the page itself.
329    pub fn flush_until(&mut self, target: u64) -> io::Result<()> {
330        if self.durable_lsn >= target {
331            return Ok(());
332        }
333        self.file.flush()?;
334        self.sync_flushed_file()?;
335        self.durable_lsn = self.current_lsn;
336        Ok(())
337    }
338
339    fn sync_flushed_file(&mut self) -> io::Result<()> {
340        let method = self.next_sync_method();
341        match method {
342            WalSyncMethod::Data => self.file.get_ref().sync_data()?,
343            WalSyncMethod::All => self.file.get_ref().sync_all()?,
344        }
345        self.mark_sync_complete(method, self.current_lsn);
346        Ok(())
347    }
348
349    fn next_sync_method(&self) -> WalSyncMethod {
350        if !self.prealloc_metadata_dirty && self.current_lsn <= self.last_synced_size {
351            WalSyncMethod::Data
352        } else {
353            WalSyncMethod::All
354        }
355    }
356
357    fn mark_sync_complete(&mut self, method: WalSyncMethod, lsn: u64) {
358        match method {
359            WalSyncMethod::Data => {}
360            WalSyncMethod::All => {
361                self.last_synced_size = self.preallocated_to.max(lsn);
362                self.prealloc_metadata_dirty = false;
363            }
364        }
365        #[cfg(test)]
366        {
367            self.last_sync_method = Some(method);
368        }
369    }
370
371    /// Highest byte offset that is durable on disk. Used by the pager
372    /// to decide whether a `flush_until` call would actually need a
373    /// `fsync`.
374    pub fn durable_lsn(&self) -> u64 {
375        self.durable_lsn
376    }
377
378    /// Get current LSN (end of file offset)
379    pub fn current_lsn(&self) -> u64 {
380        self.current_lsn
381    }
382
383    /// Drain the BufWriter into the kernel and return the captured
384    /// LSN plus a cloned file handle and sync method for the caller
385    /// **without holding the WAL writer mutex**.
386    ///
387    /// Used by the group-commit leader path. The flow is:
388    ///
389    /// 1. Take the WAL writer mutex.
390    /// 2. Call this method — drains user-space buffer to the kernel
391    ///    and captures a size-aware sync plan.
392    /// 3. Release the WAL writer mutex.
393    /// 4. Execute the sync plan — this is the expensive ~100 µs syscall,
394    ///    and other writers can keep appending while it runs.
395    /// 5. Take the WAL writer mutex briefly and call
396    ///    [`WalWriter::mark_durable`] to publish the new durable position.
397    ///
398    /// The cloned `sync_handle` shares the same kernel inode with
399    /// the writer's `file`, so syncing the clone flushes bytes that
400    /// have reached the kernel for that file.
401    /// This is the coalescing window that makes group commit win.
402    pub(crate) fn drain_for_group_sync(&mut self) -> io::Result<WalGroupSync<V::File>> {
403        // Drain user-space buffer into the kernel.
404        self.file.flush()?;
405        Ok(WalGroupSync {
406            target_lsn: self.current_lsn,
407            sync_handle: Arc::clone(&self.sync_handle),
408            method: self.next_sync_method(),
409        })
410    }
411
412    /// Manually advance `durable_lsn` after a successful out-of-lock
413    /// sync performed via [`WalWriter::drain_for_group_sync`].
414    ///
415    /// Monotonic — never lowers `durable_lsn`. Safe to call with a
416    /// stale `lsn`; just becomes a no-op.
417    pub(crate) fn mark_durable(&mut self, sync: &WalGroupSync<V::File>) {
418        let lsn = sync.target_lsn;
419        if lsn > self.durable_lsn {
420            self.durable_lsn = lsn;
421        }
422        self.mark_sync_complete(sync.method, lsn);
423    }
424
425    /// Truncate the WAL (usually after checkpoint).
426    ///
427    /// Drains the BufWriter first so no pending bytes hit the file
428    /// after the truncate. Then resets the underlying file, rewrites
429    /// the header through the buffered writer (header is small; the
430    /// followup `flush + sync_all` makes it durable), and resets
431    /// LSN bookkeeping.
432    pub fn truncate(&mut self) -> io::Result<()> {
433        // Drop any pending bytes BEFORE the truncate; otherwise the
434        // BufWriter would flush them to a re-shrunken file in
435        // confused order.
436        self.file.flush()?;
437
438        {
439            let raw = self.file.get_mut();
440            raw.set_len(0)?;
441            raw.seek(SeekFrom::Start(0))?;
442        }
443
444        // Rewrite header through the BufWriter then drain.
445        self.file.write_all(&reddb_file::encode_wal_file_header())?;
446        self.file.flush()?;
447        self.file.get_ref().sync_all()?;
448
449        self.current_lsn = reddb_file::WAL_FILE_HEADER_BYTES as u64;
450        self.durable_lsn = reddb_file::WAL_FILE_HEADER_BYTES as u64;
451        self.last_synced_size = reddb_file::WAL_FILE_HEADER_BYTES as u64;
452        self.prealloc_metadata_dirty = false;
453        #[cfg(test)]
454        {
455            self.last_sync_method = Some(WalSyncMethod::All);
456        }
457
458        // `set_len(0)` freed every reserved block, so the WAL would otherwise
459        // grow page-by-page again from here. Re-extend a fresh segment now —
460        // this is the "truncate/re-extend on checkpoint" half of issue #893.
461        self.preallocated_to = 0;
462        self.ensure_preallocated()?;
463        Ok(())
464    }
465}
466
467#[cfg(test)]
468mod tests {
469    use super::*;
470    use std::path::PathBuf;
471
472    struct FileGuard {
473        path: PathBuf,
474    }
475
476    impl Drop for FileGuard {
477        fn drop(&mut self) {
478            let _ = std::fs::remove_file(&self.path);
479        }
480    }
481
482    fn temp_wal(name: &str) -> (FileGuard, PathBuf) {
483        let path = reddb_file::layout::wal_component_temp_path(
484            &std::env::temp_dir(),
485            "writer",
486            name,
487            std::process::id(),
488        );
489        let guard = FileGuard { path: path.clone() };
490        let _ = std::fs::remove_file(&path);
491        (guard, path)
492    }
493
494    #[test]
495    fn test_create_new_wal() {
496        let (_guard, path) = temp_wal("create");
497        let writer = WalWriter::open(&path).expect("open() should succeed");
498
499        // Should start at LSN 8 (after 8-byte header)
500        assert_eq!(writer.current_lsn(), 8);
501        assert!(path.exists());
502    }
503
504    #[test]
505    fn test_append_record() {
506        let (_guard, path) = temp_wal("append");
507        let mut writer = WalWriter::open(&path).expect("open() should succeed");
508
509        let record = WalRecord::Begin { tx_id: 42 };
510        let lsn = writer.append(&record).expect("append() should succeed");
511
512        // First record starts at LSN 8
513        assert_eq!(lsn, 8);
514
515        // Next record should start after encoded size
516        // Begin record (v3 framing): 1 (type) + 8 (term) + 8 (ownership_epoch)
517        //   + 8 (tx_id) + 4 (checksum) = 29 bytes
518        assert_eq!(writer.current_lsn(), 8 + 29);
519    }
520
521    #[test]
522    fn test_append_multiple_records() {
523        let (_guard, path) = temp_wal("multi");
524        let mut writer = WalWriter::open(&path).expect("open() should succeed");
525
526        let lsn1 = writer
527            .append(&WalRecord::Begin { tx_id: 1 })
528            .expect("append() should succeed");
529        let lsn2 = writer
530            .append(&WalRecord::Begin { tx_id: 2 })
531            .expect("append() should succeed");
532        let lsn3 = writer
533            .append(&WalRecord::Commit { tx_id: 1 })
534            .expect("append() should succeed");
535
536        assert_eq!(lsn1, 8);
537        assert_eq!(lsn2, 8 + 29);
538        assert_eq!(lsn3, 8 + 29 + 29);
539    }
540
541    #[test]
542    fn test_page_write_lsn() {
543        let (_guard, path) = temp_wal("pagewrite");
544        let mut writer = WalWriter::open(&path).expect("open() should succeed");
545
546        // First record
547        let lsn1 = writer
548            .append(&WalRecord::Begin { tx_id: 1 })
549            .expect("append() should succeed");
550        assert_eq!(lsn1, 8);
551
552        // PageWrite record (v3 framing): +8 bytes for ownership_epoch vs v2
553        let data = vec![1, 2, 3, 4, 5];
554        let lsn2 = writer
555            .append(&WalRecord::PageWrite {
556                tx_id: 1,
557                page_id: 100,
558                data: data.clone(),
559            })
560            .expect("value is present");
561
562        assert_eq!(lsn2, 8 + 29); // after Begin
563
564        // Next LSN = lsn2 + (1 + 8 + 8 + 8 + 4 + 4 + 5 + 4) = lsn2 + 42
565        assert_eq!(writer.current_lsn(), 8 + 29 + 42);
566    }
567
568    #[test]
569    fn test_sync() {
570        let (_guard, path) = temp_wal("sync");
571        let mut writer = WalWriter::open(&path).expect("open() should succeed");
572
573        writer
574            .append(&WalRecord::Begin { tx_id: 1 })
575            .expect("append() should succeed");
576        writer.sync().expect("sync() should succeed");
577
578        // File should be synced, just verify no error
579        assert!(path.exists());
580    }
581
582    #[test]
583    fn test_truncate() {
584        let (_guard, path) = temp_wal("truncate");
585        let mut writer = WalWriter::open(&path).expect("open() should succeed");
586
587        // Write some records
588        writer
589            .append(&WalRecord::Begin { tx_id: 1 })
590            .expect("append() should succeed");
591        writer
592            .append(&WalRecord::PageWrite {
593                tx_id: 1,
594                page_id: 0,
595                data: vec![0; 100],
596            })
597            .expect("value is present");
598        writer
599            .append(&WalRecord::Commit { tx_id: 1 })
600            .expect("append() should succeed");
601
602        let lsn_before = writer.current_lsn();
603        assert!(lsn_before > 8);
604
605        // Truncate
606        writer.truncate().expect("truncate() should succeed");
607
608        // LSN should be back to 8
609        assert_eq!(writer.current_lsn(), 8);
610
611        // File should be 8 bytes (just header)
612        let len = std::fs::metadata(&path)
613            .expect("metadata() should succeed")
614            .len();
615        assert_eq!(len, 8);
616    }
617
618    #[test]
619    fn test_reopen_existing() {
620        let (_guard, path) = temp_wal("reopen");
621
622        // Create and write
623        let lsn_after_write;
624        {
625            let mut writer = WalWriter::open(&path).expect("open() should succeed");
626            writer
627                .append(&WalRecord::Begin { tx_id: 1 })
628                .expect("append() should succeed");
629            writer
630                .append(&WalRecord::Commit { tx_id: 1 })
631                .expect("append() should succeed");
632            lsn_after_write = writer.current_lsn();
633        }
634
635        // Reopen
636        {
637            let writer = WalWriter::open(&path).expect("open() should succeed");
638            // Should continue from where we left off
639            assert_eq!(writer.current_lsn(), lsn_after_write);
640        }
641    }
642
643    #[test]
644    fn test_checkpoint_record() {
645        let (_guard, path) = temp_wal("checkpoint");
646        let mut writer = WalWriter::open(&path).expect("open() should succeed");
647
648        // Checkpoint is same size as Begin (v3: 1 + 8 + 8 + 8 + 4 = 29)
649        let lsn = writer
650            .append(&WalRecord::Checkpoint { lsn: 12345 })
651            .expect("value is present");
652        assert_eq!(lsn, 8);
653        assert_eq!(writer.current_lsn(), 8 + 29);
654    }
655
656    // -----------------------------------------------------------------
657    // Target 3: durable_lsn / flush_until tests
658    // -----------------------------------------------------------------
659
660    #[test]
661    fn fresh_wal_has_durable_lsn_at_header_end() {
662        let (_guard, path) = temp_wal("durable_init");
663        let writer = WalWriter::open(&path).expect("open() should succeed");
664        assert_eq!(writer.durable_lsn(), 8);
665        assert_eq!(writer.current_lsn(), 8);
666    }
667
668    #[test]
669    fn flush_until_below_durable_is_noop() {
670        let (_guard, path) = temp_wal("flush_noop");
671        let mut writer = WalWriter::open(&path).expect("open() should succeed");
672        // After open, durable_lsn == 8.
673        let before = writer.durable_lsn();
674        writer.flush_until(0).expect("flush_until() should succeed");
675        writer.flush_until(8).expect("flush_until() should succeed");
676        assert_eq!(writer.durable_lsn(), before);
677    }
678
679    #[test]
680    fn flush_until_advances_durable_to_current() {
681        let (_guard, path) = temp_wal("flush_advance");
682        let mut writer = WalWriter::open(&path).expect("open() should succeed");
683        writer
684            .append(&WalRecord::Begin { tx_id: 7 })
685            .expect("append() should succeed");
686        writer
687            .append(&WalRecord::Commit { tx_id: 7 })
688            .expect("append() should succeed");
689        let target = writer.current_lsn();
690        // Before flush_until, durable still at the header.
691        assert_eq!(writer.durable_lsn(), 8);
692        writer
693            .flush_until(target)
694            .expect("flush_until() should succeed");
695        assert_eq!(writer.durable_lsn(), target);
696    }
697
698    #[test]
699    fn flush_until_is_monotonic() {
700        let (_guard, path) = temp_wal("flush_monotonic");
701        let mut writer = WalWriter::open(&path).expect("open() should succeed");
702        writer
703            .append(&WalRecord::Begin { tx_id: 1 })
704            .expect("append() should succeed");
705        let lo = writer.current_lsn();
706        writer
707            .flush_until(lo)
708            .expect("flush_until() should succeed");
709        let durable_after_lo = writer.durable_lsn();
710        writer
711            .append(&WalRecord::Commit { tx_id: 1 })
712            .expect("append() should succeed");
713        let hi = writer.current_lsn();
714        writer
715            .flush_until(hi)
716            .expect("flush_until() should succeed");
717        assert!(writer.durable_lsn() >= durable_after_lo);
718        // Calling flush_until(lo) after flush_until(hi) is a no-op.
719        writer
720            .flush_until(lo)
721            .expect("flush_until() should succeed");
722        assert_eq!(writer.durable_lsn(), hi);
723    }
724
725    #[test]
726    fn sync_advances_durable_lsn_too() {
727        let (_guard, path) = temp_wal("sync_durable");
728        let mut writer = WalWriter::open(&path).expect("open() should succeed");
729        writer
730            .append(&WalRecord::Begin { tx_id: 9 })
731            .expect("append() should succeed");
732        let before = writer.durable_lsn();
733        let after_append = writer.current_lsn();
734        assert!(after_append > before);
735        writer.sync().expect("sync() should succeed");
736        assert_eq!(writer.durable_lsn(), after_append);
737    }
738
739    #[test]
740    fn sync_all_is_used_when_wal_size_grew() {
741        let (_guard, path) = temp_wal("sync_all_grew");
742        let mut writer = WalWriter::open(&path).expect("open() should succeed");
743
744        writer
745            .append(&WalRecord::Begin { tx_id: 1 })
746            .expect("append() should succeed");
747        writer.sync().expect("sync() should succeed");
748
749        assert_eq!(writer.last_sync_method, Some(WalSyncMethod::All));
750        assert!(writer.last_synced_size >= writer.current_lsn());
751        assert!(!writer.prealloc_metadata_dirty);
752    }
753
754    #[test]
755    fn sync_all_is_used_for_metadata_only_preallocation() {
756        let (_guard, path) = temp_wal("sync_all_prealloc_metadata");
757        let mut writer = WalWriter::open(&path).expect("open() should succeed");
758        if !writer.prealloc_supported {
759            return;
760        }
761
762        assert_eq!(writer.current_lsn(), 8);
763        assert!(writer.prealloc_metadata_dirty);
764
765        writer.sync().expect("sync() should succeed");
766
767        assert_eq!(writer.last_sync_method, Some(WalSyncMethod::All));
768        assert_eq!(writer.last_synced_size, writer.preallocated_to);
769        assert!(!writer.prealloc_metadata_dirty);
770    }
771
772    #[test]
773    fn sync_data_is_used_when_wal_size_is_unchanged() {
774        let (_guard, path) = temp_wal("sync_data_unchanged");
775        let mut writer = WalWriter::open(&path).expect("open() should succeed");
776
777        writer
778            .append(&WalRecord::Begin { tx_id: 1 })
779            .expect("append() should succeed");
780        writer.sync().expect("sync() should succeed");
781        let synced_size = writer.last_synced_size;
782        writer.sync().expect("sync() should succeed");
783
784        assert_eq!(writer.last_sync_method, Some(WalSyncMethod::Data));
785        assert_eq!(writer.last_synced_size, synced_size);
786        assert_eq!(writer.durable_lsn(), writer.current_lsn());
787    }
788
789    #[test]
790    fn sync_data_is_used_for_appends_within_synced_preallocation() {
791        let (_guard, path) = temp_wal("sync_data_preallocated_append");
792        let mut writer = WalWriter::open(&path).expect("open() should succeed");
793        if !writer.prealloc_supported {
794            return;
795        }
796
797        writer
798            .append(&WalRecord::Begin { tx_id: 1 })
799            .expect("append() should succeed");
800        writer.sync().expect("sync() should succeed");
801        assert_eq!(writer.last_sync_method, Some(WalSyncMethod::All));
802
803        writer
804            .append(&WalRecord::Commit { tx_id: 1 })
805            .expect("append() should succeed");
806        writer.sync().expect("sync() should succeed");
807
808        assert_eq!(writer.last_sync_method, Some(WalSyncMethod::Data));
809        assert_eq!(writer.durable_lsn(), writer.current_lsn());
810        assert!(writer.current_lsn() <= writer.last_synced_size);
811    }
812
813    #[test]
814    fn group_sync_uses_sync_data_within_synced_preallocation() {
815        let (_guard, path) = temp_wal("group_sync_data_preallocated_append");
816        let mut writer = WalWriter::open(&path).expect("open() should succeed");
817        if !writer.prealloc_supported {
818            return;
819        }
820
821        writer
822            .append(&WalRecord::Begin { tx_id: 1 })
823            .expect("append() should succeed");
824        writer.sync().expect("sync() should succeed");
825        assert_eq!(writer.last_sync_method, Some(WalSyncMethod::All));
826
827        writer
828            .append(&WalRecord::Commit { tx_id: 1 })
829            .expect("append() should succeed");
830        let sync = writer
831            .drain_for_group_sync()
832            .expect("drain_for_group_sync() should succeed");
833        assert_eq!(sync.method, WalSyncMethod::Data);
834        sync.sync().expect("sync() should succeed");
835        writer.mark_durable(&sync);
836
837        assert_eq!(writer.last_sync_method, Some(WalSyncMethod::Data));
838        assert_eq!(writer.durable_lsn(), writer.current_lsn());
839    }
840
841    #[test]
842    fn truncate_resets_durable_lsn() {
843        let (_guard, path) = temp_wal("truncate_durable");
844        let mut writer = WalWriter::open(&path).expect("open() should succeed");
845        writer
846            .append(&WalRecord::Begin { tx_id: 1 })
847            .expect("append() should succeed");
848        writer.sync().expect("sync() should succeed");
849        assert!(writer.durable_lsn() > 8);
850        writer.truncate().expect("truncate() should succeed");
851        assert_eq!(writer.durable_lsn(), 8);
852        assert_eq!(writer.current_lsn(), 8);
853    }
854
855    #[test]
856    fn reopen_initialises_durable_to_current() {
857        let (_guard, path) = temp_wal("reopen_durable");
858        {
859            let mut writer = WalWriter::open(&path).expect("open() should succeed");
860            writer
861                .append(&WalRecord::Begin { tx_id: 1 })
862                .expect("append() should succeed");
863            writer.sync().expect("sync() should succeed");
864        }
865        let writer = WalWriter::open(&path).expect("open() should succeed");
866        // After reopen, every byte on disk is durable by definition.
867        assert_eq!(writer.durable_lsn(), writer.current_lsn());
868    }
869
870    // -----------------------------------------------------------------
871    // Perf 1.1: BufWriter coalesces small appends until sync
872    // -----------------------------------------------------------------
873
874    #[test]
875    fn bufwriter_coalesces_until_sync() {
876        // Append 100 small records but DO NOT sync. The on-disk file
877        // size must still equal the header (8 bytes) because the
878        // bytes are sitting in the BufWriter, not in the kernel.
879        let (_guard, path) = temp_wal("bufwriter_coalesce");
880        let mut writer = WalWriter::open(&path).expect("open() should succeed");
881        for tx in 0..100u64 {
882            writer
883                .append(&WalRecord::Begin { tx_id: tx })
884                .expect("append() should succeed");
885        }
886        // current_lsn reflects the in-buffer position. Each v3 Begin record is
887        // 29 bytes: 1 (type) + 8 (term) + 8 (ownership_epoch) + 8 (tx_id)
888        //   + 4 (checksum).
889        assert_eq!(writer.current_lsn(), 8 + 100 * 29);
890        // But the file on disk only has the header.
891        let on_disk = std::fs::metadata(&path)
892            .expect("metadata() should succeed")
893            .len();
894        assert_eq!(on_disk, 8, "BufWriter leaked bytes to disk before sync");
895    }
896
897    #[test]
898    fn sync_drains_bufwriter_before_fsync() {
899        // After sync(), the file size must equal current_lsn — the
900        // BufWriter has been flushed and sync_all has hit the kernel.
901        let (_guard, path) = temp_wal("sync_drains");
902        let mut writer = WalWriter::open(&path).expect("open() should succeed");
903        for tx in 0..50u64 {
904            writer
905                .append(&WalRecord::Begin { tx_id: tx })
906                .expect("append() should succeed");
907        }
908        writer.sync().expect("sync() should succeed");
909        let on_disk = std::fs::metadata(&path)
910            .expect("metadata() should succeed")
911            .len();
912        assert_eq!(on_disk, writer.current_lsn());
913        assert_eq!(writer.durable_lsn(), writer.current_lsn());
914    }
915
916    #[test]
917    fn flush_until_drains_bufwriter_too() {
918        // flush_until must drain the BufWriter before calling
919        // sync_all on the underlying file — otherwise pending bytes
920        // never become durable.
921        let (_guard, path) = temp_wal("flush_until_drains");
922        let mut writer = WalWriter::open(&path).expect("open() should succeed");
923        for tx in 0..30u64 {
924            writer
925                .append(&WalRecord::Begin { tx_id: tx })
926                .expect("append() should succeed");
927        }
928        let target = writer.current_lsn();
929        writer
930            .flush_until(target)
931            .expect("flush_until() should succeed");
932        let on_disk = std::fs::metadata(&path)
933            .expect("metadata() should succeed")
934            .len();
935        assert_eq!(on_disk, target);
936        assert_eq!(writer.durable_lsn(), target);
937    }
938
939    #[test]
940    fn truncate_drains_pending_bufwriter_bytes_first() {
941        // If truncate did NOT drain BufWriter first, the pending bytes
942        // would either land in the post-truncate file (corrupting it
943        // with stale records) or be lost. Verify the resulting file
944        // contains only a fresh header.
945        let (_guard, path) = temp_wal("truncate_drain");
946        let mut writer = WalWriter::open(&path).expect("open() should succeed");
947        // Write enough small records to fill some of the 64 KiB buffer
948        // but stay below the auto-flush threshold.
949        for tx in 0..200u64 {
950            writer
951                .append(&WalRecord::Begin { tx_id: tx })
952                .expect("append() should succeed");
953        }
954        // Sanity: bytes are buffered.
955        assert_eq!(
956            std::fs::metadata(&path)
957                .expect("metadata() should succeed")
958                .len(),
959            8
960        );
961
962        writer.truncate().expect("truncate() should succeed");
963        // After truncate the file is just the header again.
964        let on_disk = std::fs::metadata(&path)
965            .expect("metadata() should succeed")
966            .len();
967        assert_eq!(on_disk, 8);
968        assert_eq!(writer.current_lsn(), 8);
969        assert_eq!(writer.durable_lsn(), 8);
970
971        // And we can append again successfully.
972        writer
973            .append(&WalRecord::Begin { tx_id: 99 })
974            .expect("append() should succeed");
975        writer.sync().expect("sync() should succeed");
976        assert_eq!(
977            std::fs::metadata(&path)
978                .expect("metadata() should succeed")
979                .len(),
980            8 + 29
981        );
982    }
983
984    #[test]
985    fn reopen_sees_only_synced_records() {
986        // Records that were appended but never sync'd must NOT
987        // survive a reopen — they lived in the BufWriter, never made
988        // it to the kernel, and the previous WalWriter went out of
989        // scope. The new WalWriter reopens the file and reads from
990        // EOF, which reflects only the bytes that hit disk.
991        //
992        // We sync some records, then drop the writer mid-buffer, and
993        // assert the reopen LSN matches only the synced prefix.
994        let (_guard, path) = temp_wal("reopen_synced_only");
995        let synced_lsn;
996        {
997            let mut writer = WalWriter::open(&path).expect("open() should succeed");
998            writer
999                .append(&WalRecord::Begin { tx_id: 1 })
1000                .expect("append() should succeed");
1001            writer.sync().expect("sync() should succeed");
1002            synced_lsn = writer.current_lsn();
1003            // These records are never sync'd before drop. Drop runs
1004            // BufWriter::flush which DOES write them — see note below.
1005            for tx in 100..120u64 {
1006                writer
1007                    .append(&WalRecord::Begin { tx_id: tx })
1008                    .expect("append() should succeed");
1009            }
1010            // Without a sync, the in-buffer bytes are still pending.
1011            // BufWriter's Drop impl does flush to the file but does
1012            // not call sync_all. For reopen-LSN purposes, on-disk
1013            // bytes count regardless of fsync, so the reopened LSN
1014            // will reflect the dropped writes too.
1015        }
1016        let writer = WalWriter::open(&path).expect("open() should succeed");
1017        // The reopen LSN reflects what's physically on disk after
1018        // BufWriter::Drop flushes its buffer. That may or may not
1019        // include the unsync'd records depending on platform; the
1020        // contract we care about is that durable_lsn ≥ synced_lsn.
1021        assert!(writer.durable_lsn() >= synced_lsn);
1022    }
1023
1024    // -----------------------------------------------------------------
1025    // Issue #893: fallocate-based WAL segment preallocation
1026    // -----------------------------------------------------------------
1027
1028    /// On-disk blocks reserved by `fallocate`, in bytes. Returns the
1029    /// allocated size (st_blocks × 512), independent of the logical length.
1030    fn allocated_bytes(path: &std::path::Path) -> u64 {
1031        use fs2::FileExt;
1032        let f = std::fs::File::open(path).expect("open() should succeed");
1033        f.allocated_size().expect("allocated_size() should succeed")
1034    }
1035
1036    #[test]
1037    fn segment_boundary_rounds_strictly_up() {
1038        // Always lands one boundary ahead so the reservation stays in front
1039        // of the write frontier.
1040        assert_eq!(
1041            reddb_file::next_main_wal_segment_boundary(0),
1042            reddb_file::MAIN_WAL_SEGMENT_BYTES
1043        );
1044        assert_eq!(
1045            reddb_file::next_main_wal_segment_boundary(8),
1046            reddb_file::MAIN_WAL_SEGMENT_BYTES
1047        );
1048        assert_eq!(
1049            reddb_file::next_main_wal_segment_boundary(reddb_file::MAIN_WAL_SEGMENT_BYTES - 1),
1050            reddb_file::MAIN_WAL_SEGMENT_BYTES
1051        );
1052        // Exactly on a boundary still advances to the next one.
1053        assert_eq!(
1054            reddb_file::next_main_wal_segment_boundary(reddb_file::MAIN_WAL_SEGMENT_BYTES),
1055            2 * reddb_file::MAIN_WAL_SEGMENT_BYTES
1056        );
1057        assert_eq!(
1058            reddb_file::next_main_wal_segment_boundary(reddb_file::MAIN_WAL_SEGMENT_BYTES + 1),
1059            2 * reddb_file::MAIN_WAL_SEGMENT_BYTES
1060        );
1061    }
1062
1063    #[test]
1064    fn open_preallocates_first_segment() {
1065        // A freshly opened WAL must reserve a whole segment up front instead
1066        // of growing incrementally (acceptance #1).
1067        let (_guard, path) = temp_wal("prealloc_open");
1068        let writer = WalWriter::open(&path).expect("open() should succeed");
1069        if !writer.prealloc_supported {
1070            return; // filesystem without fallocate — feature is a no-op.
1071        }
1072        assert_eq!(writer.preallocated_to, reddb_file::MAIN_WAL_SEGMENT_BYTES);
1073        // The reservation is real on disk, yet the logical file is still just
1074        // the 8-byte header.
1075        assert!(allocated_bytes(&path) >= reddb_file::MAIN_WAL_SEGMENT_BYTES);
1076        assert_eq!(
1077            std::fs::metadata(&path)
1078                .expect("metadata() should succeed")
1079                .len(),
1080            8
1081        );
1082    }
1083
1084    #[test]
1085    fn preallocation_does_not_grow_logical_length() {
1086        // The load-bearing invariant for crash recovery: appending records
1087        // must NOT inflate the logical file size beyond the real data, or the
1088        // EOF scan in WalReader would walk into the reserved tail. Holds on
1089        // every filesystem (fallocate keeps i_size pinned; absent fallocate
1090        // there is no reservation at all).
1091        let (_guard, path) = temp_wal("prealloc_logical");
1092        let mut writer = WalWriter::open(&path).expect("open() should succeed");
1093        for tx in 0..50u64 {
1094            writer
1095                .append(&WalRecord::Begin { tx_id: tx })
1096                .expect("append() should succeed");
1097        }
1098        writer.sync().expect("sync() should succeed");
1099        let logical = std::fs::metadata(&path)
1100            .expect("metadata() should succeed")
1101            .len();
1102        assert_eq!(logical, 8 + 50 * 29, "preallocation inflated i_size");
1103        assert_eq!(writer.current_lsn(), logical);
1104    }
1105
1106    #[test]
1107    fn truncate_re_extends_a_fresh_segment() {
1108        // After checkpoint truncation the WAL must re-extend rather than grow
1109        // unbounded page-by-page (acceptance #2).
1110        let (_guard, path) = temp_wal("prealloc_truncate");
1111        let mut writer = WalWriter::open(&path).expect("open() should succeed");
1112        writer
1113            .append(&WalRecord::Begin { tx_id: 1 })
1114            .expect("append() should succeed");
1115        writer.sync().expect("sync() should succeed");
1116
1117        writer.truncate().expect("truncate() should succeed");
1118
1119        assert_eq!(writer.current_lsn(), 8);
1120        assert_eq!(
1121            std::fs::metadata(&path)
1122                .expect("metadata() should succeed")
1123                .len(),
1124            8
1125        );
1126        if writer.prealloc_supported {
1127            assert_eq!(writer.preallocated_to, reddb_file::MAIN_WAL_SEGMENT_BYTES);
1128            assert!(allocated_bytes(&path) >= reddb_file::MAIN_WAL_SEGMENT_BYTES);
1129        }
1130    }
1131
1132    #[test]
1133    fn preallocated_wal_recovers_records_without_trailing_garbage() {
1134        // End-to-end: a preallocated WAL must read back exactly the records
1135        // written — the reserved (unwritten) tail must be invisible to the
1136        // reader, proving crash-recovery is unchanged (acceptance #3).
1137        use super::super::reader::WalReader;
1138        let (_guard, path) = temp_wal("prealloc_recover");
1139        {
1140            let mut writer = WalWriter::open(&path).expect("open() should succeed");
1141            writer
1142                .append(&WalRecord::Begin { tx_id: 1 })
1143                .expect("append() should succeed");
1144            writer
1145                .append(&WalRecord::PageWrite {
1146                    tx_id: 1,
1147                    page_id: 7,
1148                    data: vec![1, 2, 3, 4],
1149                })
1150                .expect("value is present");
1151            writer
1152                .append(&WalRecord::Commit { tx_id: 1 })
1153                .expect("append() should succeed");
1154            writer.sync().expect("sync() should succeed");
1155        }
1156        let records: Vec<_> = WalReader::open(&path)
1157            .expect("value is present")
1158            .iter()
1159            .collect::<Result<_, _>>()
1160            .expect("reader must stop cleanly at real EOF, not in reserved tail");
1161        assert_eq!(records.len(), 3);
1162        assert_eq!(records[0].1, WalRecord::Begin { tx_id: 1 });
1163        assert_eq!(records[2].1, WalRecord::Commit { tx_id: 1 });
1164    }
1165}