Skip to content
This repository was archived by the owner on Sep 29, 2026. It is now read-only.
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions AGENT_CHAT.md
Original file line number Diff line number Diff line change
@@ -1,5 +1,9 @@
# RedlineDB Agent Chat

## 2026-09-25T19:40Z perf/positional-wal-batch

Claim: canonical checkout `/home/ubuntu/redlineDB`, branch `perf/positional-wal-batch` from `origin/main` `4c622a4fad6211133cd2bcd962375bca38170d0d`. Page and WAL bytes use positional read/write. The WAL writer emits one write for a contiguous run of records and splits the run at a segment boundary. WAL record bytes and durability are unchanged. `wal_pipeline` stays off. No speed ratio. No worktree.

## 2026-09-25T19:10Z perf/one-pass-scan

Claim: canonical checkout `/home/ubuntu/redlineDB`, branch `perf/one-pass-scan` from `origin/main` `1e21f23caa5842ca2469ec444a6edc4ac4bf36fa`. An unordered table scan keeps the row from the visibility load instead of decoding it again. Index probes stay rowid lookups. Join order and `LIMIT 1` stay as locked. No speed ratio. No worktree.
Expand Down
64 changes: 57 additions & 7 deletions crates/kernel/src/io/fs.rs
Original file line number Diff line number Diff line change
@@ -1,7 +1,10 @@
use std::fs::{self, File, OpenOptions};
use std::io::{Read, Seek, SeekFrom, Write};
use std::io::{self, ErrorKind};
use std::path::Path;

#[cfg(unix)]
use std::os::unix::fs::FileExt;

use crate::Result;

