diff --git a/AGENT_CHAT.md b/AGENT_CHAT.md index 53fcd0c00..f287c0b03 100644 --- a/AGENT_CHAT.md +++ b/AGENT_CHAT.md @@ -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. diff --git a/crates/kernel/src/io/fs.rs b/crates/kernel/src/io/fs.rs index 66ed52b62..988aa6588 100644 --- a/crates/kernel/src/io/fs.rs +++ b/crates/kernel/src/io/fs.rs @@ -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 { @@ -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<()> { @@ -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"); + } +} diff --git a/crates/kernel/src/wal/manager/coordinator/helpers.rs b/crates/kernel/src/wal/manager/coordinator/helpers.rs index d5150aade..feac2965a 100644 --- a/crates/kernel/src/wal/manager/coordinator/helpers.rs +++ b/crates/kernel/src/wal/manager/coordinator/helpers.rs @@ -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); } diff --git a/crates/kernel/src/wal/manager/coordinator/writer.rs b/crates/kernel/src/wal/manager/coordinator/writer.rs index 24bf2d8eb..9965d8a5b 100644 --- a/crates/kernel/src/wal/manager/coordinator/writer.rs +++ b/crates/kernel/src/wal/manager/coordinator/writer.rs @@ -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); diff --git a/crates/kernel/src/wal/manager/storage/write.rs b/crates/kernel/src/wal/manager/storage/write.rs index 893731a4b..ab58ea857 100644 --- a/crates/kernel/src/wal/manager/storage/write.rs +++ b/crates/kernel/src/wal/manager/storage/write.rs @@ -190,23 +190,71 @@ impl WalManager { } 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 { @@ -214,17 +262,29 @@ impl WalManager { "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(()) } diff --git a/crates/kernel/tests/wal_tests.rs b/crates/kernel/tests/wal_tests.rs index 197825b89..913f53f09 100644 --- a/crates/kernel/tests/wal_tests.rs +++ b/crates/kernel/tests/wal_tests.rs @@ -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();