WIP - buffer and flush after block

Author: arminsabouri

src/crates/primitives/src/indecies.rs

@@ -1,5 +1,5 @@
 use std::fs::{File, OpenOptions};
-use std::io;
+use std::io::{self, BufWriter, Seek, SeekFrom, Write};
 use std::os::unix::fs::FileExt;
 use std::path::Path;
 
@@ -11,9 +11,8 @@ const TXPTR_LEN_BYTES: usize = 28;
 const BLOCK_TX_END_LEN_BYTES: usize = 4;
 const LINK_LEN_BYTES: usize = 8;
 
-/// Target size of the in-memory append buffer, in bytes. Chosen to amortize
-/// the per-write syscall cost without making tail reads (which must scan the
-/// buffer) expensive. Each index instance holds at most one such buffer.
+/// Capacity of the [`BufWriter`] wrapping each index file. Chosen to amortize
+/// the per-write syscall cost. Each index instance holds at most one such buffer.
 const APPEND_BUF_CAP_BYTES: usize = 64 * 1024;
 
 pub const OUTID_NONE: u64 = u64::MAX;
@@ -95,22 +94,13 @@ impl TxPtr {
 
 /// Fixed-width append-mostly log of `N`-byte records.
 ///
-/// Appends are staged in an in-memory buffer and flushed to disk in batches
-/// via positional writes (`pwrite`), avoiding the per-record `lseek + write`
-/// pair that dominates parser IO. Random-access `get`/`set` operations are
-/// served from the buffer when they target not-yet-flushed records, so callers
-/// that read the tail mid-build (e.g. the parser resolving prevouts against
-/// just-appended tx pointers) see a consistent view without explicit flushing.
+/// Appends are buffered via [`BufWriter`] and flushed to disk in batches.
+/// `get_bytes` serves buffered tail records directly from the writer's buffer.
+/// `set_bytes` flushes the buffer first, then uses a positional write so the
+/// file cursor is undisturbed for subsequent appends.
 #[derive(Debug)]
 struct FixedWidthIndex<const N: usize> {
-    /// Handle used for all positional IO. The file cursor is never read nor
-    /// modified; we rely on `pread`/`pwrite` via `FileExt`.
-    file: File,
-    /// Bytes appended since the last flush. Always a multiple of `N`.
-    buf: Vec<u8>,
-    /// Number of records currently persisted to disk.
-    flushed_len: u64,
-    /// Total number of records (persisted + buffered). `flushed_len <= len`.
+    writer: BufWriter<File>,
     len: u64,
     mmap: Option<Mmap>,
 }
@@ -124,16 +114,14 @@ impl<const N: usize> FixedWidthIndex<N> {
             .truncate(true)
             .open(path)?;
         Ok(Self {
-            file,
-            buf: Vec::with_capacity(APPEND_BUF_CAP_BYTES),
-            flushed_len: 0,
+            writer: BufWriter::with_capacity(APPEND_BUF_CAP_BYTES, file),
             len: 0,
             mmap: None,
         })
     }
 
     fn open(path: impl AsRef<Path>, len_error: &'static str) -> io::Result<Self> {
-        let file = OpenOptions::new().read(true).write(true).open(path)?;
+        let mut file = OpenOptions::new().read(true).write(true).open(path)?;
         let len_bytes = file.metadata()?.len();
         if len_bytes % (N as u64) != 0 {
             return Err(io::Error::new(io::ErrorKind::InvalidData, len_error));
@@ -145,18 +133,17 @@ impl<const N: usize> FixedWidthIndex<N> {
         } else {
             None
         };
+        file.seek(SeekFrom::End(0))?;
         Ok(Self {
-            file,
-            buf: Vec::with_capacity(APPEND_BUF_CAP_BYTES),
-            flushed_len: len,
+            writer: BufWriter::with_capacity(APPEND_BUF_CAP_BYTES, file),
             len,
             mmap,
         })
     }
 
     /// Open an existing file or create a new one without truncating existing content.
     fn open_or_create(path: impl AsRef<Path>, len_error: &'static str) -> io::Result<Self> {
-        let file = OpenOptions::new()
+        let mut file = OpenOptions::new()
             .read(true)
             .write(true)
             .truncate(false)
@@ -167,35 +154,26 @@ impl<const N: usize> FixedWidthIndex<N> {
             return Err(io::Error::new(io::ErrorKind::InvalidData, len_error));
         }
         let len = len_bytes / (N as u64);
+        file.seek(SeekFrom::End(0))?;
         Ok(Self {
-            file,
-            buf: Vec::with_capacity(APPEND_BUF_CAP_BYTES),
-            flushed_len: len,
+            writer: BufWriter::with_capacity(APPEND_BUF_CAP_BYTES, file),
             len,
             mmap: None,
         })
     }
 
-    /// Flush the in-memory append buffer to disk via a single positional write.
     fn flush(&mut self) -> io::Result<()> {
-        if self.buf.is_empty() {
-            return Ok(());
-        }
-        let offset = self.flushed_len * (N as u64);
-        self.file.write_all_at(&self.buf, offset)?;
-        self.flushed_len = self.len;
-        self.buf.clear();
-        Ok(())
+        self.writer.flush()
     }
 
     /// Remap the file for read access after all writes are complete.
     ///
     /// Must not be called while any concurrent writes to this file are in flight.
     fn remap(&mut self) -> io::Result<()> {
-        self.flush()?;
+        self.writer.flush()?;
         self.mmap = if self.len > 0 {
             // Safety: no more writes will occur on this handle after remap.
-            Some(unsafe { Mmap::map(&self.file)? })
+            Some(unsafe { Mmap::map(self.writer.get_ref())? })
         } else {
             None
         };
@@ -211,11 +189,8 @@ impl<const N: usize> FixedWidthIndex<N> {
     }
 
     fn append_bytes(&mut self, bytes: &[u8; N]) -> io::Result<u64> {
-        self.buf.extend_from_slice(bytes);
+        self.writer.write_all(bytes)?;
         self.len += 1;
-        if self.buf.len() >= APPEND_BUF_CAP_BYTES {
-            self.flush()?;
-        }
         Ok(self.len - 1)
     }
 
@@ -226,44 +201,34 @@ impl<const N: usize> FixedWidthIndex<N> {
                 "set_bytes index out of range",
             ));
         }
-        if index >= self.flushed_len {
-            let buf_offset = ((index - self.flushed_len) as usize) * N;
-            self.buf[buf_offset..buf_offset + N].copy_from_slice(bytes);
-            return Ok(());
-        }
+        // Flush so the record is on disk before the positional write.
+        self.writer.flush()?;
         let offset = index * (N as u64);
-        self.file.write_all_at(bytes, offset)
+        self.writer.get_ref().write_all_at(bytes, offset)
     }
 
     fn get_bytes(&self, index: u64) -> io::Result<Option<[u8; N]>> {
         if index >= self.len {
             return Ok(None);
         }
-        if index >= self.flushed_len {
-            let buf_offset = ((index - self.flushed_len) as usize) * N;
+        let buf = self.writer.buffer();
+        let flushed_count = self.len - (buf.len() / N) as u64;
+        if index >= flushed_count {
+            let buf_offset = ((index - flushed_count) as usize) * N;
             let mut out = [0u8; N];
-            out.copy_from_slice(&self.buf[buf_offset..buf_offset + N]);
+            out.copy_from_slice(&buf[buf_offset..buf_offset + N]);
             return Ok(Some(out));
         }
         let offset = index * (N as u64);
         if let Some(mmap) = &self.mmap {
             let offset = offset as usize;
-            let mut buf = [0u8; N];
-            buf.copy_from_slice(&mmap[offset..offset + N]);
-            return Ok(Some(buf));
+            let mut out = [0u8; N];
+            out.copy_from_slice(&mmap[offset..offset + N]);
+            return Ok(Some(out));
         }
-        let mut buf = [0u8; N];
-        self.file.read_exact_at(&mut buf, offset)?;
-        Ok(Some(buf))
-    }
-}
-
-impl<const N: usize> Drop for FixedWidthIndex<N> {
-    fn drop(&mut self) {
-        // Best-effort flush so that simply dropping an index persists all
-        // appended records, matching the previous unbuffered behaviour where
-        // every `append_bytes` landed on disk synchronously.
-        let _ = self.flush();
+        let mut out = [0u8; N];
+        self.writer.get_ref().read_exact_at(&mut out, offset)?;
+        Ok(Some(out))
     }
 }
 
@@ -513,6 +478,17 @@ impl DenseIndexSet {
         })
     }
 
+    /// Flush all buffered appends to disk.
+    ///
+    /// Must be called before any `set` on recently appended records so those
+    /// records are on disk for the positional write.
+    pub fn flush(&mut self) -> io::Result<()> {
+        self.txptr.inner.flush()?;
+        self.block_tx.inner.flush()?;
+        self.in_prevout.inner.flush()?;
+        self.out_spent.inner.flush()
+    }
+
     /// Map all four index files into memory for zero-syscall reads.
     ///
     /// Call once after all writes are complete.