pub trait FileHandle {
Expand Down Expand Up @@ -75,15 +78,11 @@ impl FileHandle for StdFileHandle {
}

fn read_exact_at(&mut self, offset: u64, buf: &mut [u8]) -> Result<()> {
self.0.seek(SeekFrom::Start(offset))?;
self.0.read_exact(buf)?;
Ok(())
read_exact_at(&self.0, offset, buf)
}

fn write_all_at(&mut self, offset: u64, buf: &[u8]) -> Result<()> {
self.0.seek(SeekFrom::Start(offset))?;
self.0.write_all(buf)?;
Ok(())
write_all_at(&self.0, offset, buf)
}

fn sync_data(&self) -> Result<()> {
Expand All @@ -96,3 +95,54 @@ impl FileHandle for StdFileHandle {
Ok(())
}
}

#[cfg(unix)]
fn read_exact_at(file: &File, offset: u64, buf: &mut [u8]) -> Result<()> {
let mut filled = 0;
while filled < buf.len() {
let n = file.read_at(&mut buf[filled..], offset + filled as u64)?;
if n == 0 {
return Err(
io::Error::new(ErrorKind::UnexpectedEof, "failed to fill whole buffer").into(),
);
}
filled += n;
}
Ok(())
}

#[cfg(unix)]
fn write_all_at(file: &File, offset: u64, buf: &[u8]) -> Result<()> {
let mut written = 0;
while written < buf.len() {
let n = file.write_at(&buf[written..], offset + written as u64)?;
if n == 0 {
return Err(
io::Error::new(ErrorKind::WriteZero, "failed to write whole buffer").into(),
);
}
written += n;
}
Ok(())
}

#[cfg(test)]
mod tests {
use super::*;

#[test]
fn positional_reads_do_not_depend_on_write_order() {
let dir = tempfile::tempdir().expect("temp dir");
let mut file = StdFileSystem
.open_rw_create(&dir.path().join("bytes"))
.expect("open");
file.write_all_at(8, b"xyz").expect("tail");
file.write_all_at(0, b"abcd").expect("head");
let mut head = [0_u8; 4];
let mut tail = [0_u8; 3];
file.read_exact_at(0, &mut head).expect("read head");
file.read_exact_at(8, &mut tail).expect("read tail");
assert_eq!(&head, b"abcd");
assert_eq!(&tail, b"xyz");
}
}
4 changes: 2 additions & 2 deletions crates/kernel/src/wal/manager/coordinator/helpers.rs
Original file line number Diff line number Diff line change
Expand Up @@ -136,12 +136,12 @@ pub(super) fn drain_until(
}

let mut last_written = Lsn::ZERO;
for record in batch {
for record in &batch {
last_written = record.append.end_lsn;
totals.records = totals.records.saturating_add(1);
totals.bytes = totals.bytes.saturating_add(record.encoded.len() as u64);
wal.write_encoded(record.append, &record.encoded)?;
}
wal.write_encoded_batch(&batch)?;
if last_written != Lsn::ZERO {
publish_written_lsn(shared, last_written);
}
Expand Down
10 changes: 5 additions & 5 deletions crates/kernel/src/wal/manager/coordinator/writer.rs
Original file line number Diff line number Diff line change
Expand Up @@ -73,17 +73,17 @@ pub(super) fn wal_writer_loop(
}

let mut last_written = Lsn::ZERO;
for record in batch {
for record in &batch {
last_written = record.append.end_lsn;
// Lane GC: accumulate before the write so a torn-write
// failpoint doesn't desync the counter from durable
// state — the failure path returns immediately.
group_records = group_records.saturating_add(1);
group_bytes = group_bytes.saturating_add(record.encoded.len() as u64);
if let Err(_err) = wal.write_encoded(record.append, &record.encoded) {
publish_wal_failure(&shared);
return;
}
}
if let Err(_err) = wal.write_encoded_batch(&batch) {
publish_wal_failure(&shared);
return;
}
if last_written != Lsn::ZERO {
publish_written_lsn(&shared, last_written);
Expand Down
84 changes: 72 additions & 12 deletions crates/kernel/src/wal/manager/storage/write.rs
Original file line number Diff line number Diff line change
Expand Up @@ -190,41 +190,101 @@ impl<Fs: FileSystem> WalManager<Fs> {
}

pub fn write_encoded(&mut self, append: WalAppend, encoded: &[u8]) -> Result<()> {
// Lane E failpoint: armed before the WAL segment write so harnesses
// can inject torn-write or panic faults at the precise moment the
// record would land on disk.
crate::fail_point!("wal::write_encoded");
if encoded.len() > self.config.segment_bytes as usize {
return Err(Error::CorruptWal("record larger than wal segment"));
self.write_encoded_batch(&[QueuedWalRecord {
append,
encoded: encoded.to_vec(),
}])
}

/// Write a drained queue in as few positional writes as possible.
/// A run stops at a segment boundary or a gap in LSN. Each run is one
/// `write_all_at` and one `wal::write_encoded` failpoint. Record bytes
/// are unchanged.
pub(crate) fn write_encoded_batch(&mut self, records: &[QueuedWalRecord]) -> Result<()> {
let mut index = 0;
while index < records.len() {
self.prepare_for_record(records[index].append, records[index].encoded.len())?;
let offset = self.active_offset;
let mut bytes = Vec::new();
let mut last_start = records[index].append.start_lsn;
let mut last_end = records[index].append.end_lsn;
loop {
let record = &records[index];
let at = offset + bytes.len() as u64;
let expected = Lsn((self.active_segment - 1) * self.config.segment_bytes + at);
if record.append.start_lsn != expected {
break;
}
if at > 0 && at + record.encoded.len() as u64 > self.config.segment_bytes {
break;
}
bytes.extend_from_slice(&record.encoded);
last_start = record.append.start_lsn;
last_end = record.append.end_lsn;
index += 1;
if index == records.len() {
break;
}
}
if bytes.is_empty() {
return Err(Error::CorruptWal(
"record lsn does not match write position",
));
}
self.emit_wal_bytes(offset, &bytes, last_end, last_start)?;
}
Ok(())
}

fn prepare_for_record(&mut self, append: WalAppend, encoded_len: usize) -> Result<()> {
if encoded_len > self.config.segment_bytes as usize {
return Err(Error::CorruptWal("record larger than wal segment"));
}
// An exact fill leaves the cursor at `segment_bytes`. The next
// reserved LSN is the following segment, and that LSN compares
// equal to `(segment - 1) * size + offset` when offset == size,
// so the mismatch check below would not rotate.
if self.active_offset >= self.config.segment_bytes {
self.rotate_segment()?;
}
let expected_lsn =
Lsn((self.active_segment - 1) * self.config.segment_bytes + self.active_offset);
if expected_lsn != append.start_lsn
&& self.active_offset > 0
&& self.active_offset + encoded.len() as u64 > self.config.segment_bytes
&& self.active_offset + encoded_len as u64 > self.config.segment_bytes
{
self.rotate_segment()?;
}

let expected_lsn =
Lsn((self.active_segment - 1) * self.config.segment_bytes + self.active_offset);
if expected_lsn != append.start_lsn {
return Err(Error::CorruptWal(
"record lsn does not match write position",
));
}
Ok(())
}

self.active_file.write_all_at(self.active_offset, encoded)?;
fn emit_wal_bytes(
&mut self,
offset: u64,
bytes: &[u8],
end_lsn: Lsn,
last_start_lsn: Lsn,
) -> Result<()> {
// Lane E failpoint: armed before the bytes land, once per
// physical write rather than once per queued record.
crate::fail_point!("wal::write_encoded");
self.active_file.write_all_at(offset, bytes)?;
// Lane BH P1 #7: count the pwrite-equivalent before bumping
// the offset; the bench harness reads this through
// `WalCoordinator::sync_counters_snapshot`.
if let Some(counters) = &self.sync_counters {
counters.bump_pwrite();
}
self.active_offset += encoded.len() as u64;
self.written_lsn = append.end_lsn;
self.prev_lsn = append.start_lsn;
self.active_offset = offset + bytes.len() as u64;
self.written_lsn = end_lsn;
self.prev_lsn = last_start_lsn;
Ok(())
}

Expand Down
26 changes: 26 additions & 0 deletions crates/kernel/tests/wal_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -376,6 +376,32 @@ fn wal_coordinator_group_flushes_concurrent_targets() {
assert_eq!(reader.scan().unwrap().len(), 8);
}

#[test]
fn queued_exact_fill_stays_inside_the_segment() {
let temp = TempDir::new().unwrap();
let segment_bytes = 128;
let config = WalConfig {
segment_bytes,
group_commit_delay_us: 0,
..WalConfig::default()
};
let coordinator = WalCoordinator::create(temp.path(), config.clone()).unwrap();
// 48-byte header + 16-byte payload = 64. Two records fill the segment.
let mut last = Lsn::ZERO;
for _ in 0..3 {
last = coordinator
.append(WalRecordKind::PageDelta, TxId(1), vec![b'q'; 16])
.unwrap()
.end_lsn;
}
assert!(coordinator.flush_all().unwrap() >= last);
let first = temp.path().join(format!("{:020}.wal", 1_u64));
let len = std::fs::metadata(&first).unwrap().len();
assert_eq!(len, segment_bytes, "queued segment grew to {len}");
let mut reader = WalReader::new(temp.path(), config);
assert_eq!(reader.scan().unwrap().len(), 3);
}

#[test]
fn wal_coordinator_flush_until_returns_when_already_durable() {
let temp = TempDir::new().unwrap();
Expand Down
Loading