1use fsqlite_error::{FrankenError, Result};
17use fsqlite_types::cx::Cx;
18use fsqlite_types::flags::SyncFlags;
19use fsqlite_vfs::{SyncKind, VfsFile, VfsWriteCompletion};
20use tracing::{debug, error, warn};
21
22static PARANOID_DURABILITY: std::sync::LazyLock<bool> = std::sync::LazyLock::new(|| {
25 std::env::var("FRANKENSQLITE_PARANOID_DURABILITY")
26 .is_ok_and(|v| v == "1" || v.eq_ignore_ascii_case("true"))
27});
28
29use crate::checksum::{
30 SqliteWalChecksum, WAL_FORMAT_VERSION, WAL_FRAME_HEADER_SIZE, WAL_HEADER_SIZE, WAL_MAGIC_LE,
31 WalChecksumTransform, WalFrameHeader, WalHeader, WalSalts, compute_wal_frame_checksum,
32 read_wal_header_checksum, wal_header_checksum, write_wal_frame_checksum,
33 write_wal_frame_checksum_fields, write_wal_frame_salts,
34};
35
36#[inline]
37fn log_replay_decision(
38 replay_cursor: &'static str,
39 frame_no: usize,
40 commit_boundary: usize,
41 decision_reason: &'static str,
42) {
43 debug!(
44 replay_cursor,
45 frame_no, commit_boundary, decision_reason, "WAL replay decision"
46 );
47}
48
49async fn read_wal_header_torn_tolerant<F: VfsFile>(
60 file: &F,
61 cx: &Cx,
62 context: &'static str,
63 frame_count: usize,
64) -> Result<(WalHeader, SqliteWalChecksum)> {
65 const MAX_ATTEMPTS: usize = 2;
66 for attempt in 1..=MAX_ATTEMPTS {
67 let mut header_buf = [0u8; WAL_HEADER_SIZE];
68 let header_read = file.read(cx, &mut header_buf, 0).await?;
69 if header_read < WAL_HEADER_SIZE {
70 log_replay_decision(context, 0, frame_count, "header_short_read_corrupt");
71 return Err(FrankenError::WalCorrupt {
72 detail: format!(
73 "WAL file too small for header during {context}: read {header_read}, need {WAL_HEADER_SIZE}"
74 ),
75 });
76 }
77 let disk_header = WalHeader::from_bytes(&header_buf)?;
78 let disk_big_endian = disk_header.big_endian_checksum();
79 let disk_header_checksum = read_wal_header_checksum(&header_buf)?;
80 let expected_header_checksum = wal_header_checksum(&header_buf, disk_big_endian)?;
81 if disk_header_checksum == expected_header_checksum {
82 return Ok((disk_header, disk_header_checksum));
83 }
84 log_replay_decision(
85 context,
86 0,
87 frame_count,
88 if attempt < MAX_ATTEMPTS {
89 "header_checksum_mismatch_retry"
90 } else {
91 "header_checksum_mismatch_corrupt"
92 },
93 );
94 }
95 Err(FrankenError::WalCorrupt {
96 detail: format!("WAL header checksum mismatch during {context}"),
97 })
98}
99
100struct VfsWritePreflight<'a> {
101 completion: Option<&'a VfsWriteCompletion>,
102}
103
104impl<'a> VfsWritePreflight<'a> {
105 fn new(completion: Option<&'a VfsWriteCompletion>) -> Self {
106 Self { completion }
107 }
108
109 fn hand_off(&mut self) {
110 self.completion = None;
111 }
112}
113
114impl Drop for VfsWritePreflight<'_> {
115 fn drop(&mut self) {
116 if let Some(completion) = self.completion {
117 completion.complete_error();
118 }
119 }
120}
121
122#[derive(Debug, Clone, Copy)]
124pub struct WalAppendFrameRef<'a> {
125 pub page_number: u32,
127 pub page_data: &'a [u8],
129 pub db_size_if_commit: u32,
131}
132
133#[derive(Debug, Clone, Copy, PartialEq, Eq)]
140pub struct WalGenerationIdentity {
141 pub checkpoint_seq: u32,
143 pub salts: WalSalts,
145}
146
147impl WalGenerationIdentity {
148 #[must_use]
150 pub const fn from_header(header: &WalHeader) -> Self {
151 Self {
152 checkpoint_seq: header.checkpoint_seq,
153 salts: header.salts,
154 }
155 }
156}
157
158#[inline]
159fn push_wal_frame_bytes(
160 frame_scratch: &mut Vec<u8>,
161 page_number: u32,
162 db_size_if_commit: u32,
163 salts: WalSalts,
164 page_data: &[u8],
165) {
166 frame_scratch.extend_from_slice(&page_number.to_be_bytes());
167 frame_scratch.extend_from_slice(&db_size_if_commit.to_be_bytes());
168 frame_scratch.extend_from_slice(&salts.salt1.to_be_bytes());
169 frame_scratch.extend_from_slice(&salts.salt2.to_be_bytes());
170 frame_scratch.extend_from_slice(&[0_u8; 8]);
171 frame_scratch.extend_from_slice(page_data);
172}
173
174pub struct WalFile<F: VfsFile> {
179 file: F,
180 page_size: usize,
181 big_endian_checksum: bool,
182 header: WalHeader,
183 running_checksum: SqliteWalChecksum,
185 frame_count: usize,
187 last_commit_frame: Option<usize>,
189 frame_scratch: Vec<u8>,
195 last_fsynced_frame_count: usize,
200}
201
202impl<F: VfsFile> WalFile<F> {
203 pub async fn refresh(&mut self, cx: &Cx) -> Result<()> {
209 let frame_size = self.frame_size();
210 let expected_size = u64::try_from(WAL_HEADER_SIZE)
211 .expect("WAL header size fits u64")
212 .saturating_add(
213 u64::try_from(self.frame_count)
214 .unwrap_or(u64::MAX)
215 .saturating_mul(u64::try_from(frame_size).unwrap_or(u64::MAX)),
216 );
217 let file_size = self.file.file_size(cx)?;
218
219 if file_size < expected_size {
223 log_replay_decision("refresh", 0, self.frame_count, "file_shrank_rebuild");
224 return self.rebuild_state_from_file(cx).await;
225 }
226
227 let (disk_header, _disk_header_checksum) =
232 read_wal_header_torn_tolerant(&self.file, cx, "refresh", self.frame_count).await?;
233
234 if disk_header.magic != self.header.magic
236 || disk_header.format_version != self.header.format_version
237 || disk_header.page_size != self.header.page_size
238 || disk_header.checkpoint_seq != self.header.checkpoint_seq
239 || disk_header.salts != self.header.salts
240 {
241 log_replay_decision(
242 "refresh",
243 0,
244 self.frame_count,
245 "header_generation_changed_rebuild",
246 );
247 return self.rebuild_state_from_file(cx).await;
248 }
249
250 if file_size == expected_size {
251 return Ok(());
252 }
253
254 let frame_size_u64 = u64::try_from(frame_size).unwrap_or(u64::MAX);
264 let available_frames = usize::try_from(
265 file_size.saturating_sub(u64::try_from(WAL_HEADER_SIZE).unwrap_or(0)) / frame_size_u64,
266 )
267 .unwrap_or(usize::MAX);
268 if available_frames <= self.frame_count {
269 return Ok(());
270 }
271
272 let mut new_frame_count = self.frame_count;
273 let mut new_running_checksum = self.running_checksum;
274 let mut last_commit_count = self.frame_count;
275 let mut last_commit_checksum = self.running_checksum;
276
277 let mut frame_buf = vec![0u8; frame_size];
278 for frame_index in self.frame_count..available_frames {
279 let frame_no = frame_index.saturating_add(1);
280 let offset = self.frame_offset(frame_index);
281 let bytes_read = self.file.read(cx, &mut frame_buf, offset).await?;
282 if bytes_read < frame_size {
283 log_replay_decision(
284 "refresh_incremental",
285 frame_no,
286 last_commit_count,
287 "truncated_tail_stop",
288 );
289 break; }
291
292 let frame_header = WalFrameHeader::from_bytes(&frame_buf[..WAL_FRAME_HEADER_SIZE])?;
293 if frame_header.salts != self.header.salts {
294 log_replay_decision(
295 "refresh_incremental",
296 frame_no,
297 last_commit_count,
298 "salt_mismatch_stop",
299 );
300 break; }
302
303 let expected = compute_wal_frame_checksum(
304 &frame_buf,
305 self.page_size,
306 new_running_checksum,
307 self.big_endian_checksum,
308 )?;
309 if frame_header.checksum != expected {
310 log_replay_decision(
311 "refresh_incremental",
312 frame_no,
313 last_commit_count,
314 "checksum_mismatch_stop",
315 );
316 break; }
318
319 new_running_checksum = expected;
320 new_frame_count += 1;
321
322 if frame_header.is_commit() {
323 last_commit_count = new_frame_count;
324 last_commit_checksum = new_running_checksum;
325 log_replay_decision(
326 "refresh_incremental",
327 frame_no,
328 last_commit_count,
329 "accept_commit",
330 );
331 } else {
332 log_replay_decision(
333 "refresh_incremental",
334 frame_no,
335 last_commit_count,
336 "accept_non_commit",
337 );
338 }
339 }
340
341 self.frame_count = last_commit_count;
342 self.running_checksum = last_commit_checksum;
343 self.last_commit_frame = last_commit_count.checked_sub(1);
344
345 Ok(())
346 }
347
348 async fn rebuild_state_from_file(&mut self, cx: &Cx) -> Result<()> {
349 self.last_fsynced_frame_count = 0;
354
355 let (header, header_checksum) =
357 read_wal_header_torn_tolerant(&self.file, cx, "rebuild", self.frame_count).await?;
358 let page_size = usize::try_from(header.page_size).expect("WAL header page size fits usize");
359 let big_endian_checksum = header.big_endian_checksum();
360
361 self.header = header;
362 self.page_size = page_size;
363 self.big_endian_checksum = big_endian_checksum;
364 self.running_checksum = header_checksum;
365 self.frame_count = 0;
366
367 let mut new_frame_count = 0;
368 let mut new_running_checksum = header_checksum;
369 let mut last_commit_count = 0;
370 let mut last_commit_checksum = header_checksum;
371
372 let frame_size = self.frame_size();
373 let file_size = self.file.file_size(cx)?;
374 let max_frames = usize::try_from(
375 file_size.saturating_sub(u64::try_from(WAL_HEADER_SIZE).unwrap_or(0))
376 / u64::try_from(frame_size).unwrap_or(1),
377 )
378 .unwrap_or(usize::MAX);
379
380 let mut frame_buf = vec![0u8; frame_size];
381 for frame_index in 0..max_frames {
382 let frame_no = frame_index.saturating_add(1);
383 let offset = self.frame_offset(frame_index);
384 let bytes_read = self.file.read(cx, &mut frame_buf, offset).await?;
385 if bytes_read < frame_size {
386 log_replay_decision(
387 "rebuild",
388 frame_no,
389 last_commit_count,
390 "truncated_tail_stop",
391 );
392 break;
393 }
394
395 let frame_header = WalFrameHeader::from_bytes(&frame_buf[..WAL_FRAME_HEADER_SIZE])?;
396 if frame_header.salts != self.header.salts {
397 log_replay_decision("rebuild", frame_no, last_commit_count, "salt_mismatch_stop");
398 break;
399 }
400
401 let expected = compute_wal_frame_checksum(
402 &frame_buf,
403 self.page_size,
404 new_running_checksum,
405 self.big_endian_checksum,
406 )?;
407 if frame_header.checksum != expected {
408 log_replay_decision(
409 "rebuild",
410 frame_no,
411 last_commit_count,
412 "checksum_mismatch_stop",
413 );
414 break;
415 }
416
417 new_running_checksum = expected;
418 new_frame_count += 1;
419
420 if frame_header.is_commit() {
421 last_commit_count = new_frame_count;
422 last_commit_checksum = new_running_checksum;
423 log_replay_decision("rebuild", frame_no, last_commit_count, "accept_commit");
424 } else {
425 log_replay_decision("rebuild", frame_no, last_commit_count, "accept_non_commit");
426 }
427 }
428
429 self.frame_count = last_commit_count;
430 self.running_checksum = last_commit_checksum;
431 self.last_commit_frame = last_commit_count.checked_sub(1);
432
433 Ok(())
434 }
435
436 #[must_use]
438 pub fn frame_size(&self) -> usize {
439 WAL_FRAME_HEADER_SIZE + self.page_size
440 }
441
442 #[allow(clippy::cast_possible_truncation)]
444 pub(crate) fn frame_offset(&self, index: usize) -> u64 {
445 let header_size = WAL_HEADER_SIZE as u64;
448 let idx = index as u64;
449 let frame_sz = self.frame_size() as u64;
450 header_size.saturating_add(idx.saturating_mul(frame_sz))
451 }
452
453 #[must_use]
455 pub fn frame_count(&self) -> usize {
456 self.frame_count
457 }
458
459 #[must_use]
461 pub fn header(&self) -> &WalHeader {
462 &self.header
463 }
464
465 #[must_use]
467 pub fn generation_identity(&self) -> WalGenerationIdentity {
468 WalGenerationIdentity::from_header(&self.header)
469 }
470
471 #[must_use]
473 pub fn page_size(&self) -> usize {
474 self.page_size
475 }
476
477 #[must_use]
479 pub fn big_endian_checksum(&self) -> bool {
480 self.big_endian_checksum
481 }
482
483 #[must_use]
485 pub fn running_checksum(&self) -> SqliteWalChecksum {
486 self.running_checksum
487 }
488
489 #[cfg(test)]
490 #[must_use]
491 fn frame_scratch_len(&self) -> usize {
492 self.frame_scratch.len()
493 }
494
495 #[cfg(test)]
496 #[must_use]
497 fn frame_scratch_capacity(&self) -> usize {
498 self.frame_scratch.capacity()
499 }
500
501 #[cfg(test)]
502 #[must_use]
503 fn frame_scratch_ptr(&self) -> *const u8 {
504 self.frame_scratch.as_ptr()
505 }
506
507 pub async fn create(
512 cx: &Cx,
513 mut file: F,
514 page_size: u32,
515 checkpoint_seq: u32,
516 salts: WalSalts,
517 ) -> Result<Self> {
518 let header = WalHeader {
519 magic: WAL_MAGIC_LE,
520 format_version: WAL_FORMAT_VERSION,
521 page_size,
522 checkpoint_seq,
523 salts,
524 checksum: SqliteWalChecksum::default(), };
526 let header_bytes = header.to_bytes()?;
527 file.write(cx, &header_bytes, 0).await?;
528 file.truncate(
529 cx,
530 u64::try_from(WAL_HEADER_SIZE).expect("header size fits u64"),
531 )?;
532 file.sync(cx, SyncFlags::NORMAL)?;
540
541 let running_checksum = read_wal_header_checksum(&header_bytes)?;
542
543 debug!(
544 page_size,
545 checkpoint_seq,
546 salt1 = header.salts.salt1,
547 salt2 = header.salts.salt2,
548 "WAL file created"
549 );
550 crate::metrics::GLOBAL_WAL_METRICS.set_wal_frames_current(0);
551
552 Ok(Self {
553 file,
554 page_size: usize::try_from(page_size).expect("page size fits usize"),
555 big_endian_checksum: false,
556 header,
557 running_checksum,
558 frame_count: 0,
559 last_commit_frame: None,
560 frame_scratch: Vec::new(),
561 last_fsynced_frame_count: 0,
562 })
563 }
564
565 #[allow(clippy::too_many_lines)]
569 pub async fn open(cx: &Cx, file: F) -> Result<Self> {
570 let (header, header_checksum) =
573 read_wal_header_torn_tolerant(&file, cx, "startup_open", 0).await?;
574 let page_size = usize::try_from(header.page_size).expect("WAL header page size fits usize");
575 let big_endian_checksum = header.big_endian_checksum();
576 let frame_size = WAL_FRAME_HEADER_SIZE + page_size;
577
578 let file_size = file.file_size(cx)?;
580 let data_bytes =
581 file_size.saturating_sub(u64::try_from(WAL_HEADER_SIZE).expect("header size fits u64"));
582 let max_frames = usize::try_from(data_bytes / u64::try_from(frame_size).unwrap_or(1))
583 .unwrap_or(usize::MAX);
584
585 let mut running_checksum = header_checksum;
586 let mut valid_frames = 0_usize;
587 let mut last_commit_frames = 0_usize;
588 let mut last_commit_checksum = header_checksum;
589 let mut frame_buf = vec![0u8; frame_size];
590
591 for frame_index in 0..max_frames {
592 let frame_no = frame_index.saturating_add(1);
593 let header_size = WAL_HEADER_SIZE as u64;
598 let idx = frame_index as u64;
599 let frame_sz = frame_size as u64;
600 let file_offset = header_size.saturating_add(idx.saturating_mul(frame_sz));
601
602 let bytes_read = file.read(cx, &mut frame_buf, file_offset).await?;
603 if bytes_read < frame_size {
604 log_replay_decision(
605 "startup_open",
606 frame_no,
607 last_commit_frames,
608 "truncated_tail_stop",
609 );
610 break; }
612
613 let frame_header = WalFrameHeader::from_bytes(&frame_buf[..WAL_FRAME_HEADER_SIZE])?;
615 if frame_header.salts != header.salts {
616 warn!(frame_index, "WAL frame salt mismatch — chain terminated");
617 log_replay_decision(
618 "startup_open",
619 frame_no,
620 last_commit_frames,
621 "salt_mismatch_stop",
622 );
623 break; }
625
626 let expected = compute_wal_frame_checksum(
628 &frame_buf,
629 page_size,
630 running_checksum,
631 big_endian_checksum,
632 )?;
633 if frame_header.checksum != expected {
634 warn!(
635 frame_index,
636 "WAL frame checksum mismatch — chain terminated"
637 );
638 log_replay_decision(
639 "startup_open",
640 frame_no,
641 last_commit_frames,
642 "checksum_mismatch_stop",
643 );
644 break; }
646
647 running_checksum = expected;
648 valid_frames += 1;
649
650 if frame_header.is_commit() {
651 last_commit_frames = valid_frames;
652 last_commit_checksum = running_checksum;
653 log_replay_decision(
654 "startup_open",
655 frame_no,
656 last_commit_frames,
657 "accept_commit",
658 );
659 } else {
660 log_replay_decision(
661 "startup_open",
662 frame_no,
663 last_commit_frames,
664 "accept_non_commit",
665 );
666 }
667 }
668
669 debug!(
670 page_size,
671 big_endian_checksum,
672 checkpoint_seq = header.checkpoint_seq,
673 valid_frames = last_commit_frames,
674 "WAL file opened"
675 );
676 crate::metrics::GLOBAL_WAL_METRICS
677 .set_wal_frames_current(u64::try_from(last_commit_frames).unwrap_or(u64::MAX));
678
679 Ok(Self {
680 file,
681 page_size,
682 big_endian_checksum,
683 header,
684 running_checksum: last_commit_checksum,
685 frame_count: last_commit_frames,
686 last_commit_frame: last_commit_frames.checked_sub(1),
687 frame_scratch: Vec::new(),
688 last_fsynced_frame_count: last_commit_frames,
689 })
690 }
691
692 pub fn advance_state_after_write(
698 &mut self,
699 frames_written: usize,
700 new_running_checksum: SqliteWalChecksum,
701 ) -> Result<()> {
702 let new_count = self
703 .frame_count
704 .checked_add(frames_written)
705 .ok_or(FrankenError::DatabaseFull)?;
706
707 if new_count > usize::try_from(u32::MAX).unwrap_or(usize::MAX) {
708 return Err(FrankenError::DatabaseFull);
709 }
710
711 self.frame_count = new_count;
712 self.running_checksum = new_running_checksum;
713 crate::metrics::GLOBAL_WAL_METRICS
714 .set_wal_frames_current(u64::try_from(self.frame_count).unwrap_or(u64::MAX));
715 Ok(())
716 }
717
718 pub async fn append_frame(
725 &mut self,
726 cx: &Cx,
727 page_number: u32,
728 page_data: &[u8],
729 db_size_if_commit: u32,
730 ) -> Result<()> {
731 if self.frame_count >= usize::try_from(u32::MAX).unwrap_or(usize::MAX) {
732 return Err(FrankenError::DatabaseFull);
733 }
734
735 if page_data.len() != self.page_size {
736 return Err(FrankenError::WalCorrupt {
737 detail: format!(
738 "page data size mismatch: expected {}, got {}",
739 self.page_size,
740 page_data.len()
741 ),
742 });
743 }
744
745 let frame_size = self.frame_size();
747 let page_size = self.page_size;
748 let salts = self.header.salts;
749 let running_checksum = self.running_checksum;
750 let big_endian_checksum = self.big_endian_checksum;
751 let offset = self.frame_offset(self.frame_count);
752
753 let mut frame_scratch = std::mem::take(&mut self.frame_scratch);
754 frame_scratch.clear();
755 if frame_scratch.capacity() < frame_size {
756 frame_scratch.reserve(frame_size - frame_scratch.capacity());
757 }
758 let append_result = async {
759 push_wal_frame_bytes(
760 &mut frame_scratch,
761 page_number,
762 db_size_if_commit,
763 salts,
764 page_data,
765 );
766 let frame = &mut frame_scratch[..frame_size];
767
768 let new_checksum =
770 write_wal_frame_checksum(frame, page_size, running_checksum, big_endian_checksum)?;
771
772 self.file.write(cx, frame, offset).await?;
773 Ok::<_, FrankenError>(new_checksum)
774 }
775 .await;
776 self.frame_scratch = frame_scratch;
777 let new_checksum = append_result?;
778
779 self.running_checksum = new_checksum;
780 self.frame_count += 1;
781 if db_size_if_commit != 0 {
782 self.last_commit_frame = Some(self.frame_count - 1);
783 }
784 crate::metrics::GLOBAL_WAL_METRICS
785 .set_wal_frames_current(u64::try_from(self.frame_count).unwrap_or(u64::MAX));
786
787 let bytes_written = u64::try_from(frame_size).unwrap_or(u64::MAX);
788 let span = tracing::span!(
789 tracing::Level::DEBUG,
790 "wal_write",
791 frame_count = self.frame_count,
792 bytes_written = bytes_written,
793 page_number = page_number,
794 is_commit = db_size_if_commit > 0,
795 );
796 let _guard = span.enter();
797
798 debug!(
799 frame_index = self.frame_count - 1,
800 page_number,
801 is_commit = db_size_if_commit > 0,
802 "WAL frame appended"
803 );
804
805 crate::metrics::GLOBAL_WAL_METRICS.record_frame_write(bytes_written);
806
807 Ok(())
808 }
809
810 pub fn prepare_frame_bytes(&self, frames: &[WalAppendFrameRef<'_>]) -> Result<Vec<u8>> {
817 let mut frame_buf = Vec::new();
818 let mut checksum_transforms = Vec::new();
819 let _ = self.prepare_frame_bytes_with_transforms_into(
820 frames.len(),
821 frames.iter().copied(),
822 &mut frame_buf,
823 &mut checksum_transforms,
824 )?;
825 Ok(frame_buf)
826 }
827
828 pub fn prepare_frame_bytes_with_transforms_into<'a, I>(
835 &self,
836 frame_count: usize,
837 frames: I,
838 frame_buf: &mut Vec<u8>,
839 checksum_transforms: &mut Vec<WalChecksumTransform>,
840 ) -> Result<Option<usize>>
841 where
842 I: IntoIterator<Item = WalAppendFrameRef<'a>>,
843 {
844 frame_buf.clear();
845 checksum_transforms.clear();
846 if frame_count == 0 {
847 return Ok(None);
848 }
849
850 let frame_size = self.frame_size();
851 let total_bytes = frame_count
852 .checked_mul(frame_size)
853 .ok_or(FrankenError::DatabaseFull)?;
854 frame_buf.resize(total_bytes, 0);
855 if checksum_transforms.capacity() < frame_count {
856 checksum_transforms.reserve(frame_count - checksum_transforms.capacity());
857 }
858
859 let mut observed_frame_count = 0usize;
860 let mut last_commit_offset = None;
861 for (idx, frame) in frames.into_iter().enumerate() {
862 if idx >= frame_count {
863 return Err(FrankenError::WalCorrupt {
864 detail: format!(
865 "prepared batch frame count mismatch: expected {frame_count}, got more than declared"
866 ),
867 });
868 }
869 if frame.page_data.len() != self.page_size {
870 return Err(FrankenError::WalCorrupt {
871 detail: format!(
872 "page data size mismatch in batch frame {idx}: expected {}, got {}",
873 self.page_size,
874 frame.page_data.len()
875 ),
876 });
877 }
878
879 let buf_offset = idx
880 .checked_mul(frame_size)
881 .ok_or(FrankenError::DatabaseFull)?;
882 let frame_slice = &mut frame_buf[buf_offset..buf_offset + frame_size];
883
884 frame_slice[..4].copy_from_slice(&frame.page_number.to_be_bytes());
885 frame_slice[4..8].copy_from_slice(&frame.db_size_if_commit.to_be_bytes());
886 write_wal_frame_salts(&mut frame_slice[..WAL_FRAME_HEADER_SIZE], self.header.salts)?;
887 frame_slice[WAL_FRAME_HEADER_SIZE..].copy_from_slice(frame.page_data);
888 checksum_transforms.push(WalChecksumTransform::for_wal_frame(
889 frame_slice,
890 self.page_size,
891 self.big_endian_checksum,
892 )?);
893 if frame.db_size_if_commit != 0 {
894 last_commit_offset = Some(idx);
895 }
896 observed_frame_count = idx.saturating_add(1);
897 }
898
899 if observed_frame_count != frame_count {
900 return Err(FrankenError::WalCorrupt {
901 detail: format!(
902 "prepared batch frame count mismatch: expected {frame_count}, got {observed_frame_count}"
903 ),
904 });
905 }
906
907 Ok(last_commit_offset)
908 }
909
910 pub async fn prepared_append_window_still_current(
917 &self,
918 cx: &Cx,
919 generation: WalGenerationIdentity,
920 start_frame_index: usize,
921 ) -> Result<bool> {
922 let expected_size = u64::try_from(WAL_HEADER_SIZE)
923 .expect("WAL header size fits u64")
924 .saturating_add(
925 u64::try_from(start_frame_index)
926 .unwrap_or(u64::MAX)
927 .saturating_mul(u64::try_from(self.frame_size()).unwrap_or(u64::MAX)),
928 );
929 if self.file.file_size(cx)? != expected_size {
930 return Ok(false);
931 }
932
933 let mut header_buf = [0u8; WAL_HEADER_SIZE];
934 let header_read = self.file.read(cx, &mut header_buf, 0).await?;
935 if header_read < WAL_HEADER_SIZE {
936 return Err(FrankenError::WalCorrupt {
937 detail: format!(
938 "WAL file too small for header during prepared append validation: read {header_read}, need {WAL_HEADER_SIZE}"
939 ),
940 });
941 }
942
943 let disk_header = WalHeader::from_bytes(&header_buf)?;
944 Ok(WalGenerationIdentity::from_header(&disk_header) == generation)
945 }
946
947 pub fn finalize_prepared_frame_bytes(
954 &self,
955 prepared_frame_bytes: &mut [u8],
956 frame_transforms: &[WalChecksumTransform],
957 ) -> Result<SqliteWalChecksum> {
958 let frame_count = frame_transforms.len();
959 if frame_count == 0 {
960 return Ok(self.running_checksum);
961 }
962
963 let frame_size = self.frame_size();
964 let expected_bytes = frame_count
965 .checked_mul(frame_size)
966 .ok_or(FrankenError::DatabaseFull)?;
967 if prepared_frame_bytes.len() != expected_bytes {
968 return Err(FrankenError::WalCorrupt {
969 detail: format!(
970 "prepared batch byte length mismatch: expected {expected_bytes}, got {}",
971 prepared_frame_bytes.len()
972 ),
973 });
974 }
975
976 let mut running_checksum = self.running_checksum;
977 for (frame_slice, frame_transform) in prepared_frame_bytes
978 .chunks_exact_mut(frame_size)
979 .zip(frame_transforms.iter())
980 {
981 write_wal_frame_salts(&mut frame_slice[..WAL_FRAME_HEADER_SIZE], self.header.salts)?;
982 running_checksum = frame_transform.apply(running_checksum);
983 write_wal_frame_checksum_fields(frame_slice, running_checksum)?;
984 }
985
986 Ok(running_checksum)
987 }
988
989 pub async fn append_finalized_prepared_frame_bytes(
992 &mut self,
993 cx: &Cx,
994 prepared_frame_bytes: &[u8],
995 frame_count: usize,
996 final_running_checksum: SqliteWalChecksum,
997 last_commit_offset: Option<usize>,
998 ) -> Result<()> {
999 self.append_finalized_prepared_frame_bytes_with_completion(
1000 cx,
1001 prepared_frame_bytes,
1002 frame_count,
1003 final_running_checksum,
1004 last_commit_offset,
1005 None,
1006 )
1007 .await
1008 }
1009
1010 pub async fn append_finalized_prepared_frame_bytes_tracked(
1015 &mut self,
1016 cx: &Cx,
1017 prepared_frame_bytes: &[u8],
1018 frame_count: usize,
1019 final_running_checksum: SqliteWalChecksum,
1020 last_commit_offset: Option<usize>,
1021 completion: VfsWriteCompletion,
1022 ) -> Result<()> {
1023 self.append_finalized_prepared_frame_bytes_with_completion(
1024 cx,
1025 prepared_frame_bytes,
1026 frame_count,
1027 final_running_checksum,
1028 last_commit_offset,
1029 Some(&completion),
1030 )
1031 .await
1032 }
1033
1034 async fn append_finalized_prepared_frame_bytes_with_completion(
1035 &mut self,
1036 cx: &Cx,
1037 prepared_frame_bytes: &[u8],
1038 frame_count: usize,
1039 final_running_checksum: SqliteWalChecksum,
1040 last_commit_offset: Option<usize>,
1041 completion: Option<&VfsWriteCompletion>,
1042 ) -> Result<()> {
1043 let mut preflight = VfsWritePreflight::new(completion);
1044 if frame_count == 0 {
1045 if let Some(completion) = completion {
1046 completion.complete_success();
1047 }
1048 preflight.hand_off();
1049 return Ok(());
1050 }
1051
1052 let new_count = self
1053 .frame_count
1054 .checked_add(frame_count)
1055 .ok_or(FrankenError::DatabaseFull)?;
1056 if new_count > usize::try_from(u32::MAX).unwrap_or(usize::MAX) {
1057 return Err(FrankenError::DatabaseFull);
1058 }
1059
1060 let frame_size = self.frame_size();
1061 let expected_bytes = frame_count
1062 .checked_mul(frame_size)
1063 .ok_or(FrankenError::DatabaseFull)?;
1064 if prepared_frame_bytes.len() != expected_bytes {
1065 return Err(FrankenError::WalCorrupt {
1066 detail: format!(
1067 "prepared batch byte length mismatch: expected {expected_bytes}, got {}",
1068 prepared_frame_bytes.len()
1069 ),
1070 });
1071 }
1072
1073 let start_frame_index = self.frame_count;
1074 let offset = self.frame_offset(start_frame_index);
1075
1076 #[cfg(any(test, feature = "fault-injection"))]
1077 crate::fault_hooks::maybe_inject_crash_at(
1078 crate::fault_hooks::CrashBoundary::BeforeWalFrameAppend,
1079 &format!("start_frame={start_frame_index} frame_count={frame_count}"),
1080 )?;
1081
1082 preflight.hand_off();
1083 if let Some(completion) = completion {
1084 self.file
1085 .write_tracked(cx, prepared_frame_bytes, offset, completion.clone())
1086 .await?;
1087 } else {
1088 self.file.write(cx, prepared_frame_bytes, offset).await?;
1089 }
1090 self.advance_state_after_write(frame_count, final_running_checksum)?;
1091 if let Some(last_commit_offset) = last_commit_offset {
1092 self.last_commit_frame = Some(start_frame_index + last_commit_offset);
1093 }
1094
1095 #[cfg(any(test, feature = "fault-injection"))]
1096 crate::fault_hooks::maybe_inject_crash_at(
1097 crate::fault_hooks::CrashBoundary::AfterWalFrameAppendBeforeFsync,
1098 &format!(
1099 "end_frame={} frames_written={frame_count}",
1100 self.frame_count
1101 ),
1102 )?;
1103
1104 let bytes_per_frame = u64::try_from(frame_size).unwrap_or(u64::MAX);
1105 let bytes_written = u64::try_from(expected_bytes).unwrap_or(u64::MAX);
1106 let span = tracing::span!(
1107 tracing::Level::DEBUG,
1108 "wal_batch_write",
1109 start_frame_index = start_frame_index,
1110 frames_written = frame_count,
1111 bytes_written = bytes_written,
1112 );
1113 let _guard = span.enter();
1114
1115 debug!(
1116 end_frame_count = self.frame_count,
1117 frames_written = frame_count,
1118 "WAL frames appended in batch"
1119 );
1120
1121 for _ in 0..frame_count {
1122 crate::metrics::GLOBAL_WAL_METRICS.record_frame_write(bytes_per_frame);
1123 }
1124
1125 Ok(())
1126 }
1127
1128 pub async fn append_prepared_frame_bytes(
1135 &mut self,
1136 cx: &Cx,
1137 prepared_frame_bytes: &mut [u8],
1138 frame_transforms: &[WalChecksumTransform],
1139 ) -> Result<()> {
1140 self.append_prepared_frame_bytes_with_completion(
1141 cx,
1142 prepared_frame_bytes,
1143 frame_transforms,
1144 None,
1145 )
1146 .await
1147 }
1148
1149 pub async fn append_prepared_frame_bytes_tracked(
1151 &mut self,
1152 cx: &Cx,
1153 prepared_frame_bytes: &mut [u8],
1154 frame_transforms: &[WalChecksumTransform],
1155 completion: VfsWriteCompletion,
1156 ) -> Result<()> {
1157 self.append_prepared_frame_bytes_with_completion(
1158 cx,
1159 prepared_frame_bytes,
1160 frame_transforms,
1161 Some(&completion),
1162 )
1163 .await
1164 }
1165
1166 async fn append_prepared_frame_bytes_with_completion(
1167 &mut self,
1168 cx: &Cx,
1169 prepared_frame_bytes: &mut [u8],
1170 frame_transforms: &[WalChecksumTransform],
1171 completion: Option<&VfsWriteCompletion>,
1172 ) -> Result<()> {
1173 let mut preflight = VfsWritePreflight::new(completion);
1174 let frame_count = frame_transforms.len();
1175 if frame_count == 0 {
1176 if let Some(completion) = completion {
1177 completion.complete_success();
1178 }
1179 preflight.hand_off();
1180 return Ok(());
1181 }
1182
1183 let new_count = self
1184 .frame_count
1185 .checked_add(frame_count)
1186 .ok_or(FrankenError::DatabaseFull)?;
1187 if new_count > usize::try_from(u32::MAX).unwrap_or(usize::MAX) {
1188 return Err(FrankenError::DatabaseFull);
1189 }
1190
1191 let frame_size = self.frame_size();
1192 let running_checksum =
1193 self.finalize_prepared_frame_bytes(prepared_frame_bytes, frame_transforms)?;
1194 let last_commit_offset = prepared_frame_bytes
1195 .chunks_exact(frame_size)
1196 .enumerate()
1197 .rev()
1198 .find_map(|(offset, frame_slice)| {
1199 let db_size_if_commit = u32::from_be_bytes([
1200 frame_slice[4],
1201 frame_slice[5],
1202 frame_slice[6],
1203 frame_slice[7],
1204 ]);
1205 (db_size_if_commit != 0).then_some(offset)
1206 });
1207 preflight.hand_off();
1208 self.append_finalized_prepared_frame_bytes_with_completion(
1209 cx,
1210 prepared_frame_bytes,
1211 frame_count,
1212 running_checksum,
1213 last_commit_offset,
1214 completion,
1215 )
1216 .await
1217 }
1218
1219 pub async fn append_frames(&mut self, cx: &Cx, frames: &[WalAppendFrameRef<'_>]) -> Result<()> {
1225 self.append_frame_iter(cx, frames.len(), frames.iter().copied())
1226 .await
1227 }
1228
1229 pub async fn append_frames_tracked(
1231 &mut self,
1232 cx: &Cx,
1233 frames: &[WalAppendFrameRef<'_>],
1234 completion: VfsWriteCompletion,
1235 ) -> Result<()> {
1236 self.append_frame_iter_tracked(cx, frames.len(), frames.iter().copied(), completion)
1237 .await
1238 }
1239
1240 pub(crate) async fn append_frame_iter<'a, I>(
1243 &mut self,
1244 cx: &Cx,
1245 frame_count: usize,
1246 frames: I,
1247 ) -> Result<()>
1248 where
1249 I: IntoIterator<Item = WalAppendFrameRef<'a>>,
1250 {
1251 self.append_frame_iter_with_completion(cx, frame_count, frames, None)
1252 .await
1253 }
1254
1255 pub(crate) async fn append_frame_iter_tracked<'a, I>(
1256 &mut self,
1257 cx: &Cx,
1258 frame_count: usize,
1259 frames: I,
1260 completion: VfsWriteCompletion,
1261 ) -> Result<()>
1262 where
1263 I: IntoIterator<Item = WalAppendFrameRef<'a>>,
1264 {
1265 self.append_frame_iter_with_completion(cx, frame_count, frames, Some(&completion))
1266 .await
1267 }
1268
1269 async fn append_frame_iter_with_completion<'a, I>(
1270 &mut self,
1271 cx: &Cx,
1272 frame_count: usize,
1273 frames: I,
1274 completion: Option<&VfsWriteCompletion>,
1275 ) -> Result<()>
1276 where
1277 I: IntoIterator<Item = WalAppendFrameRef<'a>>,
1278 {
1279 let mut preflight = VfsWritePreflight::new(completion);
1280 if frame_count == 0 {
1281 if let Some(completion) = completion {
1282 completion.complete_success();
1283 }
1284 preflight.hand_off();
1285 return Ok(());
1286 }
1287
1288 #[cfg(any(test, feature = "fault-injection"))]
1289 crate::fault_hooks::maybe_inject_append_busy(self.frame_count, frame_count)?;
1290
1291 let frame_size = self.frame_size();
1292 let total_bytes = frame_count
1293 .checked_mul(frame_size)
1294 .ok_or(FrankenError::DatabaseFull)?;
1295 let page_size = self.page_size;
1296 let salts = self.header.salts;
1297 let big_endian_checksum = self.big_endian_checksum;
1298 #[cfg(any(test, feature = "fault-injection"))]
1299 let frame_count_before = self.frame_count;
1300
1301 let mut frame_scratch = std::mem::take(&mut self.frame_scratch);
1302 frame_scratch.clear();
1303 if frame_scratch.capacity() < total_bytes {
1304 frame_scratch.reserve(total_bytes - frame_scratch.capacity());
1305 }
1306 let append_result = async {
1311 let mut running_checksum = self.running_checksum;
1312 let mut last_commit_offset: Option<usize> = None;
1313 let mut observed_frame_count = 0usize;
1314
1315 for (idx, frame) in frames.into_iter().enumerate() {
1316 if idx >= frame_count {
1317 return Err(FrankenError::WalCorrupt {
1318 detail: format!(
1319 "append batch frame count mismatch: expected {frame_count}, got more than declared"
1320 ),
1321 });
1322 }
1323 if frame.page_data.len() != page_size {
1324 return Err(FrankenError::WalCorrupt {
1325 detail: format!(
1326 "page data size mismatch in batch frame {idx}: expected {page_size}, got {}",
1327 frame.page_data.len()
1328 ),
1329 });
1330 }
1331
1332 let buf_offset = idx
1333 .checked_mul(frame_size)
1334 .ok_or(FrankenError::DatabaseFull)?;
1335
1336 push_wal_frame_bytes(
1338 &mut frame_scratch,
1339 frame.page_number,
1340 frame.db_size_if_commit,
1341 salts,
1342 frame.page_data,
1343 );
1344 let frame_slice = &mut frame_scratch[buf_offset..buf_offset + frame_size];
1345
1346 running_checksum = write_wal_frame_checksum(
1348 frame_slice,
1349 page_size,
1350 running_checksum,
1351 big_endian_checksum,
1352 )?;
1353
1354 if frame.db_size_if_commit != 0 {
1355 last_commit_offset = Some(idx);
1356 }
1357 observed_frame_count = idx + 1;
1358 }
1359
1360 if observed_frame_count != frame_count {
1361 return Err(FrankenError::WalCorrupt {
1362 detail: format!(
1363 "append batch frame count mismatch: expected {frame_count}, got {observed_frame_count}"
1364 ),
1365 });
1366 }
1367
1368 preflight.hand_off();
1369 self.append_finalized_prepared_frame_bytes_with_completion(
1370 cx,
1371 &frame_scratch,
1372 frame_count,
1373 running_checksum,
1374 last_commit_offset,
1375 completion,
1376 )
1377 .await
1378 }
1379 .await;
1380 self.frame_scratch = frame_scratch;
1381
1382 #[cfg(any(test, feature = "fault-injection"))]
1383 if append_result.is_ok() {
1384 crate::fault_hooks::maybe_inject_after_append(frame_count_before, frame_count)?;
1385 }
1386
1387 append_result
1388 }
1389
1390 pub async fn read_frame(
1392 &self,
1393 cx: &Cx,
1394 frame_index: usize,
1395 ) -> Result<(WalFrameHeader, Vec<u8>)> {
1396 let frame_size = self.frame_size();
1397 let mut buf = vec![0u8; frame_size];
1398 let header = self.read_frame_into(cx, frame_index, &mut buf).await?;
1399 let page_data = buf[WAL_FRAME_HEADER_SIZE..].to_vec();
1400 Ok((header, page_data))
1401 }
1402
1403 pub async fn read_frame_into(
1409 &self,
1410 cx: &Cx,
1411 frame_index: usize,
1412 buf: &mut [u8],
1413 ) -> Result<WalFrameHeader> {
1414 if frame_index >= self.frame_count {
1415 return Err(FrankenError::WalCorrupt {
1416 detail: format!(
1417 "frame index {frame_index} out of range (count: {})",
1418 self.frame_count
1419 ),
1420 });
1421 }
1422
1423 let frame_size = self.frame_size();
1424 if buf.len() < frame_size {
1425 return Err(FrankenError::Internal(format!(
1426 "read_frame_into buffer too small: got {}, need {}",
1427 buf.len(),
1428 frame_size
1429 )));
1430 }
1431
1432 let offset = self.frame_offset(frame_index);
1433 let bytes_read = self.file.read(cx, &mut buf[..frame_size], offset).await?;
1434 if bytes_read < frame_size {
1435 return Err(FrankenError::WalCorrupt {
1436 detail: format!(
1437 "short read at frame {frame_index}: got {bytes_read}, need {frame_size}"
1438 ),
1439 });
1440 }
1441
1442 WalFrameHeader::from_bytes(&buf[..WAL_FRAME_HEADER_SIZE])
1443 }
1444
1445 pub async fn read_frame_header(&self, cx: &Cx, frame_index: usize) -> Result<WalFrameHeader> {
1447 if frame_index >= self.frame_count {
1448 return Err(FrankenError::WalCorrupt {
1449 detail: format!(
1450 "frame index {frame_index} out of range (count: {})",
1451 self.frame_count
1452 ),
1453 });
1454 }
1455
1456 let mut header_buf = [0u8; WAL_FRAME_HEADER_SIZE];
1457 let offset = self.frame_offset(frame_index);
1458 let bytes_read = self.file.read(cx, &mut header_buf, offset).await?;
1459 if bytes_read < WAL_FRAME_HEADER_SIZE {
1460 return Err(FrankenError::WalCorrupt {
1461 detail: format!("short header read at frame {frame_index}: got {bytes_read}"),
1462 });
1463 }
1464
1465 WalFrameHeader::from_bytes(&header_buf)
1466 }
1467
1468 pub fn last_commit_frame(&mut self, cx: &Cx) -> Result<Option<usize>> {
1470 let _ = cx;
1471 Ok(self.last_commit_frame)
1472 }
1473
1474 pub fn sync(&mut self, cx: &Cx, flags: SyncFlags) -> Result<()> {
1477 #[cfg(any(test, feature = "fault-injection"))]
1478 crate::fault_hooks::maybe_inject_sync_failure(self.frame_count, flags)?;
1479
1480 self.file.sync(cx, flags)?;
1481 self.last_fsynced_frame_count = self.frame_count;
1482 Ok(())
1483 }
1484
1485 pub fn repair_uncommitted_tail(&mut self, cx: &Cx) -> Result<()> {
1494 let committed_size = u64::try_from(WAL_HEADER_SIZE)
1495 .unwrap_or(u64::MAX)
1496 .checked_add(
1497 u64::try_from(self.frame_count)
1498 .unwrap_or(u64::MAX)
1499 .checked_mul(u64::try_from(self.frame_size()).unwrap_or(u64::MAX))
1500 .ok_or(FrankenError::DatabaseFull)?,
1501 )
1502 .ok_or(FrankenError::DatabaseFull)?;
1503 let file_size = self.file.file_size(cx)?;
1504 if file_size < committed_size {
1505 return Err(FrankenError::WalCorrupt {
1506 detail: format!(
1507 "WAL file shrank below its committed prefix: file_size={file_size}, committed_size={committed_size}"
1508 ),
1509 });
1510 }
1511 if file_size > committed_size {
1512 self.file.truncate(cx, committed_size)?;
1513 }
1514 Ok(())
1515 }
1516
1517 pub fn durable_sync(&mut self, cx: &Cx, kind: SyncKind) -> Result<()> {
1523 #[cfg(any(test, feature = "fault-injection"))]
1524 {
1525 let flags = match kind {
1526 SyncKind::DataOnly => SyncFlags::DATAONLY,
1527 SyncKind::DataAndMetadata | SyncKind::FullDurable => SyncFlags::FULL,
1528 };
1529 crate::fault_hooks::maybe_inject_sync_failure(self.frame_count, flags)?;
1530 }
1531
1532 self.file.durable_sync(cx, kind)?;
1533 self.last_fsynced_frame_count = self.frame_count;
1534
1535 debug!(
1536 target: "fsqlite_wal::durability",
1537 fsynced_up_to = self.frame_count,
1538 kind = ?kind,
1539 "WAL durable sync complete"
1540 );
1541
1542 #[cfg(any(test, feature = "fault-injection"))]
1543 crate::fault_hooks::maybe_inject_crash_at(
1544 crate::fault_hooks::CrashBoundary::AfterFsyncBeforePublish,
1545 &format!("fsynced_up_to={}", self.frame_count),
1546 )?;
1547
1548 Ok(())
1549 }
1550
1551 pub fn assert_publish_safe(&self, publish_frame_count: usize) -> Result<()> {
1557 if self.last_fsynced_frame_count >= publish_frame_count {
1558 return Ok(());
1559 }
1560
1561 let msg = format!(
1562 "publish-before-fsync: attempting to publish frame_count={publish_frame_count} \
1563 but last fsynced only up to {fsynced}",
1564 fsynced = self.last_fsynced_frame_count,
1565 );
1566
1567 debug_assert!(false, "WAL durability invariant violated: {msg}");
1568
1569 if *PARANOID_DURABILITY {
1570 error!(
1571 target: "fsqlite_wal::durability",
1572 publish_frame_count,
1573 last_fsynced = self.last_fsynced_frame_count,
1574 "PARANOID_DURABILITY: publish-before-fsync detected"
1575 );
1576 return Err(FrankenError::Internal(msg));
1577 }
1578
1579 Ok(())
1580 }
1581
1582 #[must_use]
1584 pub fn last_fsynced_frame_count(&self) -> usize {
1585 self.last_fsynced_frame_count
1586 }
1587
1588 pub async fn reset(
1594 &mut self,
1595 cx: &Cx,
1596 new_checkpoint_seq: u32,
1597 new_salts: WalSalts,
1598 truncate_file: bool,
1599 ) -> Result<()> {
1600 let new_header = WalHeader {
1601 magic: self.header.magic,
1602 format_version: WAL_FORMAT_VERSION,
1603 page_size: self.header.page_size,
1604 checkpoint_seq: new_checkpoint_seq,
1605 salts: new_salts,
1606 checksum: SqliteWalChecksum::default(),
1607 };
1608 let header_bytes = new_header.to_bytes()?;
1609
1610 #[cfg(any(test, feature = "fault-injection"))]
1611 crate::fault_hooks::maybe_inject_crash_at(
1612 crate::fault_hooks::CrashBoundary::BeforeWalHeaderWrite,
1613 &format!("checkpoint_seq={new_checkpoint_seq}"),
1614 )?;
1615
1616 self.file.write(cx, &header_bytes, 0).await?;
1617
1618 #[cfg(any(test, feature = "fault-injection"))]
1623 {
1624 let old_fc = self.frame_count;
1625 crate::fault_hooks::maybe_inject_crash_header_truncate(old_fc, new_checkpoint_seq)?;
1626 }
1627
1628 if truncate_file {
1629 self.file.truncate(
1630 cx,
1631 u64::try_from(WAL_HEADER_SIZE).expect("header size fits u64"),
1632 )?;
1633 }
1634
1635 self.file.sync(cx, SyncFlags::NORMAL)?;
1638
1639 self.running_checksum = read_wal_header_checksum(&header_bytes)?;
1640 self.header = WalHeader::from_bytes(&header_bytes)?;
1641 self.frame_count = 0;
1642 self.last_commit_frame = None;
1643 self.last_fsynced_frame_count = 0;
1644 self.frame_scratch.clear();
1645 crate::metrics::GLOBAL_WAL_METRICS.set_wal_frames_current(0);
1646
1647 debug!(
1648 checkpoint_seq = new_checkpoint_seq,
1649 salt1 = new_salts.salt1,
1650 salt2 = new_salts.salt2,
1651 "WAL reset"
1652 );
1653
1654 crate::metrics::GLOBAL_WAL_METRICS.record_wal_reset();
1655
1656 Ok(())
1657 }
1658
1659 pub fn close(mut self, cx: &Cx) -> Result<()> {
1661 self.file.close(cx)
1662 }
1663
1664 #[must_use]
1666 pub fn file(&self) -> &F {
1667 &self.file
1668 }
1669
1670 pub fn file_mut(&mut self) -> &mut F {
1672 &mut self.file
1673 }
1674}
1675
1676#[cfg(test)]
1677mod tests {
1678 use std::time::Instant;
1679
1680 use fsqlite_types::flags::VfsOpenFlags;
1681 use fsqlite_vfs::MemoryVfs;
1682 use fsqlite_vfs::traits::Vfs;
1683 use serde_json::{Value, json};
1684
1685 use super::*;
1686 use crate::test_support::FutureResultTestExt as _;
1687
1688 static FAULT_TEST_LOCK: crate::fault_hooks::FaultInjectionSessionLock =
1690 crate::fault_hooks::FaultInjectionSessionLock::new();
1691
1692 const PAGE_SIZE: u32 = 4096;
1693 const TRACK_C_SCRATCH_BENCH_BEAD_ID: &str = "bd-db300.3.4.3";
1694 const TRACK_C_SCRATCH_BENCH_WARMUP_ITERS: usize = 4;
1695 const TRACK_C_SCRATCH_BENCH_MEASURE_ITERS: usize = 12;
1696
1697 #[derive(Clone, Copy)]
1698 enum TrackCScratchBenchMode {
1699 FreshAllocBaseline,
1700 ScratchReuseCandidate,
1701 }
1702
1703 #[derive(Clone, Copy)]
1704 struct TrackCScratchBenchRun {
1705 elapsed_ns: u64,
1706 explicit_fresh_buffer_allocations: usize,
1707 scratch_capacity_growth_events: usize,
1708 peak_scratch_capacity_bytes: usize,
1709 frame_buffer_bytes_per_operation: usize,
1710 operations_per_sample: usize,
1711 }
1712
1713 fn test_cx() -> Cx {
1714 Cx::default()
1715 }
1716
1717 fn test_salts() -> WalSalts {
1718 WalSalts {
1719 salt1: 0xDEAD_BEEF,
1720 salt2: 0xCAFE_BABE,
1721 }
1722 }
1723
1724 fn sample_page(seed: u8) -> Vec<u8> {
1725 let page_size = usize::try_from(PAGE_SIZE).expect("page size fits usize");
1726 let mut page = vec![0u8; page_size];
1727 for (i, byte) in page.iter_mut().enumerate() {
1728 let reduced = u8::try_from(i % 251).expect("modulo fits u8");
1729 *byte = reduced ^ seed;
1730 }
1731 page
1732 }
1733
1734 fn frame_ref(
1735 page_number: u32,
1736 page_data: &[u8],
1737 db_size_if_commit: u32,
1738 ) -> WalAppendFrameRef<'_> {
1739 WalAppendFrameRef {
1740 page_number,
1741 page_data,
1742 db_size_if_commit,
1743 }
1744 }
1745
1746 fn open_wal_file(vfs: &MemoryVfs, cx: &Cx) -> <MemoryVfs as Vfs>::File {
1747 let flags = VfsOpenFlags::READWRITE | VfsOpenFlags::CREATE | VfsOpenFlags::WAL;
1748 let (file, _) = vfs
1749 .open(cx, Some(std::path::Path::new("test.db-wal")), flags)
1750 .expect("open WAL file");
1751 file
1752 }
1753
1754 struct ScriptedHeaderFile {
1759 headers: Vec<[u8; WAL_HEADER_SIZE]>,
1760 reads: std::sync::atomic::AtomicUsize,
1761 }
1762
1763 impl ScriptedHeaderFile {
1764 fn new(headers: Vec<[u8; WAL_HEADER_SIZE]>) -> Self {
1765 Self {
1766 headers,
1767 reads: std::sync::atomic::AtomicUsize::new(0),
1768 }
1769 }
1770 fn read_count(&self) -> usize {
1771 self.reads.load(std::sync::atomic::Ordering::SeqCst)
1772 }
1773 }
1774
1775 impl VfsFile for ScriptedHeaderFile {
1776 fn read<'a>(
1777 &'a self,
1778 _cx: &'a Cx,
1779 buf: &'a mut [u8],
1780 offset: u64,
1781 ) -> impl std::future::Future<Output = Result<usize>> + Send + 'a {
1782 assert_eq!(offset, 0, "scripted file only serves the header at offset 0");
1783 let idx = self
1784 .reads
1785 .fetch_add(1, std::sync::atomic::Ordering::SeqCst)
1786 .min(self.headers.len() - 1);
1787 let n = buf.len().min(WAL_HEADER_SIZE);
1788 buf[..n].copy_from_slice(&self.headers[idx][..n]);
1789 std::future::ready(Ok(n))
1790 }
1791
1792 fn close(&mut self, _cx: &Cx) -> Result<()> {
1793 Ok(())
1794 }
1795 fn write<'a>(
1796 &'a self,
1797 _cx: &'a Cx,
1798 _buf: &'a [u8],
1799 _offset: u64,
1800 ) -> impl std::future::Future<Output = Result<()>> + Send + 'a {
1801 unreachable!("scripted header file is read-only");
1802 #[expect(unreachable_code, reason = "type witness for the impl-Future return")]
1803 std::future::ready(Ok(()))
1804 }
1805 fn truncate(&mut self, _cx: &Cx, _size: u64) -> Result<()> {
1806 unreachable!()
1807 }
1808 fn sync(&mut self, _cx: &Cx, _flags: SyncFlags) -> Result<()> {
1809 unreachable!()
1810 }
1811 fn file_size(&self, _cx: &Cx) -> Result<u64> {
1812 Ok(u64::try_from(WAL_HEADER_SIZE).unwrap_or(0))
1813 }
1814 fn lock(&mut self, _cx: &Cx, _level: fsqlite_types::LockLevel) -> Result<()> {
1815 unreachable!()
1816 }
1817 fn unlock(&mut self, _cx: &Cx, _level: fsqlite_types::LockLevel) -> Result<()> {
1818 unreachable!()
1819 }
1820 fn lock_external_shared_snapshot(&mut self, _cx: &Cx) -> Result<()> {
1821 unreachable!()
1822 }
1823 fn restore_external_shared_snapshot_attempt(&mut self, _cx: &Cx) -> Result<()> {
1824 unreachable!()
1825 }
1826 fn lock_external_maintenance(&mut self, _cx: &Cx, _wal_mode: bool) -> Result<()> {
1827 unreachable!()
1828 }
1829 fn restore_external_maintenance_attempt(&mut self, _cx: &Cx) -> Result<()> {
1830 unreachable!()
1831 }
1832 fn check_reserved_lock(&self, _cx: &Cx) -> Result<bool> {
1833 unreachable!()
1834 }
1835 fn shm_map(
1836 &mut self,
1837 _cx: &Cx,
1838 _region: u32,
1839 _size: u32,
1840 _extend: bool,
1841 ) -> Result<fsqlite_vfs::ShmRegion> {
1842 unreachable!()
1843 }
1844 fn shm_lock(&mut self, _cx: &Cx, _offset: u32, _n: u32, _flags: u32) -> Result<()> {
1845 unreachable!()
1846 }
1847 fn shm_barrier(&self) {
1848 unreachable!()
1849 }
1850 fn shm_unmap(&mut self, _cx: &Cx, _delete: bool) -> Result<()> {
1851 unreachable!()
1852 }
1853 }
1854
1855 fn valid_and_corrupt_wal_headers(cx: &Cx) -> ([u8; WAL_HEADER_SIZE], [u8; WAL_HEADER_SIZE]) {
1859 let vfs = MemoryVfs::new();
1860 let create_file = open_wal_file(&vfs, cx);
1861 let _wal =
1862 WalFile::create(cx, create_file, PAGE_SIZE, 0, test_salts()).expect("create WAL");
1863 let reader = open_wal_file(&vfs, cx);
1864 let mut valid = [0u8; WAL_HEADER_SIZE];
1865 let n = reader.read(cx, &mut valid, 0).expect("read valid header");
1866 assert_eq!(n, WAL_HEADER_SIZE);
1867 let mut corrupt = valid;
1868 corrupt[WAL_HEADER_SIZE - 1] ^= 0xFF;
1869 (valid, corrupt)
1870 }
1871
1872 #[test]
1873 fn torn_wal_header_read_retries_before_corrupt_bd_mlz2t() {
1874 let cx = test_cx();
1875 let (valid, corrupt) = valid_and_corrupt_wal_headers(&cx);
1876 let expected_page_size = PAGE_SIZE;
1877
1878 let torn = ScriptedHeaderFile::new(vec![corrupt, valid]);
1881 let (header, _checksum) = read_wal_header_torn_tolerant(&torn, &cx, "test_torn", 0)
1882 .expect("torn read recovers on retry");
1883 assert_eq!(torn.read_count(), 2, "recovery takes exactly two reads");
1884 assert_eq!(header.page_size, expected_page_size);
1885
1886 let bad = ScriptedHeaderFile::new(vec![corrupt]);
1889 let err = read_wal_header_torn_tolerant(&bad, &cx, "test_bad", 0)
1890 .expect_err("persistent corruption stays corrupt");
1891 assert!(matches!(err, FrankenError::WalCorrupt { .. }));
1892 assert_eq!(bad.read_count(), 2, "both attempts are exhausted first");
1893
1894 let good = ScriptedHeaderFile::new(vec![valid]);
1896 read_wal_header_torn_tolerant(&good, &cx, "test_good", 0).expect("valid header reads");
1897 assert_eq!(good.read_count(), 1, "a valid header does not retry");
1898 }
1899
1900 fn track_c_scratch_run_summary(runs: &[TrackCScratchBenchRun]) -> Value {
1901 let mut elapsed_samples: Vec<_> = runs.iter().map(|run| run.elapsed_ns).collect();
1902 elapsed_samples.sort_unstable();
1903 let sample_count = elapsed_samples.len();
1904 let min_ns = elapsed_samples.first().copied().unwrap_or(0);
1905 let median_ns = if sample_count == 0 {
1906 0
1907 } else {
1908 elapsed_samples[sample_count / 2]
1909 };
1910 let max_ns = elapsed_samples.last().copied().unwrap_or(0);
1911 let mean_ns = if sample_count == 0 {
1912 0.0
1913 } else {
1914 let total_ns: u128 = elapsed_samples.iter().map(|ns| u128::from(*ns)).sum();
1915 (total_ns as f64) / (sample_count as f64)
1916 };
1917 let explicit_fresh_buffer_allocations = runs
1918 .iter()
1919 .map(|run| run.explicit_fresh_buffer_allocations)
1920 .max()
1921 .unwrap_or(0);
1922 let scratch_capacity_growth_events = runs
1923 .iter()
1924 .map(|run| run.scratch_capacity_growth_events)
1925 .max()
1926 .unwrap_or(0);
1927 let peak_scratch_capacity_bytes = runs
1928 .iter()
1929 .map(|run| run.peak_scratch_capacity_bytes)
1930 .max()
1931 .unwrap_or(0);
1932 let frame_buffer_bytes_per_operation = runs
1933 .first()
1934 .map(|run| run.frame_buffer_bytes_per_operation)
1935 .unwrap_or(0);
1936 let operations_per_sample = runs
1937 .first()
1938 .map(|run| run.operations_per_sample)
1939 .unwrap_or(0);
1940
1941 json!({
1942 "samples_ns": elapsed_samples,
1943 "min_ns": min_ns,
1944 "median_ns": median_ns,
1945 "max_ns": max_ns,
1946 "mean_ns": mean_ns,
1947 "explicit_fresh_buffer_allocations_per_sample": explicit_fresh_buffer_allocations,
1948 "scratch_capacity_growth_events_per_sample": scratch_capacity_growth_events,
1949 "peak_scratch_capacity_bytes": peak_scratch_capacity_bytes,
1950 "frame_buffer_bytes_per_operation": frame_buffer_bytes_per_operation,
1951 "operations_per_sample": operations_per_sample,
1952 "frame_buffer_bytes_requested_per_sample": explicit_fresh_buffer_allocations
1953 .saturating_mul(frame_buffer_bytes_per_operation),
1954 })
1955 }
1956
1957 fn append_frames_fresh_alloc<F: VfsFile>(
1958 wal: &mut WalFile<F>,
1959 cx: &Cx,
1960 frames: &[WalAppendFrameRef<'_>],
1961 ) -> Result<()> {
1962 let frame_size = wal.frame_size();
1963 let mut frame_buf = wal.prepare_frame_bytes(frames)?;
1964 let frame_transforms = frame_buf
1965 .chunks_exact(frame_size)
1966 .map(|frame| {
1967 WalChecksumTransform::for_wal_frame(
1968 frame,
1969 wal.page_size(),
1970 wal.big_endian_checksum(),
1971 )
1972 })
1973 .collect::<Result<Vec<_>>>()?;
1974 wal.append_prepared_frame_bytes(cx, &mut frame_buf, &frame_transforms)
1975 .wait()
1976 }
1977
1978 fn track_c_measure_single_frame_case(
1979 mode: TrackCScratchBenchMode,
1980 operations: usize,
1981 ) -> TrackCScratchBenchRun {
1982 let cx = test_cx();
1983 let vfs = MemoryVfs::new();
1984 let file = open_wal_file(&vfs, &cx);
1985 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create");
1986 let pages: Vec<Vec<u8>> = (0..operations)
1987 .map(|i| sample_page(u8::try_from(i % 251).expect("modulo fits u8")))
1988 .collect();
1989 let frame_buffer_bytes_per_operation = wal.frame_size();
1990 let mut explicit_fresh_buffer_allocations = 0usize;
1991 let mut scratch_capacity_growth_events = 0usize;
1992 let mut previous_scratch_capacity = 0usize;
1993
1994 let start = Instant::now();
1995 for (idx, page) in pages.iter().enumerate() {
1996 let page_number = u32::try_from(idx).expect("index fits u32") + 1;
1997 match mode {
1998 TrackCScratchBenchMode::FreshAllocBaseline => {
1999 let frames = [WalAppendFrameRef {
2000 page_number,
2001 page_data: page,
2002 db_size_if_commit: page_number,
2003 }];
2004 append_frames_fresh_alloc(&mut wal, &cx, &frames).expect("append baseline");
2005 explicit_fresh_buffer_allocations += 1;
2006 }
2007 TrackCScratchBenchMode::ScratchReuseCandidate => {
2008 wal.append_frame(&cx, page_number, page, page_number)
2009 .expect("append candidate");
2010 let scratch_capacity = wal.frame_scratch_capacity();
2011 if scratch_capacity > previous_scratch_capacity {
2012 scratch_capacity_growth_events += 1;
2013 previous_scratch_capacity = scratch_capacity;
2014 }
2015 }
2016 }
2017 }
2018
2019 TrackCScratchBenchRun {
2020 elapsed_ns: u64::try_from(start.elapsed().as_nanos()).unwrap_or(u64::MAX),
2021 explicit_fresh_buffer_allocations,
2022 scratch_capacity_growth_events,
2023 peak_scratch_capacity_bytes: wal.frame_scratch_capacity(),
2024 frame_buffer_bytes_per_operation,
2025 operations_per_sample: operations,
2026 }
2027 }
2028
2029 fn track_c_measure_batch_case<const N: usize>(
2030 mode: TrackCScratchBenchMode,
2031 operations: usize,
2032 ) -> TrackCScratchBenchRun {
2033 let cx = test_cx();
2034 let vfs = MemoryVfs::new();
2035 let file = open_wal_file(&vfs, &cx);
2036 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create");
2037 let pages_per_operation: Vec<[Vec<u8>; N]> = (0..operations)
2038 .map(|operation_idx| {
2039 std::array::from_fn(|frame_idx| {
2040 sample_page(
2041 u8::try_from((operation_idx * N + frame_idx) % 251)
2042 .expect("modulo fits u8"),
2043 )
2044 })
2045 })
2046 .collect();
2047 let frame_buffer_bytes_per_operation = wal
2048 .frame_size()
2049 .checked_mul(N)
2050 .expect("frame bytes per operation fit usize");
2051 let mut explicit_fresh_buffer_allocations = 0usize;
2052 let mut scratch_capacity_growth_events = 0usize;
2053 let mut previous_scratch_capacity = 0usize;
2054
2055 let start = Instant::now();
2056 for (operation_idx, pages) in pages_per_operation.iter().enumerate() {
2057 let page_base = u32::try_from(
2058 operation_idx
2059 .checked_mul(N)
2060 .expect("operation frame base fits usize"),
2061 )
2062 .expect("frame base fits u32")
2063 + 1;
2064 let commit_db_size = page_base + u32::try_from(N).expect("N fits u32") - 1;
2065 let frames: [WalAppendFrameRef<'_>; N] =
2066 std::array::from_fn(|frame_idx| WalAppendFrameRef {
2067 page_number: page_base
2068 + u32::try_from(frame_idx).expect("frame index fits u32"),
2069 page_data: &pages[frame_idx],
2070 db_size_if_commit: if frame_idx + 1 == N {
2071 commit_db_size
2072 } else {
2073 0
2074 },
2075 });
2076
2077 match mode {
2078 TrackCScratchBenchMode::FreshAllocBaseline => {
2079 append_frames_fresh_alloc(&mut wal, &cx, &frames).expect("append baseline");
2080 explicit_fresh_buffer_allocations += 1;
2081 }
2082 TrackCScratchBenchMode::ScratchReuseCandidate => {
2083 wal.append_frames(&cx, &frames).expect("append candidate");
2084 let scratch_capacity = wal.frame_scratch_capacity();
2085 if scratch_capacity > previous_scratch_capacity {
2086 scratch_capacity_growth_events += 1;
2087 previous_scratch_capacity = scratch_capacity;
2088 }
2089 }
2090 }
2091 }
2092
2093 TrackCScratchBenchRun {
2094 elapsed_ns: u64::try_from(start.elapsed().as_nanos()).unwrap_or(u64::MAX),
2095 explicit_fresh_buffer_allocations,
2096 scratch_capacity_growth_events,
2097 peak_scratch_capacity_bytes: wal.frame_scratch_capacity(),
2098 frame_buffer_bytes_per_operation,
2099 operations_per_sample: operations,
2100 }
2101 }
2102
2103 fn track_c_scratch_case_report(
2104 scenario_id: &str,
2105 frames_per_operation: usize,
2106 operations_per_sample: usize,
2107 baseline_measure: impl Fn() -> TrackCScratchBenchRun,
2108 candidate_measure: impl Fn() -> TrackCScratchBenchRun,
2109 ) -> Value {
2110 for _ in 0..TRACK_C_SCRATCH_BENCH_WARMUP_ITERS {
2111 let _ = baseline_measure();
2112 let _ = candidate_measure();
2113 }
2114
2115 let baseline_runs: Vec<_> = (0..TRACK_C_SCRATCH_BENCH_MEASURE_ITERS)
2116 .map(|_| baseline_measure())
2117 .collect();
2118 let candidate_runs: Vec<_> = (0..TRACK_C_SCRATCH_BENCH_MEASURE_ITERS)
2119 .map(|_| candidate_measure())
2120 .collect();
2121 let baseline_summary = track_c_scratch_run_summary(&baseline_runs);
2122 let candidate_summary = track_c_scratch_run_summary(&candidate_runs);
2123 let baseline_median = baseline_summary["median_ns"].as_u64().unwrap_or(0);
2124 let candidate_median = candidate_summary["median_ns"].as_u64().unwrap_or(0);
2125 let baseline_allocations = baseline_summary["explicit_fresh_buffer_allocations_per_sample"]
2126 .as_u64()
2127 .unwrap_or(0);
2128 let candidate_growths = candidate_summary["scratch_capacity_growth_events_per_sample"]
2129 .as_u64()
2130 .unwrap_or(0);
2131 let baseline_requested_bytes = baseline_summary["frame_buffer_bytes_requested_per_sample"]
2132 .as_u64()
2133 .unwrap_or(0);
2134 let candidate_peak_scratch_bytes = candidate_summary["peak_scratch_capacity_bytes"]
2135 .as_u64()
2136 .unwrap_or(0);
2137
2138 json!({
2139 "scenario_id": scenario_id,
2140 "frames_per_operation": frames_per_operation,
2141 "operations_per_sample": operations_per_sample,
2142 "fresh_alloc_baseline": baseline_summary,
2143 "scratch_reuse_candidate": candidate_summary,
2144 "fresh_buffer_allocations_avoided_per_sample": baseline_allocations.saturating_sub(candidate_growths),
2145 "buffer_bytes_saved_vs_fresh_requested_per_sample": baseline_requested_bytes.saturating_sub(candidate_peak_scratch_bytes),
2146 "speedup_vs_baseline_median": if candidate_median == 0 {
2147 0.0
2148 } else {
2149 (baseline_median as f64) / (candidate_median as f64)
2150 },
2151 "faster_variant_by_median": if candidate_median <= baseline_median {
2152 "scratch_reuse_candidate"
2153 } else {
2154 "fresh_alloc_baseline"
2155 },
2156 })
2157 }
2158
2159 #[test]
2160 fn test_create_and_open_empty_wal() {
2161 let cx = test_cx();
2162 let vfs = MemoryVfs::new();
2163 let file = open_wal_file(&vfs, &cx);
2164
2165 let wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create WAL");
2166 assert_eq!(wal.frame_count(), 0);
2167 assert_eq!(wal.page_size(), usize::try_from(PAGE_SIZE).unwrap());
2168 assert!(!wal.big_endian_checksum());
2169 assert_eq!(wal.header().checkpoint_seq, 0);
2170 assert_eq!(wal.header().salts, test_salts());
2171
2172 wal.close(&cx).expect("close WAL");
2173
2174 let file2 = open_wal_file(&vfs, &cx);
2176 let wal2 = WalFile::open(&cx, file2).expect("open WAL");
2177 assert_eq!(wal2.frame_count(), 0);
2178 assert_eq!(wal2.header().salts, test_salts());
2179
2180 wal2.close(&cx).expect("close WAL");
2181 }
2182
2183 #[test]
2184 fn test_append_and_read_single_frame() {
2185 let cx = test_cx();
2186 let vfs = MemoryVfs::new();
2187 let file = open_wal_file(&vfs, &cx);
2188
2189 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 1, test_salts()).expect("create WAL");
2190
2191 let page = sample_page(0x42);
2192 wal.append_frame(&cx, 1, &page, 0).expect("append frame");
2193 assert_eq!(wal.frame_count(), 1);
2194
2195 let (header, data) = wal.read_frame(&cx, 0).expect("read frame");
2196 assert_eq!(header.page_number, 1);
2197 assert_eq!(header.db_size, 0);
2198 assert_eq!(header.salts, test_salts());
2199 assert_eq!(data, page);
2200
2201 wal.close(&cx).expect("close WAL");
2202 }
2203
2204 #[test]
2205 fn test_fault_hook_after_wal_append_returns_error_and_records_context() {
2206 let _guard = FAULT_TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner());
2207 crate::fault_hooks::clear();
2208
2209 let cx = test_cx();
2210 let vfs = MemoryVfs::new();
2211 let file = open_wal_file(&vfs, &cx);
2212 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 1, test_salts()).expect("create WAL");
2213 let page = sample_page(0x33);
2214 let frames = [WalAppendFrameRef {
2215 page_number: 1,
2216 page_data: &page,
2217 db_size_if_commit: 1,
2218 }];
2219
2220 crate::fault_hooks::arm_after_append(crate::fault_hooks::FaultHookArm::new(
2221 "bd-db300.7.2.2-after-append",
2222 "WAL-AFTER-APPEND",
2223 "wal_append_recovery",
2224 ));
2225
2226 let error = wal
2227 .append_frames(&cx, &frames)
2228 .expect_err("fault hook should force an error after append");
2229 assert!(
2230 error.to_string().contains("fault_inject:wal_after_append"),
2231 "fault error should identify the append hook: {error}"
2232 );
2233 assert_eq!(
2234 wal.frame_count(),
2235 1,
2236 "append should still have reached the WAL"
2237 );
2238
2239 wal.close(&cx).expect("close WAL");
2240 let reopened_file = open_wal_file(&vfs, &cx);
2241 let reopened = WalFile::open(&cx, reopened_file).expect("reopen WAL");
2242 assert_eq!(
2243 reopened.frame_count(),
2244 1,
2245 "reopened WAL should preserve the appended frame for later recovery checks"
2246 );
2247
2248 let records = crate::fault_hooks::take_records();
2249 assert_eq!(
2250 records.len(),
2251 1,
2252 "exactly one append fault should be recorded"
2253 );
2254 assert_eq!(records[0].point, "wal_after_append");
2255 assert_eq!(records[0].run_id, "bd-db300.7.2.2-after-append");
2256 assert_eq!(records[0].scenario_id, "WAL-AFTER-APPEND");
2257 assert_eq!(records[0].invariant_family, "wal_append_recovery");
2258 assert!(
2259 records[0].detail.contains("appended_frames=1"),
2260 "record should preserve append context: {}",
2261 records[0].detail
2262 );
2263
2264 crate::fault_hooks::clear();
2265 }
2266
2267 #[test]
2268 fn test_append_commit_frame() {
2269 let cx = test_cx();
2270 let vfs = MemoryVfs::new();
2271 let file = open_wal_file(&vfs, &cx);
2272
2273 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create WAL");
2274
2275 let page = sample_page(0x10);
2276 wal.append_frame(&cx, 5, &page, 10)
2277 .expect("append commit frame");
2278
2279 let header = wal.read_frame_header(&cx, 0).expect("read header");
2280 assert!(header.is_commit());
2281 assert_eq!(header.db_size, 10);
2282 assert_eq!(header.page_number, 5);
2283
2284 wal.close(&cx).expect("close WAL");
2285 }
2286
2287 #[test]
2288 fn test_multi_frame_checksum_chain() {
2289 let cx = test_cx();
2290 let vfs = MemoryVfs::new();
2291 let file = open_wal_file(&vfs, &cx);
2292
2293 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 3, test_salts()).expect("create WAL");
2294
2295 for i in 0..5u32 {
2297 let page = sample_page(u8::try_from(i).expect("fits"));
2298 let db_size = if i == 4 { 5 } else { 0 };
2299 wal.append_frame(&cx, i + 1, &page, db_size)
2300 .expect("append frame");
2301 }
2302 assert_eq!(wal.frame_count(), 5);
2303
2304 wal.close(&cx).expect("close WAL");
2305
2306 let file2 = open_wal_file(&vfs, &cx);
2308 let wal2 = WalFile::open(&cx, file2).expect("open WAL");
2309 assert_eq!(wal2.frame_count(), 5);
2310
2311 for i in 0..5u32 {
2313 let (header, data) = wal2
2314 .read_frame(&cx, usize::try_from(i).unwrap())
2315 .expect("read frame");
2316 assert_eq!(header.page_number, i + 1);
2317 let expected = sample_page(u8::try_from(i).expect("fits"));
2318 assert_eq!(data, expected);
2319 }
2320
2321 let last_header = wal2.read_frame_header(&cx, 4).expect("read header");
2323 assert!(last_header.is_commit());
2324 assert_eq!(last_header.db_size, 5);
2325
2326 wal2.close(&cx).expect("close WAL");
2327 }
2328
2329 #[test]
2330 fn test_last_commit_frame() {
2331 let cx = test_cx();
2332 let vfs = MemoryVfs::new();
2333 let file = open_wal_file(&vfs, &cx);
2334
2335 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create WAL");
2336
2337 assert_eq!(wal.last_commit_frame(&cx).expect("query"), None);
2339
2340 wal.append_frame(&cx, 1, &sample_page(1), 0)
2342 .expect("append");
2343 assert_eq!(wal.last_commit_frame(&cx).expect("query"), None);
2344
2345 wal.append_frame(&cx, 2, &sample_page(2), 3)
2347 .expect("append");
2348 assert_eq!(wal.last_commit_frame(&cx).expect("query"), Some(1));
2349
2350 wal.append_frame(&cx, 3, &sample_page(3), 0)
2352 .expect("append");
2353 wal.append_frame(&cx, 4, &sample_page(4), 5)
2354 .expect("append");
2355 assert_eq!(wal.last_commit_frame(&cx).expect("query"), Some(3));
2356
2357 wal.close(&cx).expect("close WAL");
2358 }
2359
2360 #[test]
2361 fn test_reset_clears_frames() {
2362 let cx = test_cx();
2363 let vfs = MemoryVfs::new();
2364 let file = open_wal_file(&vfs, &cx);
2365
2366 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create WAL");
2367
2368 for i in 0..3u8 {
2370 let db_size = if i == 2 { 3 } else { 0 };
2371 wal.append_frame(&cx, u32::from(i) + 1, &sample_page(i), db_size)
2372 .expect("append");
2373 }
2374 assert_eq!(wal.frame_count(), 3);
2375
2376 let new_salts = WalSalts {
2378 salt1: 0x1111_2222,
2379 salt2: 0x3333_4444,
2380 };
2381 wal.reset(&cx, 1, new_salts, true).expect("reset");
2382 assert_eq!(wal.frame_count(), 0);
2383 assert_eq!(wal.last_commit_frame(&cx).expect("query"), None);
2384 assert_eq!(wal.header().checkpoint_seq, 1);
2385 assert_eq!(wal.header().salts, new_salts);
2386
2387 wal.append_frame(&cx, 10, &sample_page(0xAA), 1)
2389 .expect("append after reset");
2390 assert_eq!(wal.frame_count(), 1);
2391 assert_eq!(wal.last_commit_frame(&cx).expect("query"), Some(0));
2392
2393 wal.close(&cx).expect("close WAL");
2394
2395 let file2 = open_wal_file(&vfs, &cx);
2397 let wal2 = WalFile::open(&cx, file2).expect("open WAL");
2398 assert_eq!(wal2.frame_count(), 1);
2399 assert_eq!(wal2.header().checkpoint_seq, 1);
2400 assert_eq!(wal2.header().salts, new_salts);
2401
2402 wal2.close(&cx).expect("close WAL");
2403 }
2404
2405 #[test]
2406 fn test_page_size_mismatch_rejected() {
2407 let cx = test_cx();
2408 let vfs = MemoryVfs::new();
2409 let file = open_wal_file(&vfs, &cx);
2410
2411 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create WAL");
2412
2413 let short_page = vec![0u8; 100];
2415 let result = wal.append_frame(&cx, 1, &short_page, 0);
2416 assert!(result.is_err());
2417
2418 let long_page = vec![0u8; 8192];
2419 let result = wal.append_frame(&cx, 1, &long_page, 0);
2420 assert!(result.is_err());
2421
2422 wal.close(&cx).expect("close WAL");
2423 }
2424
2425 #[test]
2426 fn test_frame_index_out_of_range() {
2427 let cx = test_cx();
2428 let vfs = MemoryVfs::new();
2429 let file = open_wal_file(&vfs, &cx);
2430
2431 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create WAL");
2432
2433 assert!(wal.read_frame(&cx, 0).is_err());
2435 assert!(wal.read_frame_header(&cx, 0).is_err());
2436
2437 wal.append_frame(&cx, 1, &sample_page(0), 0)
2439 .expect("append");
2440 assert!(wal.read_frame(&cx, 0).is_ok());
2441 assert!(wal.read_frame(&cx, 1).is_err());
2442
2443 wal.close(&cx).expect("close WAL");
2444 }
2445
2446 #[test]
2447 fn test_reopen_preserves_checksum_chain() {
2448 let cx = test_cx();
2449 let vfs = MemoryVfs::new();
2450 let file = open_wal_file(&vfs, &cx);
2451
2452 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create WAL");
2453
2454 for i in 0..3u8 {
2456 let db_size = if i == 2 { 3 } else { 0 };
2457 wal.append_frame(&cx, u32::from(i) + 1, &sample_page(i), db_size)
2458 .expect("append");
2459 }
2460 let checksum_after_3 = wal.running_checksum();
2461 wal.close(&cx).expect("close WAL");
2462
2463 let file2 = open_wal_file(&vfs, &cx);
2465 let mut wal2 = WalFile::open(&cx, file2).expect("open WAL");
2466 assert_eq!(wal2.frame_count(), 3);
2467 assert_eq!(wal2.running_checksum(), checksum_after_3);
2468
2469 wal2.append_frame(&cx, 4, &sample_page(3), 0)
2470 .expect("append");
2471 wal2.append_frame(&cx, 5, &sample_page(4), 5)
2472 .expect("append commit");
2473 assert_eq!(wal2.frame_count(), 5);
2474 wal2.close(&cx).expect("close WAL");
2475
2476 let file3 = open_wal_file(&vfs, &cx);
2478 let wal3 = WalFile::open(&cx, file3).expect("open WAL");
2479 assert_eq!(wal3.frame_count(), 5);
2480 wal3.close(&cx).expect("close WAL");
2481 }
2482
2483 #[test]
2484 fn sync_records_fsynced_frame_count_for_production_flags() {
2485 let cx = test_cx();
2486 let vfs = MemoryVfs::new();
2487 let file = open_wal_file(&vfs, &cx);
2488
2489 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create WAL");
2490 wal.append_frame(&cx, 1, &sample_page(0), 1)
2491 .expect("append");
2492 wal.sync(&cx, SyncFlags::NORMAL).expect("sync");
2493 assert_eq!(wal.last_fsynced_frame_count(), 1);
2494
2495 wal.append_frame(&cx, 2, &sample_page(1), 2)
2496 .expect("append");
2497 wal.sync(&cx, SyncFlags::FULL).expect("full sync");
2498 assert_eq!(wal.last_fsynced_frame_count(), 2);
2499
2500 wal.append_frame(&cx, 3, &sample_page(2), 3)
2501 .expect("append");
2502 wal.sync(&cx, SyncFlags::DATAONLY).expect("data-only sync");
2503 assert_eq!(wal.last_fsynced_frame_count(), 3);
2504
2505 wal.close(&cx).expect("close WAL");
2506 }
2507
2508 #[test]
2509 fn test_fault_hook_sync_failure_returns_error_and_records_context() {
2510 let _guard = FAULT_TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner());
2511 crate::fault_hooks::clear();
2512
2513 let cx = test_cx();
2514 let vfs = MemoryVfs::new();
2515 let file = open_wal_file(&vfs, &cx);
2516 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 1, test_salts()).expect("create WAL");
2517 wal.append_frame(&cx, 1, &sample_page(0x44), 1)
2518 .expect("append frame");
2519 wal.sync(&cx, SyncFlags::NORMAL)
2520 .expect("establish durable watermark");
2521 assert_eq!(wal.last_fsynced_frame_count(), 1);
2522 wal.append_frame(&cx, 2, &sample_page(0x45), 2)
2523 .expect("append unsynced frame");
2524
2525 crate::fault_hooks::arm_sync_failure(crate::fault_hooks::FaultHookArm::new(
2526 "bd-db300.7.2.2-sync-failure",
2527 "WAL-SYNC-FAILURE",
2528 "wal_sync_recovery",
2529 ));
2530
2531 let error = wal
2532 .sync(&cx, SyncFlags::NORMAL)
2533 .expect_err("fault hook should force sync failure");
2534 assert!(
2535 error.to_string().contains("fault_inject:wal_sync_failure"),
2536 "fault error should identify the sync hook: {error}"
2537 );
2538 assert_eq!(
2539 wal.last_fsynced_frame_count(),
2540 1,
2541 "a failed sync must preserve the prior durable-barrier accounting"
2542 );
2543
2544 let records = crate::fault_hooks::take_records();
2545 assert_eq!(
2546 records.len(),
2547 1,
2548 "exactly one sync fault should be recorded"
2549 );
2550 assert_eq!(records[0].point, "wal_sync_failure");
2551 assert_eq!(records[0].run_id, "bd-db300.7.2.2-sync-failure");
2552 assert!(
2553 records[0].detail.contains("frame_count_before=2"),
2554 "record should capture sync context: {}",
2555 records[0].detail
2556 );
2557
2558 crate::fault_hooks::clear();
2559 }
2560
2561 #[test]
2562 fn test_fault_hook_append_busy_countdown_fires_once_and_preserves_retry_surface() {
2563 let _guard = FAULT_TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner());
2564 crate::fault_hooks::clear();
2565
2566 let cx = test_cx();
2567 let vfs = MemoryVfs::new();
2568 let file = open_wal_file(&vfs, &cx);
2569 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 1, test_salts()).expect("create WAL");
2570
2571 crate::fault_hooks::arm_append_busy_countdown(
2572 crate::fault_hooks::FaultHookArm::new(
2573 "bd-db300.7.2.2-busy-countdown",
2574 "WAL-APPEND-BUSY",
2575 "wal_append_retry",
2576 ),
2577 2,
2578 );
2579
2580 let first_page = sample_page(0x55);
2581 let first_frames = [WalAppendFrameRef {
2582 page_number: 1,
2583 page_data: &first_page,
2584 db_size_if_commit: 1,
2585 }];
2586 wal.append_frames(&cx, &first_frames)
2587 .expect("countdown should not fire on first append");
2588
2589 let second_page = sample_page(0x66);
2590 let second_frames = [WalAppendFrameRef {
2591 page_number: 2,
2592 page_data: &second_page,
2593 db_size_if_commit: 2,
2594 }];
2595 let busy = wal
2596 .append_frames(&cx, &second_frames)
2597 .expect_err("countdown should fire on second append");
2598 assert!(matches!(busy, FrankenError::Busy));
2599 assert_eq!(
2600 wal.frame_count(),
2601 1,
2602 "busy fault should fire before the second append mutates WAL state"
2603 );
2604
2605 wal.append_frames(&cx, &second_frames)
2606 .expect("hook should disarm after firing once");
2607 assert_eq!(
2608 wal.frame_count(),
2609 2,
2610 "retry should succeed once the hook is spent"
2611 );
2612
2613 let records = crate::fault_hooks::take_records();
2614 assert_eq!(records.len(), 1, "busy countdown should record one trigger");
2615 assert_eq!(records[0].point, "wal_append_busy_countdown");
2616 assert_eq!(records[0].run_id, "bd-db300.7.2.2-busy-countdown");
2617 assert!(
2618 records[0].detail.contains("submitted_frames=1"),
2619 "record should preserve append batch context: {}",
2620 records[0].detail
2621 );
2622
2623 crate::fault_hooks::clear();
2624 }
2625
2626 #[test]
2637 fn test_fault_crash_between_header_and_truncate_recovers_to_zero_frames() {
2638 let _guard = FAULT_TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner());
2639 crate::fault_hooks::clear();
2640
2641 let cx = test_cx();
2642 let vfs = MemoryVfs::new();
2643 let file = open_wal_file(&vfs, &cx);
2644 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create WAL");
2645
2646 for i in 1..=3_u32 {
2648 let page = sample_page(i as u8);
2649 wal.append_frame(&cx, i, &page, i)
2650 .expect("append frame before reset fault injection");
2651 }
2652 wal.sync(&cx, SyncFlags::NORMAL).expect("sync WAL");
2653 assert_eq!(wal.frame_count(), 3, "pre-reset: 3 frames");
2654
2655 let original_salts = wal.generation_identity().salts;
2656
2657 crate::fault_hooks::arm_crash_header_truncate(crate::fault_hooks::FaultHookArm::new(
2659 "bd-db300.7.2.2-h9",
2660 "WAL-CRASH-HEADER-TRUNCATE",
2661 "wal_reset_recovery",
2662 ));
2663
2664 let new_salts = WalSalts {
2666 salt1: original_salts.salt1.wrapping_add(1),
2667 salt2: original_salts.salt2.wrapping_add(1),
2668 };
2669 let err = wal
2670 .reset(&cx, 1, new_salts, true)
2671 .expect_err("fault hook should fire between header write and truncate");
2672 assert!(
2673 err.to_string()
2674 .contains("fault_inject:wal_crash_header_truncate"),
2675 "error should identify the hook: {err}"
2676 );
2677
2678 wal.close(&cx).expect("close WAL handle");
2683
2684 let recovered_file = open_wal_file(&vfs, &cx);
2686 let recovered = WalFile::open(&cx, recovered_file).expect("reopen WAL");
2687
2688 assert_eq!(
2691 recovered.frame_count(),
2692 0,
2693 "recovery must discard all old-salt frames after header rewrite"
2694 );
2695 assert_eq!(
2696 recovered.generation_identity().salts,
2697 new_salts,
2698 "recovered header must have the new salts"
2699 );
2700
2701 let records = crate::fault_hooks::take_records();
2703 assert_eq!(records.len(), 1, "exactly one crash hook should fire");
2704 assert_eq!(records[0].point, "wal_crash_header_truncate");
2705 assert_eq!(records[0].scenario_id, "WAL-CRASH-HEADER-TRUNCATE");
2706 assert!(
2707 records[0].detail.contains("old_frame_count=3"),
2708 "record should capture pre-reset frame count: {}",
2709 records[0].detail
2710 );
2711 assert!(
2712 records[0].detail.contains("new_checkpoint_seq=1"),
2713 "record should capture checkpoint seq: {}",
2714 records[0].detail
2715 );
2716
2717 crate::fault_hooks::clear();
2718 }
2719
2720 #[test]
2721 fn test_file_accessors() {
2722 let cx = test_cx();
2723 let vfs = MemoryVfs::new();
2724 let file = open_wal_file(&vfs, &cx);
2725
2726 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create WAL");
2727
2728 let _size = wal.file().file_size(&cx).expect("file_size");
2730 let _size = wal.file_mut().file_size(&cx).expect("file_size via mut");
2731
2732 wal.close(&cx).expect("close WAL");
2733 }
2734
2735 #[test]
2738 fn test_truncated_wal_recovers_committed_prefix() {
2739 let cx = test_cx();
2743 let vfs = MemoryVfs::new();
2744 let file = open_wal_file(&vfs, &cx);
2745
2746 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create WAL");
2747 for i in 0..5u8 {
2748 let db_size = if i == 2 {
2749 3
2750 } else if i == 4 {
2751 5
2752 } else {
2753 0
2754 };
2755 wal.append_frame(&cx, u32::from(i) + 1, &sample_page(i), db_size)
2756 .expect("append");
2757 }
2758 assert_eq!(wal.frame_count(), 5);
2759
2760 let frame_size = wal.frame_size();
2762 let truncate_at = WAL_HEADER_SIZE + frame_size * 3 + frame_size / 2;
2764 let truncate_at_u64 = u64::try_from(truncate_at).expect("truncate_at fits u64");
2765 wal.file_mut()
2766 .truncate(&cx, truncate_at_u64)
2767 .expect("truncate");
2768 wal.close(&cx).expect("close WAL");
2769
2770 let file2 = open_wal_file(&vfs, &cx);
2772 let wal2 = WalFile::open(&cx, file2).expect("open WAL after truncation");
2773 assert_eq!(
2774 wal2.frame_count(),
2775 3,
2776 "only the 3 complete frames before truncation should survive"
2777 );
2778
2779 for i in 0..3u8 {
2781 let (header, data) = wal2.read_frame(&cx, usize::from(i)).expect("read frame");
2782 assert_eq!(header.page_number, u32::from(i) + 1);
2783 assert_eq!(data, sample_page(i));
2784 }
2785 wal2.close(&cx).expect("close WAL");
2786 }
2787
2788 #[test]
2789 fn test_corrupt_frame_payload_detected_on_reopen() {
2790 let cx = test_cx();
2794 let vfs = MemoryVfs::new();
2795 let file = open_wal_file(&vfs, &cx);
2796
2797 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create WAL");
2798 for i in 0..5u8 {
2799 let db_size = if i == 2 {
2800 3
2801 } else if i == 4 {
2802 5
2803 } else {
2804 0
2805 };
2806 wal.append_frame(&cx, u32::from(i) + 1, &sample_page(i), db_size)
2807 .expect("append");
2808 }
2809 let frame_size = wal.frame_size();
2810 wal.close(&cx).expect("close WAL");
2811
2812 let corrupt_offset = WAL_HEADER_SIZE + frame_size * 3 + WAL_FRAME_HEADER_SIZE + 42;
2814 let corrupt_offset_u64 = u64::try_from(corrupt_offset).expect("corrupt_offset fits u64");
2815 let f = open_wal_file(&vfs, &cx);
2816 let mut buf = [0u8; 1];
2817 f.read(&cx, &mut buf, corrupt_offset_u64)
2818 .expect("read byte");
2819 buf[0] ^= 0xFF;
2820 f.write(&cx, &buf, corrupt_offset_u64)
2821 .expect("write corrupted byte");
2822 drop(f);
2823
2824 let file3 = open_wal_file(&vfs, &cx);
2826 let wal3 = WalFile::open(&cx, file3).expect("open WAL after corruption");
2827 assert_eq!(
2828 wal3.frame_count(),
2829 3,
2830 "frames after corruption point should be discarded"
2831 );
2832 wal3.close(&cx).expect("close WAL");
2833 }
2834
2835 #[test]
2836 fn test_multi_commit_recovery_to_last_valid() {
2837 let cx = test_cx();
2841 let vfs = MemoryVfs::new();
2842 let file = open_wal_file(&vfs, &cx);
2843
2844 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create WAL");
2845
2846 for i in 1..=3u32 {
2848 let db_size = if i == 3 { 3 } else { 0 };
2849 wal.append_frame(
2850 &cx,
2851 i,
2852 &sample_page(u8::try_from(i).expect("i fits u8")),
2853 db_size,
2854 )
2855 .expect("append");
2856 }
2857
2858 for i in 4..=6u32 {
2860 let db_size = if i == 6 { 6 } else { 0 };
2861 wal.append_frame(
2862 &cx,
2863 i,
2864 &sample_page(u8::try_from(i).expect("i fits u8")),
2865 db_size,
2866 )
2867 .expect("append");
2868 }
2869 assert_eq!(wal.frame_count(), 6);
2870 let frame_size = wal.frame_size();
2871 wal.close(&cx).expect("close WAL");
2872
2873 let corrupt_offset = WAL_HEADER_SIZE + frame_size * 4 + WAL_FRAME_HEADER_SIZE + 10;
2875 let corrupt_offset_u64 = u64::try_from(corrupt_offset).expect("corrupt_offset fits u64");
2876 let f = open_wal_file(&vfs, &cx);
2877 let mut buf = [0u8; 1];
2878 f.read(&cx, &mut buf, corrupt_offset_u64).expect("read");
2879 buf[0] ^= 0xAA;
2880 f.write(&cx, &buf, corrupt_offset_u64).expect("corrupt");
2881 drop(f);
2882
2883 let file2 = open_wal_file(&vfs, &cx);
2886 let wal2 = WalFile::open(&cx, file2).expect("open WAL after corruption");
2887 assert_eq!(
2888 wal2.frame_count(),
2889 3,
2890 "chain should break at corrupted frame 5, keeping committed prefix (frames 1-3)"
2891 );
2892
2893 let header3 = wal2.read_frame_header(&cx, 2).expect("read frame 3 header");
2895 assert!(header3.is_commit(), "frame 3 should be a commit frame");
2896
2897 wal2.close(&cx).expect("close WAL");
2898 }
2899
2900 #[test]
2901 fn test_wal_growth_bounded_by_restart_checkpoint() {
2902 use crate::checkpoint::{CheckpointMode, CheckpointState};
2905 use crate::checkpoint_executor::execute_checkpoint;
2906 use crate::checkpoint_executor::{CheckpointTarget, CheckpointTargetFuture};
2907 use fsqlite_types::PageNumber;
2908
2909 struct DummyTarget;
2910 impl CheckpointTarget for DummyTarget {
2911 fn write_page<'a>(
2912 &'a mut self,
2913 _: &'a Cx,
2914 _: PageNumber,
2915 _: &'a [u8],
2916 ) -> CheckpointTargetFuture<'a, ()> {
2917 Box::pin(async { Ok(()) })
2918 }
2919 fn truncate_db<'a>(&'a mut self, _: &'a Cx, _: u32) -> CheckpointTargetFuture<'a, ()> {
2920 Box::pin(async { Ok(()) })
2921 }
2922 fn sync_db<'a>(&'a mut self, _: &'a Cx) -> CheckpointTargetFuture<'a, ()> {
2923 Box::pin(async { Ok(()) })
2924 }
2925 }
2926
2927 let cx = test_cx();
2928 let vfs = MemoryVfs::new();
2929 let file = open_wal_file(&vfs, &cx);
2930 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create WAL");
2931
2932 for i in 1..=100u32 {
2934 let seed = u8::try_from(i % 256).expect("seed fits u8");
2935 let db_size = if i % 10 == 0 { i } else { 0 };
2936 wal.append_frame(&cx, (i - 1) % 50 + 1, &sample_page(seed), db_size)
2937 .expect("append");
2938 }
2939 assert_eq!(wal.frame_count(), 100);
2940
2941 let state = CheckpointState {
2943 total_frames: 100,
2944 backfilled_frames: 0,
2945 oldest_reader_frame: None,
2946 };
2947 let mut target = DummyTarget;
2948 let result = execute_checkpoint(&cx, &mut wal, CheckpointMode::Restart, state, &mut target)
2949 .expect("restart checkpoint");
2950
2951 assert_eq!(result.frames_backfilled, 100);
2952 assert!(result.wal_was_reset);
2953 assert_eq!(wal.frame_count(), 0, "WAL should be empty after restart");
2954
2955 wal.append_frame(&cx, 1, &sample_page(0xAA), 1)
2957 .expect("append after reset");
2958 assert_eq!(wal.frame_count(), 1);
2959 assert_eq!(wal.header().checkpoint_seq, 1, "checkpoint_seq incremented");
2960
2961 wal.close(&cx).expect("close WAL");
2962 }
2963
2964 #[test]
2965 fn test_wal_header_corruption_detected() {
2966 let cx = test_cx();
2969 let vfs = MemoryVfs::new();
2970 let file = open_wal_file(&vfs, &cx);
2971
2972 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create WAL");
2973 wal.append_frame(&cx, 1, &sample_page(1), 1)
2974 .expect("append");
2975 wal.close(&cx).expect("close WAL");
2976
2977 let f = open_wal_file(&vfs, &cx);
2979 let corrupted_magic = [0xFF, 0xFF, 0xFF, 0xFF];
2980 f.write(&cx, &corrupted_magic, 0).expect("corrupt header");
2981 drop(f);
2982
2983 let file2 = open_wal_file(&vfs, &cx);
2985 let result = WalFile::open(&cx, file2);
2986 assert!(
2987 result.is_err(),
2988 "opening WAL with corrupted header magic should fail"
2989 );
2990 }
2991
2992 #[test]
2993 fn test_empty_wal_after_crash_reopen() {
2994 let cx = test_cx();
2997 let vfs = MemoryVfs::new();
2998 let file = open_wal_file(&vfs, &cx);
2999
3000 let wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create WAL");
3001 wal.close(&cx).expect("close WAL");
3002
3003 let file2 = open_wal_file(&vfs, &cx);
3004 let wal2 = WalFile::open(&cx, file2).expect("reopen empty WAL");
3005 assert_eq!(wal2.frame_count(), 0);
3006 wal2.close(&cx).expect("close WAL");
3007 }
3008
3009 #[test]
3010 fn test_crash_after_single_uncommitted_frame() {
3011 let cx = test_cx();
3016 let vfs = MemoryVfs::new();
3017 let file = open_wal_file(&vfs, &cx);
3018
3019 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create WAL");
3020 wal.append_frame(&cx, 1, &sample_page(0x77), 0)
3021 .expect("append non-commit");
3022 wal.close(&cx).expect("close WAL");
3023
3024 let file2 = open_wal_file(&vfs, &cx);
3025 let wal2 = WalFile::open(&cx, file2).expect("reopen WAL");
3026 assert_eq!(
3027 wal2.frame_count(),
3028 0,
3029 "uncommitted frame excluded from recovery"
3030 );
3031 wal2.close(&cx).expect("close WAL");
3032 }
3033
3034 #[test]
3035 fn test_frame_offset_calculation_overflow_safety() {
3036 let page_size: u64 = 4096;
3040 let wal_header_size: u64 = 32;
3041 let wal_frame_header_size: u64 = 24;
3042 let frame_size = wal_frame_header_size + page_size;
3043
3044 let large_index: u64 = 1_042_468;
3050
3051 let idx_u64 = large_index;
3052 let expected_offset = wal_header_size + idx_u64 * frame_size;
3053
3054 let calculated_offset = wal_header_size + idx_u64 * frame_size;
3056
3057 assert_eq!(calculated_offset, expected_offset);
3058
3059 }
3062
3063 #[test]
3068 fn test_frame_offsets_sequential_no_gaps() {
3069 let cx = test_cx();
3072 let vfs = MemoryVfs::new();
3073 let file = open_wal_file(&vfs, &cx);
3074
3075 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create WAL");
3076
3077 let n = 20u32;
3078 for i in 0..n {
3079 let db_size = if i == n - 1 { n } else { 0 };
3080 wal.append_frame(
3081 &cx,
3082 i + 1,
3083 &sample_page(u8::try_from(i % 251).unwrap()),
3084 db_size,
3085 )
3086 .expect("append");
3087 }
3088
3089 let frame_size = wal.frame_size();
3090 let file_size = wal.file().file_size(&cx).expect("file_size");
3091 let expected_size =
3092 u64::try_from(WAL_HEADER_SIZE + usize::try_from(n).unwrap() * frame_size).unwrap();
3093 assert_eq!(
3094 file_size, expected_size,
3095 "WAL file size must equal header + n*frame_size with no padding or gaps"
3096 );
3097
3098 for i in 0..n {
3100 let header = wal
3101 .read_frame_header(&cx, usize::try_from(i).unwrap())
3102 .expect("read header");
3103 assert_eq!(header.page_number, i + 1, "frame {i} page_number");
3104 }
3105
3106 wal.close(&cx).expect("close WAL");
3107 }
3108
3109 #[test]
3110 fn test_checksum_determinism_same_input() {
3111 let cx = test_cx();
3114 let vfs1 = MemoryVfs::new();
3115 let vfs2 = MemoryVfs::new();
3116
3117 let mut checksums_a = Vec::new();
3118 let mut checksums_b = Vec::new();
3119
3120 for (vfs, checksums) in [(&vfs1, &mut checksums_a), (&vfs2, &mut checksums_b)] {
3121 let file = open_wal_file(vfs, &cx);
3122 let mut wal =
3123 WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create WAL");
3124
3125 for i in 0..10u8 {
3126 let page = sample_page(i);
3127 let db_size = if i == 9 { 10 } else { 0 };
3128 wal.append_frame(&cx, u32::from(i) + 1, &page, db_size)
3129 .expect("append");
3130 checksums.push(wal.running_checksum());
3131 }
3132 wal.close(&cx).expect("close WAL");
3133 }
3134
3135 assert_eq!(
3136 checksums_a, checksums_b,
3137 "identical inputs must produce identical checksum chains"
3138 );
3139 }
3140
3141 #[test]
3142 fn test_checksum_sensitivity_one_byte_difference() {
3143 let cx = test_cx();
3146 let vfs1 = MemoryVfs::new();
3147 let vfs2 = MemoryVfs::new();
3148
3149 let mut checksums_a = Vec::new();
3150 let mut checksums_b = Vec::new();
3151
3152 let file1 = open_wal_file(&vfs1, &cx);
3153 let mut wal1 = WalFile::create(&cx, file1, PAGE_SIZE, 0, test_salts()).expect("create");
3154 let file2 = open_wal_file(&vfs2, &cx);
3155 let mut wal2 = WalFile::create(&cx, file2, PAGE_SIZE, 0, test_salts()).expect("create");
3156
3157 for i in 0..5u8 {
3158 let mut page = sample_page(i);
3159 let db_size = if i == 4 { 5 } else { 0 };
3160 wal1.append_frame(&cx, u32::from(i) + 1, &page, db_size)
3161 .expect("append");
3162 checksums_a.push(wal1.running_checksum());
3163
3164 if i == 2 {
3166 page[0] ^= 0x01;
3167 }
3168 wal2.append_frame(&cx, u32::from(i) + 1, &page, db_size)
3169 .expect("append");
3170 checksums_b.push(wal2.running_checksum());
3171 }
3172
3173 assert_eq!(checksums_a[0], checksums_b[0], "frame 0 should match");
3175 assert_eq!(checksums_a[1], checksums_b[1], "frame 1 should match");
3176 assert_ne!(checksums_a[2], checksums_b[2], "frame 2 must diverge");
3177 assert_ne!(checksums_a[3], checksums_b[3], "frame 3 must diverge");
3178 assert_ne!(checksums_a[4], checksums_b[4], "frame 4 must diverge");
3179
3180 wal1.close(&cx).expect("close");
3181 wal2.close(&cx).expect("close");
3182 }
3183
3184 #[test]
3185 fn test_commit_boundary_every_frame() {
3186 let cx = test_cx();
3189 let vfs = MemoryVfs::new();
3190 let file = open_wal_file(&vfs, &cx);
3191
3192 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create");
3193
3194 let n = 8u32;
3195 for i in 0..n {
3196 wal.append_frame(&cx, i + 1, &sample_page(u8::try_from(i).unwrap()), i + 1)
3197 .expect("append");
3198 }
3199 assert_eq!(wal.frame_count(), usize::try_from(n).unwrap());
3200 wal.close(&cx).expect("close");
3201
3202 let file2 = open_wal_file(&vfs, &cx);
3203 let mut wal2 = WalFile::open(&cx, file2).expect("reopen");
3204 assert_eq!(
3205 wal2.frame_count(),
3206 usize::try_from(n).unwrap(),
3207 "all frames are commits so all should survive reopen"
3208 );
3209
3210 for i in 0..n {
3212 let h = wal2
3213 .read_frame_header(&cx, usize::try_from(i).unwrap())
3214 .expect("read");
3215 assert!(h.is_commit(), "frame {i} must be a commit");
3216 assert_eq!(h.db_size, i + 1);
3217 }
3218
3219 let last = wal2.last_commit_frame(&cx).expect("query");
3221 assert_eq!(last, Some(usize::try_from(n - 1).unwrap()));
3222
3223 wal2.close(&cx).expect("close");
3224 }
3225
3226 #[test]
3227 fn test_commit_boundary_interleaved_multi_txn() {
3228 let cx = test_cx();
3233 let vfs = MemoryVfs::new();
3234 let file = open_wal_file(&vfs, &cx);
3235 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create");
3236
3237 let frames: [(u32, u32); 6] = [
3238 (1, 0),
3239 (2, 0),
3240 (3, 3), (4, 0),
3242 (5, 5), (6, 6), ];
3245
3246 for (pg, db_sz) in frames {
3247 wal.append_frame(&cx, pg, &sample_page(u8::try_from(pg).unwrap()), db_sz)
3248 .expect("append");
3249 }
3250 assert_eq!(wal.frame_count(), 6);
3251 wal.close(&cx).expect("close");
3252
3253 let file2 = open_wal_file(&vfs, &cx);
3255 let mut wal2 = WalFile::open(&cx, file2).expect("reopen");
3256 assert_eq!(wal2.frame_count(), 6);
3257
3258 for i in 0..6u32 {
3260 let (header, data) = wal2
3261 .read_frame(&cx, usize::try_from(i).unwrap())
3262 .expect("read frame");
3263 assert_eq!(header.page_number, i + 1);
3264 let expected = sample_page(u8::try_from(i + 1).expect("fits"));
3265 assert_eq!(data, expected);
3266 }
3267
3268 let last_header = wal2.read_frame_header(&cx, 5).expect("read header");
3270 assert!(last_header.is_commit());
3271 assert_eq!(last_header.db_size, 6);
3272 let last = wal2.last_commit_frame(&cx).expect("query");
3273 assert_eq!(last, Some(5), "last commit is frame 6 (index 5)");
3274
3275 wal2.close(&cx).expect("close");
3276 }
3277
3278 #[test]
3279 fn test_same_page_overwritten_multiple_times() {
3280 let cx = test_cx();
3284 let vfs = MemoryVfs::new();
3285 let file = open_wal_file(&vfs, &cx);
3286 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create");
3287
3288 let page_num = 42u32;
3289 let versions = 5;
3290 for v in 0..versions {
3291 let db_size = if v == versions - 1 { 100 } else { 0 };
3292 wal.append_frame(&cx, page_num, &sample_page(v), db_size)
3293 .expect("append");
3294 }
3295 assert_eq!(wal.frame_count(), usize::from(versions));
3296
3297 for v in 0..versions {
3299 let (header, data) = wal.read_frame(&cx, usize::from(v)).expect("read frame");
3300 assert_eq!(header.page_number, page_num);
3301 assert_eq!(data, sample_page(v), "frame {v} data mismatch");
3302 }
3303
3304 wal.close(&cx).expect("close");
3305 }
3306
3307 #[test]
3308 fn test_refresh_detects_concurrent_append() {
3309 let cx = test_cx();
3312 let vfs = MemoryVfs::new();
3313 let file1 = open_wal_file(&vfs, &cx);
3314 let mut wal1 = WalFile::create(&cx, file1, PAGE_SIZE, 0, test_salts()).expect("create");
3315
3316 for i in 0..3u8 {
3318 let db_size = if i == 2 { 3 } else { 0 };
3319 wal1.append_frame(&cx, u32::from(i) + 1, &sample_page(i), db_size)
3320 .expect("append");
3321 }
3322 let checksum_after_3 = wal1.running_checksum();
3323 wal1.close(&cx).expect("close wal1");
3324
3325 let file_reader = open_wal_file(&vfs, &cx);
3327 let mut reader = WalFile::open(&cx, file_reader).expect("open reader");
3328 assert_eq!(reader.frame_count(), 3);
3329 assert_eq!(reader.last_commit_frame(&cx).expect("query"), Some(2));
3330 reader.sync(&cx, SyncFlags::NORMAL).expect("sync reader");
3331 assert_eq!(reader.last_fsynced_frame_count(), 3);
3332
3333 let file_w2 = open_wal_file(&vfs, &cx);
3335 let mut w2 = WalFile::open(&cx, file_w2).expect("open w2");
3336 assert_eq!(w2.running_checksum(), checksum_after_3);
3337 w2.append_frame(&cx, 4, &sample_page(3), 0).expect("append");
3338 w2.append_frame(&cx, 5, &sample_page(4), 5)
3339 .expect("append commit");
3340 assert_eq!(w2.frame_count(), 5);
3341 w2.close(&cx).expect("close w2");
3342
3343 assert_eq!(reader.frame_count(), 3);
3345 reader.refresh(&cx).expect("refresh");
3346 assert_eq!(
3347 reader.frame_count(),
3348 5,
3349 "after refresh, reader must see the 2 new committed frames"
3350 );
3351 assert_eq!(
3352 reader.last_fsynced_frame_count(),
3353 3,
3354 "same-generation incremental refresh preserves only the known watermark"
3355 );
3356 assert_eq!(reader.last_commit_frame(&cx).expect("query"), Some(4));
3357
3358 reader.close(&cx).expect("close reader");
3359 }
3360
3361 #[test]
3362 fn refresh_after_reset_clears_stale_durable_watermark() {
3363 let cx = test_cx();
3366 let vfs = MemoryVfs::new();
3367 let file = open_wal_file(&vfs, &cx);
3368 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create");
3369
3370 wal.append_frame(&cx, 1, &sample_page(1), 1)
3372 .expect("append");
3373 wal.close(&cx).expect("close");
3374
3375 let file_r = open_wal_file(&vfs, &cx);
3377 let mut reader = WalFile::open(&cx, file_r).expect("open reader");
3378 assert_eq!(reader.frame_count(), 1);
3379 assert_eq!(reader.last_commit_frame(&cx).expect("query"), Some(0));
3380 reader
3381 .sync(&cx, SyncFlags::NORMAL)
3382 .expect("sync old generation");
3383 assert_eq!(reader.last_fsynced_frame_count(), 1);
3384
3385 let file_cp = open_wal_file(&vfs, &cx);
3387 let mut cp = WalFile::open(&cx, file_cp).expect("open cp");
3388 let new_salts = WalSalts {
3389 salt1: 0xAAAA_BBBB,
3390 salt2: 0xCCCC_DDDD,
3391 };
3392 cp.reset(&cx, 1, new_salts, false).expect("reset");
3393 cp.append_frame(&cx, 1, &sample_page(0xAA), 1)
3394 .expect("append after reset");
3395 cp.close(&cx).expect("close cp");
3396
3397 reader.refresh(&cx).expect("refresh");
3399 assert_eq!(reader.frame_count(), 1);
3400 assert_eq!(
3401 reader.last_fsynced_frame_count(),
3402 0,
3403 "a new generation's unsynced frame must not inherit the old generation watermark"
3404 );
3405 assert_eq!(reader.last_commit_frame(&cx).expect("query"), Some(0));
3406 assert_eq!(
3407 reader.header().salts,
3408 new_salts,
3409 "salts should be new generation"
3410 );
3411
3412 reader.close(&cx).expect("close reader");
3413 }
3414
3415 #[test]
3416 fn test_refresh_after_reset_with_same_salts_detects_new_generation() {
3417 let cx = test_cx();
3420 let vfs = MemoryVfs::new();
3421 let file = open_wal_file(&vfs, &cx);
3422 let salts = test_salts();
3423 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, salts).expect("create");
3424
3425 wal.append_frame(&cx, 1, &sample_page(1), 1)
3426 .expect("append");
3427 wal.close(&cx).expect("close");
3428
3429 let file_r = open_wal_file(&vfs, &cx);
3430 let mut reader = WalFile::open(&cx, file_r).expect("open reader");
3431 let before = reader.generation_identity();
3432 assert_eq!(before.checkpoint_seq, 0);
3433 assert_eq!(before.salts, salts);
3434 assert_eq!(reader.frame_count(), 1);
3435
3436 let file_cp = open_wal_file(&vfs, &cx);
3437 let mut cp = WalFile::open(&cx, file_cp).expect("open cp");
3438 cp.reset(&cx, 1, salts, false)
3439 .expect("reset with same salts");
3440 cp.append_frame(&cx, 2, &sample_page(0xAA), 2)
3441 .expect("append after reset");
3442 cp.close(&cx).expect("close cp");
3443
3444 reader.refresh(&cx).expect("refresh");
3445 let after = reader.generation_identity();
3446 assert_eq!(
3447 after.checkpoint_seq, 1,
3448 "refresh must observe new checkpoint_seq"
3449 );
3450 assert_eq!(
3451 after.salts, salts,
3452 "same-salt reset is intentional in this test"
3453 );
3454 assert_ne!(
3455 before, after,
3456 "generation identity must change even when salts are reused"
3457 );
3458 assert_eq!(
3459 reader.frame_count(),
3460 1,
3461 "reader must rebuild to new generation"
3462 );
3463
3464 let (header, data) = reader.read_frame(&cx, 0).expect("read rebuilt frame");
3465 assert_eq!(
3466 header.page_number, 2,
3467 "reader must see new-generation frame"
3468 );
3469 assert_eq!(data, sample_page(0xAA));
3470
3471 reader.close(&cx).expect("close reader");
3472 }
3473
3474 #[test]
3475 fn test_group_commit_checksum_chain_matches_single_append() {
3476 use crate::group_commit::{
3479 FrameSubmission, TransactionFrameBatch, write_consolidated_frames,
3480 };
3481
3482 let cx = test_cx();
3483 let vfs_single = MemoryVfs::new();
3484 let vfs_group = MemoryVfs::new();
3485
3486 let pages: Vec<Vec<u8>> = (0..6u8).map(sample_page).collect();
3487 let page_nums: Vec<u32> = (1..=6u32).collect();
3488 let commit_sizes: Vec<u32> = vec![0, 0, 3, 0, 0, 6];
3490
3491 let file_s = open_wal_file(&vfs_single, &cx);
3493 let mut wal_s =
3494 WalFile::create(&cx, file_s, PAGE_SIZE, 0, test_salts()).expect("create single");
3495 for i in 0..6 {
3496 wal_s
3497 .append_frame(&cx, page_nums[i], &pages[i], commit_sizes[i])
3498 .expect("append single");
3499 }
3500 let single_checksum = wal_s.running_checksum();
3501 let single_count = wal_s.frame_count();
3502
3503 let file_g = open_wal_file(&vfs_group, &cx);
3505 let mut wal_g =
3506 WalFile::create(&cx, file_g, PAGE_SIZE, 0, test_salts()).expect("create group");
3507
3508 let batch1 = TransactionFrameBatch::new(
3509 (0..3)
3510 .map(|i| FrameSubmission {
3511 page_number: page_nums[i],
3512 page_data: pages[i].clone(),
3513 db_size_if_commit: commit_sizes[i],
3514 })
3515 .collect(),
3516 );
3517 let batch2 = TransactionFrameBatch::new(
3518 (3..6)
3519 .map(|i| FrameSubmission {
3520 page_number: page_nums[i],
3521 page_data: pages[i].clone(),
3522 db_size_if_commit: commit_sizes[i],
3523 })
3524 .collect(),
3525 );
3526
3527 write_consolidated_frames(&cx, &mut wal_g, &[batch1, batch2]).expect("group write");
3528 let group_checksum = wal_g.running_checksum();
3529 let group_count = wal_g.frame_count();
3530
3531 assert_eq!(single_count, group_count, "frame counts must match");
3532 assert_eq!(
3533 single_checksum, group_checksum,
3534 "group commit must produce identical checksum chain as single-frame append"
3535 );
3536
3537 for i in 0..6 {
3539 let (h_s, d_s) = wal_s.read_frame(&cx, i).expect("read single");
3540 let (h_g, d_g) = wal_g.read_frame(&cx, i).expect("read group");
3541 assert_eq!(h_s.page_number, h_g.page_number, "frame {i} page_number");
3542 assert_eq!(h_s.db_size, h_g.db_size, "frame {i} db_size");
3543 assert_eq!(h_s.checksum, h_g.checksum, "frame {i} checksum");
3544 assert_eq!(h_s.salts, h_g.salts, "frame {i} salts");
3545 assert_eq!(d_s, d_g, "frame {i} data");
3546 }
3547
3548 wal_s.close(&cx).expect("close single");
3549 wal_g.close(&cx).expect("close group");
3550 }
3551
3552 #[test]
3553 fn test_batch_append_checksum_chain_matches_single_append() {
3554 let cx = test_cx();
3555 let vfs_single = MemoryVfs::new();
3556 let vfs_batch = MemoryVfs::new();
3557
3558 let pages: Vec<Vec<u8>> = (0..6u8).map(sample_page).collect();
3559 let page_nums: Vec<u32> = (1..=6u32).collect();
3560 let commit_sizes: Vec<u32> = vec![0, 0, 3, 0, 0, 6];
3561
3562 let file_single = open_wal_file(&vfs_single, &cx);
3563 let mut wal_single =
3564 WalFile::create(&cx, file_single, PAGE_SIZE, 0, test_salts()).expect("create single");
3565 for i in 0..6 {
3566 wal_single
3567 .append_frame(&cx, page_nums[i], &pages[i], commit_sizes[i])
3568 .expect("append single");
3569 }
3570
3571 let file_batch = open_wal_file(&vfs_batch, &cx);
3572 let mut wal_batch =
3573 WalFile::create(&cx, file_batch, PAGE_SIZE, 0, test_salts()).expect("create batch");
3574 let frames: Vec<_> = (0..6)
3575 .map(|i| WalAppendFrameRef {
3576 page_number: page_nums[i],
3577 page_data: &pages[i],
3578 db_size_if_commit: commit_sizes[i],
3579 })
3580 .collect();
3581 wal_batch.append_frames(&cx, &frames).expect("append batch");
3582
3583 assert_eq!(
3584 wal_single.frame_count(),
3585 wal_batch.frame_count(),
3586 "batch append must preserve frame count"
3587 );
3588 assert_eq!(
3589 wal_single.running_checksum(),
3590 wal_batch.running_checksum(),
3591 "batch append must preserve checksum chain"
3592 );
3593
3594 for i in 0..6 {
3595 let (single_header, single_data) = wal_single.read_frame(&cx, i).expect("read single");
3596 let (batch_header, batch_data) = wal_batch.read_frame(&cx, i).expect("read batch");
3597 assert_eq!(single_header, batch_header, "frame header {i} must match");
3598 assert_eq!(single_data, batch_data, "frame payload {i} must match");
3599 }
3600 }
3601
3602 #[test]
3603 fn tracked_batch_write_completion_is_reported_by_memory_source() {
3604 let cx = test_cx();
3605 let vfs = MemoryVfs::new();
3606 let file = open_wal_file(&vfs, &cx);
3607 let mut wal =
3608 WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create tracked WAL");
3609 let page = sample_page(0xA5);
3610 let frames = [WalAppendFrameRef {
3611 page_number: 1,
3612 page_data: &page,
3613 db_size_if_commit: 1,
3614 }];
3615 let completion = VfsWriteCompletion::new();
3616
3617 wal.append_frames_tracked(&cx, &frames, completion.clone())
3618 .expect("append tracked frame batch");
3619
3620 assert_eq!(
3621 completion.state(),
3622 fsqlite_vfs::VfsWriteCompletionState::Success
3623 );
3624 assert_eq!(wal.frame_count(), 1);
3625 }
3626
3627 #[test]
3631 fn test_fused_append_frames_produces_byte_identical_wal_file() {
3632 let cx = test_cx();
3633 let vfs_single = MemoryVfs::new();
3634 let vfs_fused = MemoryVfs::new();
3635
3636 let pages: Vec<Vec<u8>> = (0..4u8).map(sample_page).collect();
3637 let page_nums: Vec<u32> = vec![3, 1, 4, 2];
3638 let commit_sizes: Vec<u32> = vec![0, 0, 0, 4];
3639
3640 let file_s = open_wal_file(&vfs_single, &cx);
3642 let mut wal_s =
3643 WalFile::create(&cx, file_s, PAGE_SIZE, 0, test_salts()).expect("create single");
3644 for i in 0..4 {
3645 wal_s
3646 .append_frame(&cx, page_nums[i], &pages[i], commit_sizes[i])
3647 .expect("append single");
3648 }
3649
3650 let file_f = open_wal_file(&vfs_fused, &cx);
3652 let mut wal_f =
3653 WalFile::create(&cx, file_f, PAGE_SIZE, 0, test_salts()).expect("create fused");
3654 let frames: Vec<_> = (0..4)
3655 .map(|i| WalAppendFrameRef {
3656 page_number: page_nums[i],
3657 page_data: &pages[i],
3658 db_size_if_commit: commit_sizes[i],
3659 })
3660 .collect();
3661 wal_f.append_frames(&cx, &frames).expect("append fused");
3662
3663 assert_eq!(wal_s.frame_count(), wal_f.frame_count());
3665 assert_eq!(wal_s.running_checksum(), wal_f.running_checksum());
3666
3667 let frame_size = wal_s.frame_size();
3668 for i in 0..4 {
3669 let mut buf_s = vec![0u8; frame_size];
3670 let mut buf_f = vec![0u8; frame_size];
3671 wal_s
3672 .read_frame_into(&cx, i, &mut buf_s)
3673 .expect("read single");
3674 wal_f
3675 .read_frame_into(&cx, i, &mut buf_f)
3676 .expect("read fused");
3677 assert_eq!(
3678 buf_s, buf_f,
3679 "raw frame bytes at index {i} must be identical"
3680 );
3681 }
3682 }
3683
3684 #[test]
3688 fn test_append_frames_restores_scratch_on_error() {
3689 let cx = test_cx();
3690 let vfs = MemoryVfs::new();
3691 let file = open_wal_file(&vfs, &cx);
3692 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create");
3693
3694 let good_page = sample_page(0x11);
3696 wal.append_frame(&cx, 1, &good_page, 0)
3697 .expect("first append");
3698 let checksum_before = wal.running_checksum();
3699 let scratch_cap_before = wal.frame_scratch_capacity();
3700 let mut first_frame_before = vec![0u8; wal.frame_size()];
3701 wal.read_frame_into(&cx, 0, &mut first_frame_before)
3702 .expect("read baseline frame");
3703
3704 let bad_page = vec![0xBBu8; PAGE_SIZE as usize + 1]; let good_page2 = sample_page(0x22);
3707 let bad_frames = vec![
3708 WalAppendFrameRef {
3709 page_number: 2,
3710 page_data: &good_page2,
3711 db_size_if_commit: 0,
3712 },
3713 WalAppendFrameRef {
3714 page_number: 3,
3715 page_data: &bad_page, db_size_if_commit: 3,
3717 },
3718 ];
3719 let err = wal.append_frames(&cx, &bad_frames);
3720 assert!(err.is_err(), "bad page size should cause error");
3721
3722 assert_eq!(
3724 wal.frame_count(),
3725 1,
3726 "failed append must not advance frame count"
3727 );
3728 assert_eq!(
3729 wal.running_checksum(),
3730 checksum_before,
3731 "failed append must preserve the running checksum of prior committed frames"
3732 );
3733 assert!(
3734 wal.frame_scratch_capacity() >= scratch_cap_before,
3735 "scratch capacity must not shrink after error"
3736 );
3737 let mut first_frame_after = vec![0u8; wal.frame_size()];
3738 wal.read_frame_into(&cx, 0, &mut first_frame_after)
3739 .expect("read preserved frame");
3740 assert_eq!(
3741 first_frame_after, first_frame_before,
3742 "failed append must not rewrite previously committed raw frame bytes"
3743 );
3744
3745 let recovery_page = sample_page(0x33);
3747 wal.append_frame(&cx, 2, &recovery_page, 2)
3748 .expect("recovery append after error");
3749 assert_eq!(wal.frame_count(), 2, "recovery append should succeed");
3750 }
3751
3752 #[test]
3753 fn test_append_frame_reuses_frame_scratch_between_calls() {
3754 let cx = test_cx();
3755 let vfs = MemoryVfs::new();
3756 let file = open_wal_file(&vfs, &cx);
3757 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create");
3758
3759 wal.append_frame(&cx, 1, &sample_page(0x11), 0)
3760 .expect("append first");
3761 let scratch_len = wal.frame_scratch_len();
3762 let scratch_capacity = wal.frame_scratch_capacity();
3763 let scratch_ptr = wal.frame_scratch_ptr();
3764
3765 wal.append_frame(&cx, 2, &sample_page(0x22), 2)
3766 .expect("append second");
3767
3768 assert_eq!(
3769 wal.frame_scratch_len(),
3770 wal.frame_size(),
3771 "single-frame append should keep one frame sized scratch"
3772 );
3773 assert_eq!(
3774 wal.frame_scratch_len(),
3775 scratch_len,
3776 "single-frame scratch length should stay constant across appends"
3777 );
3778 assert_eq!(
3779 wal.frame_scratch_capacity(),
3780 scratch_capacity,
3781 "single-frame scratch should retain its allocation"
3782 );
3783 assert_eq!(
3784 wal.frame_scratch_ptr(),
3785 scratch_ptr,
3786 "single-frame scratch should reuse the same backing buffer"
3787 );
3788 }
3789
3790 #[test]
3791 fn test_batch_append_reuses_frame_scratch_between_calls() {
3792 let cx = test_cx();
3793 let vfs = MemoryVfs::new();
3794 let file = open_wal_file(&vfs, &cx);
3795 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create");
3796
3797 let first_pages: Vec<Vec<u8>> = (0..3u8).map(sample_page).collect();
3798 let first_frames: Vec<_> = (0..3)
3799 .map(|i| WalAppendFrameRef {
3800 page_number: u32::try_from(i).expect("index fits u32") + 1,
3801 page_data: &first_pages[i],
3802 db_size_if_commit: 0,
3803 })
3804 .collect();
3805 wal.append_frames(&cx, &first_frames)
3806 .expect("append first batch");
3807 let scratch_capacity = wal.frame_scratch_capacity();
3808 let scratch_ptr = wal.frame_scratch_ptr();
3809
3810 let second_pages: Vec<Vec<u8>> = (0..2u8).map(|i| sample_page(i + 10)).collect();
3811 let second_frames: Vec<_> = (0..2)
3812 .map(|i| WalAppendFrameRef {
3813 page_number: u32::try_from(i).expect("index fits u32") + 10,
3814 page_data: &second_pages[i],
3815 db_size_if_commit: if i == 1 { 11 } else { 0 },
3816 })
3817 .collect();
3818 wal.append_frames(&cx, &second_frames)
3819 .expect("append second batch");
3820
3821 assert_eq!(
3822 wal.frame_scratch_len(),
3823 second_frames.len() * wal.frame_size(),
3824 "batch scratch length should track the active batch size"
3825 );
3826 assert_eq!(
3827 wal.frame_scratch_capacity(),
3828 scratch_capacity,
3829 "smaller follow-on batches should retain the existing scratch allocation"
3830 );
3831 assert_eq!(
3832 wal.frame_scratch_ptr(),
3833 scratch_ptr,
3834 "smaller follow-on batches should reuse the same backing buffer"
3835 );
3836 }
3837
3838 #[test]
3839 fn test_reset_clears_frame_scratch_len_without_dropping_capacity() {
3840 let cx = test_cx();
3841 let vfs = MemoryVfs::new();
3842 let file = open_wal_file(&vfs, &cx);
3843 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create");
3844 let reset_salts = WalSalts {
3845 salt1: 0xABCD_EF01,
3846 salt2: 0x1020_3040,
3847 };
3848
3849 let pages: Vec<Vec<u8>> = (0..4u8).map(sample_page).collect();
3850 let frames: Vec<_> = (0..4)
3851 .map(|i| WalAppendFrameRef {
3852 page_number: u32::try_from(i).expect("index fits u32") + 1,
3853 page_data: &pages[i],
3854 db_size_if_commit: if i == 3 { 4 } else { 0 },
3855 })
3856 .collect();
3857 wal.append_frames(&cx, &frames).expect("append batch");
3858 let scratch_capacity = wal.frame_scratch_capacity();
3859 let scratch_ptr = wal.frame_scratch_ptr();
3860
3861 wal.reset(&cx, 1, reset_salts, false).expect("reset WAL");
3862
3863 assert_eq!(
3864 wal.frame_scratch_len(),
3865 0,
3866 "reset should leave scratch empty for the next append cycle"
3867 );
3868 assert_eq!(
3869 wal.frame_scratch_capacity(),
3870 scratch_capacity,
3871 "reset should preserve scratch capacity for reuse"
3872 );
3873 assert_eq!(
3874 wal.frame_scratch_ptr(),
3875 scratch_ptr,
3876 "reset should keep the existing scratch allocation alive"
3877 );
3878 }
3879
3880 #[test]
3881 #[ignore = "benchmark evidence only"]
3882 fn wal_frame_scratch_benchmark_report() {
3883 let cases = vec![
3884 track_c_scratch_case_report(
3885 "single_frame_256_ops",
3886 1,
3887 256,
3888 || {
3889 track_c_measure_single_frame_case(
3890 TrackCScratchBenchMode::FreshAllocBaseline,
3891 256,
3892 )
3893 },
3894 || {
3895 track_c_measure_single_frame_case(
3896 TrackCScratchBenchMode::ScratchReuseCandidate,
3897 256,
3898 )
3899 },
3900 ),
3901 track_c_scratch_case_report(
3902 "batch_8_frames_64_ops",
3903 8,
3904 64,
3905 || track_c_measure_batch_case::<8>(TrackCScratchBenchMode::FreshAllocBaseline, 64),
3906 || {
3907 track_c_measure_batch_case::<8>(
3908 TrackCScratchBenchMode::ScratchReuseCandidate,
3909 64,
3910 )
3911 },
3912 ),
3913 track_c_scratch_case_report(
3914 "batch_32_frames_16_ops",
3915 32,
3916 16,
3917 || track_c_measure_batch_case::<32>(TrackCScratchBenchMode::FreshAllocBaseline, 16),
3918 || {
3919 track_c_measure_batch_case::<32>(
3920 TrackCScratchBenchMode::ScratchReuseCandidate,
3921 16,
3922 )
3923 },
3924 ),
3925 ];
3926
3927 let report = json!({
3928 "schema_version": "fsqlite.track_c.wal_scratch_benchmark.v1",
3929 "bead_id": TRACK_C_SCRATCH_BENCH_BEAD_ID,
3930 "parent_bead_id": "bd-db300.3.4",
3931 "measured_operation": "wal_frame_assembly_and_append",
3932 "warmup_iterations": TRACK_C_SCRATCH_BENCH_WARMUP_ITERS,
3933 "measurement_iterations": TRACK_C_SCRATCH_BENCH_MEASURE_ITERS,
3934 "vfs": "memory",
3935 "baseline_variant": "fresh_frame_buffer_per_operation",
3936 "candidate_variant": "reusable_wal_handle_frame_scratch",
3937 "cases": cases,
3938 });
3939
3940 println!("BEGIN_BD_DB300_3_4_3_REPORT");
3941 println!("{}", serde_json::to_string_pretty(&report).unwrap());
3942 println!("END_BD_DB300_3_4_3_REPORT");
3943 }
3944
3945 #[test]
3946 fn test_prepared_batch_append_checksum_chain_matches_single_append() {
3947 let cx = test_cx();
3948 let vfs_single = MemoryVfs::new();
3949 let vfs_prepared = MemoryVfs::new();
3950 let page_size = usize::try_from(PAGE_SIZE).expect("page size fits usize");
3951
3952 let pages: Vec<Vec<u8>> = (0..6u8).map(sample_page).collect();
3953 let page_nums: Vec<u32> = (1..=6u32).collect();
3954 let commit_sizes: Vec<u32> = vec![0, 0, 3, 0, 0, 6];
3955
3956 let file_single = open_wal_file(&vfs_single, &cx);
3957 let mut wal_single =
3958 WalFile::create(&cx, file_single, PAGE_SIZE, 0, test_salts()).expect("create single");
3959 for i in 0..6 {
3960 wal_single
3961 .append_frame(&cx, page_nums[i], &pages[i], commit_sizes[i])
3962 .expect("append single");
3963 }
3964
3965 let file_prepared = open_wal_file(&vfs_prepared, &cx);
3966 let mut wal_prepared = WalFile::create(&cx, file_prepared, PAGE_SIZE, 0, test_salts())
3967 .expect("create prepared");
3968 let frames: Vec<_> = (0..6)
3969 .map(|i| WalAppendFrameRef {
3970 page_number: page_nums[i],
3971 page_data: &pages[i],
3972 db_size_if_commit: commit_sizes[i],
3973 })
3974 .collect();
3975 let mut prepared_bytes = wal_prepared
3976 .prepare_frame_bytes(&frames)
3977 .expect("prepare frame bytes");
3978 let frame_transforms = prepared_bytes
3979 .chunks_exact(wal_prepared.frame_size())
3980 .map(|frame| {
3981 WalChecksumTransform::for_wal_frame(
3982 frame,
3983 page_size,
3984 wal_prepared.big_endian_checksum(),
3985 )
3986 })
3987 .collect::<Result<Vec<_>>>()
3988 .expect("compute frame transforms");
3989 wal_prepared
3990 .append_prepared_frame_bytes(&cx, &mut prepared_bytes, &frame_transforms)
3991 .expect("append prepared batch");
3992
3993 assert_eq!(
3994 wal_single.frame_count(),
3995 wal_prepared.frame_count(),
3996 "prepared append must preserve frame count"
3997 );
3998 assert_eq!(
3999 wal_single.running_checksum(),
4000 wal_prepared.running_checksum(),
4001 "prepared append must preserve checksum chain"
4002 );
4003
4004 for i in 0..6 {
4005 let (single_header, single_data) = wal_single.read_frame(&cx, i).expect("read single");
4006 let (prepared_header, prepared_data) =
4007 wal_prepared.read_frame(&cx, i).expect("read prepared");
4008 assert_eq!(
4009 single_header, prepared_header,
4010 "frame header {i} must match"
4011 );
4012 assert_eq!(single_data, prepared_data, "frame payload {i} must match");
4013 }
4014 }
4015
4016 #[test]
4017 fn test_prepared_batch_reseeds_after_intervening_growth() {
4018 let cx = test_cx();
4019 let vfs_single = MemoryVfs::new();
4020 let vfs_prepared = MemoryVfs::new();
4021 let page_size = usize::try_from(PAGE_SIZE).expect("page size fits usize");
4022
4023 let pages: Vec<Vec<u8>> = (0..3u8).map(sample_page).collect();
4024 let page_nums: Vec<u32> = (1..=3u32).collect();
4025 let commit_sizes: Vec<u32> = vec![0, 0, 3];
4026 let intervening_page = sample_page(0xAA);
4027
4028 let file_single = open_wal_file(&vfs_single, &cx);
4029 let mut wal_single =
4030 WalFile::create(&cx, file_single, PAGE_SIZE, 0, test_salts()).expect("create single");
4031 wal_single
4032 .append_frame(&cx, 99, &intervening_page, 0)
4033 .expect("append intervening single");
4034 for i in 0..3 {
4035 wal_single
4036 .append_frame(&cx, page_nums[i], &pages[i], commit_sizes[i])
4037 .expect("append single");
4038 }
4039
4040 let file_prepared = open_wal_file(&vfs_prepared, &cx);
4041 let mut wal_prepared = WalFile::create(&cx, file_prepared, PAGE_SIZE, 0, test_salts())
4042 .expect("create prepared");
4043 let frames: Vec<_> = (0..3)
4044 .map(|i| WalAppendFrameRef {
4045 page_number: page_nums[i],
4046 page_data: &pages[i],
4047 db_size_if_commit: commit_sizes[i],
4048 })
4049 .collect();
4050 let mut prepared_bytes = wal_prepared
4051 .prepare_frame_bytes(&frames)
4052 .expect("prepare frame bytes");
4053 let frame_transforms = prepared_bytes
4054 .chunks_exact(wal_prepared.frame_size())
4055 .map(|frame| {
4056 WalChecksumTransform::for_wal_frame(
4057 frame,
4058 page_size,
4059 wal_prepared.big_endian_checksum(),
4060 )
4061 })
4062 .collect::<Result<Vec<_>>>()
4063 .expect("compute frame transforms");
4064 wal_prepared
4065 .append_frame(&cx, 99, &intervening_page, 0)
4066 .expect("append intervening prepared");
4067 wal_prepared
4068 .append_prepared_frame_bytes(&cx, &mut prepared_bytes, &frame_transforms)
4069 .expect("append prepared batch");
4070
4071 assert_eq!(
4072 wal_single.frame_count(),
4073 wal_prepared.frame_count(),
4074 "prepared append after growth must preserve frame count"
4075 );
4076 assert_eq!(
4077 wal_single.running_checksum(),
4078 wal_prepared.running_checksum(),
4079 "prepared append after growth must rebind to the live checksum seed"
4080 );
4081
4082 for i in 0..wal_single.frame_count() {
4083 let (single_header, single_data) = wal_single.read_frame(&cx, i).expect("read single");
4084 let (prepared_header, prepared_data) =
4085 wal_prepared.read_frame(&cx, i).expect("read prepared");
4086 assert_eq!(
4087 single_header, prepared_header,
4088 "frame header {i} must match"
4089 );
4090 assert_eq!(single_data, prepared_data, "frame payload {i} must match");
4091 }
4092 }
4093
4094 #[test]
4095 fn test_prepared_batch_rewrites_salts_after_reset() {
4096 let cx = test_cx();
4097 let vfs_fresh = MemoryVfs::new();
4098 let vfs_reset = MemoryVfs::new();
4099 let page_size = usize::try_from(PAGE_SIZE).expect("page size fits usize");
4100 let reset_salts = WalSalts {
4101 salt1: 0x0102_0304,
4102 salt2: 0xA0B0_C0D0,
4103 };
4104
4105 let pages: Vec<Vec<u8>> = (0..2u8).map(sample_page).collect();
4106 let page_nums: Vec<u32> = (1..=2u32).collect();
4107 let commit_sizes: Vec<u32> = vec![0, 2];
4108
4109 let file_fresh = open_wal_file(&vfs_fresh, &cx);
4110 let mut wal_fresh =
4111 WalFile::create(&cx, file_fresh, PAGE_SIZE, 7, reset_salts).expect("create fresh");
4112 for i in 0..2 {
4113 wal_fresh
4114 .append_frame(&cx, page_nums[i], &pages[i], commit_sizes[i])
4115 .expect("append fresh");
4116 }
4117
4118 let file_reset = open_wal_file(&vfs_reset, &cx);
4119 let mut wal_reset =
4120 WalFile::create(&cx, file_reset, PAGE_SIZE, 0, test_salts()).expect("create reset");
4121 let frames: Vec<_> = (0..2)
4122 .map(|i| WalAppendFrameRef {
4123 page_number: page_nums[i],
4124 page_data: &pages[i],
4125 db_size_if_commit: commit_sizes[i],
4126 })
4127 .collect();
4128 let mut prepared_bytes = wal_reset
4129 .prepare_frame_bytes(&frames)
4130 .expect("prepare frame bytes");
4131 let frame_transforms = prepared_bytes
4132 .chunks_exact(wal_reset.frame_size())
4133 .map(|frame| {
4134 WalChecksumTransform::for_wal_frame(
4135 frame,
4136 page_size,
4137 wal_reset.big_endian_checksum(),
4138 )
4139 })
4140 .collect::<Result<Vec<_>>>()
4141 .expect("compute frame transforms");
4142 wal_reset
4143 .reset(&cx, 7, reset_salts, false)
4144 .expect("reset WAL");
4145 wal_reset
4146 .append_prepared_frame_bytes(&cx, &mut prepared_bytes, &frame_transforms)
4147 .expect("append prepared batch");
4148
4149 assert_eq!(
4150 wal_fresh.frame_count(),
4151 wal_reset.frame_count(),
4152 "prepared append after reset must preserve frame count"
4153 );
4154 assert_eq!(
4155 wal_fresh.running_checksum(),
4156 wal_reset.running_checksum(),
4157 "prepared append after reset must rebind to the reset checksum seed"
4158 );
4159
4160 for i in 0..2 {
4161 let (fresh_header, fresh_data) = wal_fresh.read_frame(&cx, i).expect("read fresh");
4162 let (reset_header, reset_data) = wal_reset.read_frame(&cx, i).expect("read reset");
4163 assert_eq!(fresh_header, reset_header, "frame header {i} must match");
4164 assert_eq!(
4165 fresh_header.salts, reset_salts,
4166 "frame {i} must use reset salts"
4167 );
4168 assert_eq!(fresh_data, reset_data, "frame payload {i} must match");
4169 }
4170 }
4171
4172 #[test]
4173 fn test_uncommitted_tail_trimmed_on_reopen() {
4174 let cx = test_cx();
4177 let vfs = MemoryVfs::new();
4178 let file = open_wal_file(&vfs, &cx);
4179 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create");
4180
4181 let commit_map: [(u32, u32); 5] = [(1, 0), (2, 0), (3, 3), (4, 0), (5, 0)];
4182 for (pg, db_sz) in commit_map {
4183 wal.append_frame(&cx, pg, &sample_page(u8::try_from(pg).unwrap()), db_sz)
4184 .expect("append");
4185 }
4186 assert_eq!(wal.frame_count(), 5);
4187 wal.close(&cx).expect("close");
4188
4189 let file2 = open_wal_file(&vfs, &cx);
4191 let wal2 = WalFile::open(&cx, file2).expect("reopen");
4192 assert_eq!(
4193 wal2.frame_count(),
4194 3,
4195 "frames after last commit should be trimmed on reopen"
4196 );
4197 wal2.close(&cx).expect("close");
4198 }
4199
4200 #[test]
4201 fn test_large_transaction_50_frames() {
4202 let cx = test_cx();
4204 let vfs = MemoryVfs::new();
4205 let file = open_wal_file(&vfs, &cx);
4206 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create");
4207
4208 let n = 50u32;
4209 for i in 0..n {
4210 let db_size = if i == n - 1 { n } else { 0 };
4211 let seed = u8::try_from(i % 251).unwrap();
4212 wal.append_frame(&cx, i + 1, &sample_page(seed), db_size)
4213 .expect("append");
4214 }
4215 assert_eq!(wal.frame_count(), usize::try_from(n).unwrap());
4216 let final_checksum = wal.running_checksum();
4217 wal.close(&cx).expect("close");
4218
4219 let file2 = open_wal_file(&vfs, &cx);
4221 let wal2 = WalFile::open(&cx, file2).expect("reopen");
4222 assert_eq!(wal2.frame_count(), usize::try_from(n).unwrap());
4223 assert_eq!(wal2.running_checksum(), final_checksum);
4224
4225 for idx in [0, 24, 49] {
4227 let (h, d) = wal2.read_frame(&cx, idx).expect("read");
4228 let i = u32::try_from(idx).unwrap();
4229 assert_eq!(h.page_number, i + 1);
4230 assert_eq!(d, sample_page(u8::try_from(i % 251).unwrap()));
4231 }
4232
4233 wal2.close(&cx).expect("close");
4234 }
4235
4236 #[test]
4237 fn test_append_after_reset_checksum_independent() {
4238 let cx = test_cx();
4242
4243 let salts = WalSalts {
4244 salt1: 0x1234_5678,
4245 salt2: 0x9ABC_DEF0,
4246 };
4247
4248 let vfs1 = MemoryVfs::new();
4250 let file1 = open_wal_file(&vfs1, &cx);
4251 let mut wal_fresh = WalFile::create(&cx, file1, PAGE_SIZE, 1, salts).expect("create fresh");
4252 wal_fresh
4253 .append_frame(&cx, 1, &sample_page(0x42), 1)
4254 .expect("append fresh");
4255 let fresh_checksum = wal_fresh.running_checksum();
4256 wal_fresh.close(&cx).expect("close fresh");
4257
4258 let vfs2 = MemoryVfs::new();
4260 let file2 = open_wal_file(&vfs2, &cx);
4261 let mut wal_reset =
4262 WalFile::create(&cx, file2, PAGE_SIZE, 0, test_salts()).expect("create reset");
4263 wal_reset
4265 .append_frame(&cx, 99, &sample_page(0xFF), 99)
4266 .expect("append old");
4267 wal_reset.reset(&cx, 1, salts, false).expect("reset");
4269 wal_reset
4270 .append_frame(&cx, 1, &sample_page(0x42), 1)
4271 .expect("append after reset");
4272 let reset_checksum = wal_reset.running_checksum();
4273 wal_reset.close(&cx).expect("close reset");
4274
4275 assert_eq!(
4276 fresh_checksum, reset_checksum,
4277 "after reset with same salts, checksum chain must match fresh WAL"
4278 );
4279 }
4280
4281 fn build_two_txn_wal() -> (MemoryVfs, Vec<Vec<u8>>) {
4289 let cx = test_cx();
4290 let vfs = MemoryVfs::new();
4291 let file = open_wal_file(&vfs, &cx);
4292 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create");
4293
4294 let mut pages = Vec::new();
4295 let frame_specs: [(u32, u32); 6] = [
4296 (1, 0),
4297 (2, 0),
4298 (3, 3), (4, 0),
4300 (5, 0),
4301 (6, 6), ];
4303 for (pg, db_sz) in frame_specs {
4304 let page = sample_page(u8::try_from(pg).unwrap());
4305 wal.append_frame(&cx, pg, &page, db_sz).expect("append");
4306 pages.push(page);
4307 }
4308 wal.close(&cx).expect("close");
4309 (vfs, pages)
4310 }
4311
4312 #[test]
4313 fn test_crash_matrix_truncate_at_every_frame_boundary() {
4314 let cx = test_cx();
4317 let frame_size = WAL_FRAME_HEADER_SIZE + usize::try_from(PAGE_SIZE).unwrap();
4318
4319 for cut_frames in 0..=6usize {
4320 let (vfs, _) = build_two_txn_wal();
4322
4323 let cut_at = WAL_HEADER_SIZE + cut_frames * frame_size;
4324 let mut f = open_wal_file(&vfs, &cx);
4325 f.truncate(&cx, u64::try_from(cut_at).unwrap())
4326 .expect("truncate");
4327 drop(f);
4328
4329 let f2 = open_wal_file(&vfs, &cx);
4330 let wal = WalFile::open(&cx, f2).expect("open after truncation");
4331 let expected = match cut_frames {
4332 0..=2 => 0, 3..=5 => 3, 6 => 6, _ => unreachable!(),
4336 };
4337 assert_eq!(
4338 wal.frame_count(),
4339 expected,
4340 "truncated at {cut_frames} frames should give {expected} committed"
4341 );
4342 wal.close(&cx).expect("close");
4343 }
4344
4345 for partial in 0..20usize {
4347 let (vfs, _) = build_two_txn_wal();
4348 let cx = test_cx();
4349
4350 let cut_byte = WAL_HEADER_SIZE + partial * frame_size / 3;
4351 let mut f = open_wal_file(&vfs, &cx);
4352 f.truncate(&cx, u64::try_from(cut_byte).unwrap())
4353 .expect("truncate");
4354 drop(f);
4355
4356 let f2 = open_wal_file(&vfs, &cx);
4357 let wal = WalFile::open(&cx, f2).expect("open");
4358 assert!(
4360 wal.frame_count() == 0 || wal.frame_count() == 3 || wal.frame_count() == 6,
4361 "cut_byte={cut_byte} gave frame_count={}, expected 0/3/6",
4362 wal.frame_count()
4363 );
4364 wal.close(&cx).expect("close");
4365 }
4366 }
4367
4368 #[test]
4369 fn test_crash_matrix_bit_flip_at_every_frame() {
4370 for target_frame in 0..6usize {
4373 let (vfs, _) = build_two_txn_wal();
4374 let cx = test_cx();
4375
4376 let frame_size = WAL_FRAME_HEADER_SIZE + usize::try_from(PAGE_SIZE).unwrap();
4377 let corrupt_offset =
4378 WAL_HEADER_SIZE + target_frame * frame_size + WAL_FRAME_HEADER_SIZE + 42;
4379
4380 let f = open_wal_file(&vfs, &cx);
4382 let mut buf = [0u8; 1];
4383 let off = u64::try_from(corrupt_offset).unwrap();
4384 f.read(&cx, &mut buf, off).expect("read");
4385 buf[0] ^= 0xFF;
4386 f.write(&cx, &buf, off).expect("write corrupt");
4387 drop(f);
4388
4389 let f2 = open_wal_file(&vfs, &cx);
4390 let wal = WalFile::open(&cx, f2).expect("open");
4391 let expected = if target_frame < 3 {
4392 0 } else {
4394 3 };
4396 assert_eq!(
4397 wal.frame_count(),
4398 expected,
4399 "bit flip in frame {target_frame} should give {expected}"
4400 );
4401 wal.close(&cx).expect("close");
4402 }
4403 }
4404
4405 #[test]
4406 fn test_crash_matrix_continue_after_recovery() {
4407 let (vfs, _) = build_two_txn_wal();
4410 let cx = test_cx();
4411
4412 let frame_size = WAL_FRAME_HEADER_SIZE + usize::try_from(PAGE_SIZE).unwrap();
4413
4414 let corrupt_offset = WAL_HEADER_SIZE + 4 * frame_size + WAL_FRAME_HEADER_SIZE + 10;
4416 let f = open_wal_file(&vfs, &cx);
4417 let mut buf = [0u8; 1];
4418 let off = u64::try_from(corrupt_offset).unwrap();
4419 f.read(&cx, &mut buf, off).expect("read");
4420 buf[0] ^= 0xAA;
4421 f.write(&cx, &buf, off).expect("write corrupt");
4422 drop(f);
4423
4424 let f2 = open_wal_file(&vfs, &cx);
4426 let mut wal = WalFile::open(&cx, f2).expect("open");
4427 assert_eq!(wal.frame_count(), 3);
4428
4429 wal.append_frame(&cx, 10, &sample_page(0xAA), 0)
4431 .expect("append");
4432 wal.append_frame(&cx, 11, &sample_page(0xBB), 5)
4433 .expect("append commit");
4434 assert_eq!(wal.frame_count(), 5);
4435 let checksum_after = wal.running_checksum();
4436 wal.close(&cx).expect("close");
4437
4438 let f3 = open_wal_file(&vfs, &cx);
4440 let wal2 = WalFile::open(&cx, f3).expect("reopen");
4441 assert_eq!(wal2.frame_count(), 5);
4442 assert_eq!(wal2.running_checksum(), checksum_after);
4443
4444 for i in 0..3 {
4446 let (h, d) = wal2.read_frame(&cx, i).expect("read");
4447 let pg = u32::try_from(i + 1).unwrap();
4448 assert_eq!(h.page_number, pg);
4449 assert_eq!(d, sample_page(u8::try_from(pg).unwrap()));
4450 }
4451
4452 let (h4, d4) = wal2.read_frame(&cx, 3).expect("read new frame 4");
4454 assert_eq!(h4.page_number, 10);
4455 assert_eq!(d4, sample_page(0xAA));
4456
4457 wal2.close(&cx).expect("close");
4458 }
4459
4460 #[test]
4461 fn test_crash_matrix_zero_length_wal() {
4462 let cx = test_cx();
4464 let vfs = MemoryVfs::new();
4465 let file = open_wal_file(&vfs, &cx);
4466 let wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create");
4467 wal.close(&cx).expect("close");
4468
4469 let f2 = open_wal_file(&vfs, &cx);
4470 let wal2 = WalFile::open(&cx, f2).expect("open");
4471 assert_eq!(wal2.frame_count(), 0);
4472 wal2.close(&cx).expect("close");
4473 }
4474
4475 #[test]
4476 fn test_crash_matrix_header_only_partial_first_frame() {
4477 let cx = test_cx();
4479 let vfs = MemoryVfs::new();
4480 let file = open_wal_file(&vfs, &cx);
4481
4482 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create");
4483 wal.append_frame(&cx, 1, &sample_page(1), 1)
4484 .expect("append");
4485 let partial_size = WAL_HEADER_SIZE + WAL_FRAME_HEADER_SIZE + 10;
4486 wal.file_mut()
4487 .truncate(&cx, u64::try_from(partial_size).unwrap())
4488 .expect("truncate");
4489 wal.close(&cx).expect("close");
4490
4491 let f2 = open_wal_file(&vfs, &cx);
4492 let wal2 = WalFile::open(&cx, f2).expect("open");
4493 assert_eq!(wal2.frame_count(), 0, "partial frame should be dropped");
4494 wal2.close(&cx).expect("close");
4495 }
4496
4497 #[test]
4498 fn test_crash_matrix_many_txns_deterministic_recovery() {
4499 let cx = test_cx();
4502
4503 for crash_txn in 0..=10usize {
4504 let vfs = MemoryVfs::new();
4505 let file = open_wal_file(&vfs, &cx);
4506 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create");
4507
4508 let total_frames = crash_txn * 3;
4509 for txn in 0..crash_txn {
4510 for f in 0..3u32 {
4511 let pg = u32::try_from(txn * 3).unwrap() + f + 1;
4512 let db_size = if f == 2 {
4513 u32::try_from(txn * 3 + 3).unwrap()
4514 } else {
4515 0
4516 };
4517 let seed = u8::try_from(pg % 251).unwrap();
4518 wal.append_frame(&cx, pg, &sample_page(seed), db_size)
4519 .expect("append");
4520 }
4521 }
4522 assert_eq!(wal.frame_count(), total_frames);
4523 wal.close(&cx).expect("close");
4524
4525 let f2 = open_wal_file(&vfs, &cx);
4527 let wal2 = WalFile::open(&cx, f2).expect("open");
4528 assert_eq!(
4529 wal2.frame_count(),
4530 total_frames,
4531 "crash_txn={crash_txn}: all {total_frames} committed frames should survive"
4532 );
4533 wal2.close(&cx).expect("close");
4534
4535 if crash_txn < 10 {
4537 let f3 = open_wal_file(&vfs, &cx);
4539 let mut wal3 = WalFile::open(&cx, f3).expect("open");
4540 let extra_pg = u32::try_from(total_frames + 1).unwrap();
4541 wal3.append_frame(
4542 &cx,
4543 extra_pg,
4544 &sample_page(u8::try_from(extra_pg % 251).unwrap()),
4545 0,
4546 )
4547 .expect("append uncommitted");
4548 wal3.close(&cx).expect("close");
4549
4550 let f4 = open_wal_file(&vfs, &cx);
4552 let wal4 = WalFile::open(&cx, f4).expect("open");
4553 assert_eq!(
4554 wal4.frame_count(),
4555 total_frames,
4556 "crash_txn={crash_txn}: uncommitted extra frame dropped"
4557 );
4558 wal4.close(&cx).expect("close");
4559 }
4560 }
4561 }
4562
4563 #[test]
4564 fn test_crash_matrix_reset_then_crash() {
4565 let cx = test_cx();
4567 let vfs = MemoryVfs::new();
4568 let file = open_wal_file(&vfs, &cx);
4569 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create");
4570
4571 wal.append_frame(&cx, 1, &sample_page(1), 1)
4573 .expect("append");
4574 let new_salts = WalSalts {
4576 salt1: 0x5555_6666,
4577 salt2: 0x7777_8888,
4578 };
4579 wal.reset(&cx, 1, new_salts, true).expect("reset");
4580 assert_eq!(wal.frame_count(), 0);
4581
4582 wal.append_frame(&cx, 1, &sample_page(0xCC), 0)
4584 .expect("append");
4585 wal.append_frame(&cx, 2, &sample_page(0xDD), 0)
4586 .expect("append");
4587 wal.close(&cx).expect("close");
4588
4589 let f2 = open_wal_file(&vfs, &cx);
4591 let wal2 = WalFile::open(&cx, f2).expect("open");
4592 assert_eq!(wal2.frame_count(), 0, "no commits after reset");
4593 assert_eq!(wal2.header().salts, new_salts);
4594 wal2.close(&cx).expect("close");
4595 }
4596
4597 #[test]
4598 fn test_truncated_file_mid_second_txn_recovers_first_commit() {
4599 let cx = test_cx();
4600 let vfs = MemoryVfs::new();
4601 let file = open_wal_file(&vfs, &cx);
4602
4603 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create");
4604
4605 wal.append_frame(&cx, 1, &sample_page(0x11), 0)
4607 .expect("append");
4608 wal.append_frame(&cx, 2, &sample_page(0x22), 2)
4609 .expect("commit txn1");
4610
4611 wal.append_frame(&cx, 3, &sample_page(0x33), 0)
4613 .expect("append");
4614 wal.append_frame(&cx, 4, &sample_page(0x44), 0)
4615 .expect("append");
4616 assert_eq!(wal.frame_count(), 4);
4617 let frame_size = wal.frame_size();
4618 wal.close(&cx).expect("close");
4619
4620 let truncate_offset =
4622 u64::try_from(WAL_HEADER_SIZE + frame_size * 2 + frame_size / 2).unwrap();
4623 let mut f = open_wal_file(&vfs, &cx);
4624 f.truncate(&cx, truncate_offset).expect("truncate");
4625 drop(f);
4626
4627 let file2 = open_wal_file(&vfs, &cx);
4628 let wal2 = WalFile::open(&cx, file2).expect("reopen after truncation");
4629 assert_eq!(
4630 wal2.frame_count(),
4631 2,
4632 "only first committed transaction (2 frames) should survive truncation"
4633 );
4634 let h = wal2.read_frame_header(&cx, 1).expect("read frame 2 header");
4635 assert!(h.is_commit(), "frame 2 must be a commit frame");
4636 wal2.close(&cx).expect("close");
4637 }
4638
4639 #[test]
4640 fn test_recovery_is_idempotent_across_multiple_reopens() {
4641 let cx = test_cx();
4642 let vfs = MemoryVfs::new();
4643 let file = open_wal_file(&vfs, &cx);
4644
4645 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create");
4646
4647 wal.append_frame(&cx, 1, &sample_page(0xAA), 0)
4649 .expect("append");
4650 wal.append_frame(&cx, 2, &sample_page(0xBB), 2)
4651 .expect("commit txn1");
4652 wal.append_frame(&cx, 3, &sample_page(0xCC), 0)
4653 .expect("append");
4654 wal.append_frame(&cx, 4, &sample_page(0xDD), 4)
4655 .expect("commit txn2");
4656 wal.append_frame(&cx, 5, &sample_page(0xEE), 0)
4658 .expect("append");
4659 wal.close(&cx).expect("close");
4660
4661 for reopen in 0..3_u32 {
4663 let f = open_wal_file(&vfs, &cx);
4664 let wal_reopened = WalFile::open(&cx, f).expect("reopen");
4665 assert_eq!(
4666 wal_reopened.frame_count(),
4667 4,
4668 "reopen {reopen}: committed frame count must be 4"
4669 );
4670 let (_, data) = wal_reopened.read_frame(&cx, 3).expect("read frame 4");
4671 assert_eq!(
4672 data,
4673 sample_page(0xDD),
4674 "reopen {reopen}: frame 4 content must be intact"
4675 );
4676 wal_reopened.close(&cx).expect("close");
4677 }
4678 }
4679
4680 #[test]
4681 fn wal_generation_identity_from_header_and_eq() {
4682 let header = WalHeader {
4683 magic: WAL_MAGIC_LE,
4684 format_version: WAL_FORMAT_VERSION,
4685 page_size: PAGE_SIZE,
4686 checkpoint_seq: 7,
4687 salts: test_salts(),
4688 checksum: SqliteWalChecksum { s1: 0, s2: 0 },
4689 };
4690 let identity = WalGenerationIdentity::from_header(&header);
4691 assert_eq!(identity.checkpoint_seq, 7);
4692 assert_eq!(identity.salts, test_salts());
4693 let copied = identity;
4694 assert_eq!(copied, identity);
4695 let other = WalGenerationIdentity {
4696 checkpoint_seq: 8,
4697 salts: test_salts(),
4698 };
4699 assert_ne!(identity, other);
4700 let dbg = format!("{identity:?}");
4701 assert!(dbg.contains("WalGenerationIdentity"));
4702 }
4703
4704 #[test]
4705 fn wal_append_frame_ref_debug_clone_copy() {
4706 let data = [0xABu8; 16];
4707 let frame = WalAppendFrameRef {
4708 page_number: 3,
4709 page_data: &data,
4710 db_size_if_commit: 10,
4711 };
4712 let copied = frame;
4713 assert_eq!(copied.page_number, 3);
4714 assert_eq!(copied.db_size_if_commit, 10);
4715 assert_eq!(copied.page_data[0], 0xAB);
4716 let cloned = frame;
4717 assert_eq!(cloned.page_number, frame.page_number);
4718 let dbg = format!("{frame:?}");
4719 assert!(dbg.contains("WalAppendFrameRef"));
4720 }
4721
4722 #[test]
4723 fn wal_file_generation_identity_matches_create_params() {
4724 let cx = test_cx();
4725 let vfs = MemoryVfs::new();
4726 let file = open_wal_file(&vfs, &cx);
4727 let wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create");
4728 let identity = wal.generation_identity();
4729 assert_eq!(identity.checkpoint_seq, 0);
4730 assert_eq!(identity.salts, test_salts());
4731 wal.close(&cx).expect("close");
4732 }
4733
4734 #[test]
4735 fn wal_file_page_size_and_frame_count_after_create() {
4736 let cx = test_cx();
4737 let vfs = MemoryVfs::new();
4738 let file = open_wal_file(&vfs, &cx);
4739 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create");
4740 assert_eq!(wal.page_size(), PAGE_SIZE as usize);
4741 assert_eq!(wal.frame_count(), 0);
4742 assert!(wal.last_commit_frame(&cx).expect("query").is_none());
4743 wal.append_frame(&cx, 1, &sample_page(1), 5)
4744 .expect("append");
4745 assert_eq!(wal.frame_count(), 1);
4746 assert_eq!(wal.last_commit_frame(&cx).expect("query"), Some(0));
4747 wal.close(&cx).expect("close");
4748 }
4749
4750 #[test]
4751 fn wal_file_big_endian_checksum_and_running_checksum() {
4752 let cx = test_cx();
4753 let vfs = MemoryVfs::new();
4754 let file = open_wal_file(&vfs, &cx);
4755 let wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create");
4756 let _be = wal.big_endian_checksum();
4757 let rc = wal.running_checksum();
4758 assert_eq!(rc.s1.wrapping_add(0), rc.s1);
4759 wal.close(&cx).expect("close");
4760 }
4761
4762 #[test]
4763 fn wal_file_frame_size_equals_header_plus_page() {
4764 let cx = test_cx();
4765 let vfs = MemoryVfs::new();
4766 let file = open_wal_file(&vfs, &cx);
4767 let wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create");
4768 assert_eq!(wal.frame_size(), WAL_FRAME_HEADER_SIZE + PAGE_SIZE as usize);
4769 wal.close(&cx).expect("close");
4770 }
4771
4772 #[test]
4773 fn wal_file_header_accessor() {
4774 let cx = test_cx();
4775 let vfs = MemoryVfs::new();
4776 let file = open_wal_file(&vfs, &cx);
4777 let wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create");
4778 let hdr = wal.header();
4779 assert_eq!(hdr.page_size, PAGE_SIZE);
4780 assert_eq!(hdr.salts, test_salts());
4781 wal.close(&cx).expect("close");
4782 }
4783
4784 #[test]
4785 fn wal_file_file_and_file_mut_accessors() {
4786 let cx = test_cx();
4787 let vfs = MemoryVfs::new();
4788 let file = open_wal_file(&vfs, &cx);
4789 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create");
4790 let _f_ref = wal.file();
4791 let _f_mut = wal.file_mut();
4792 wal.close(&cx).expect("close");
4793 }
4794
4795 #[test]
4796 fn durable_sync_records_fsynced_frame_count() {
4797 let cx = test_cx();
4798 let vfs = MemoryVfs::new();
4799 let file = open_wal_file(&vfs, &cx);
4800 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create");
4801 assert_eq!(wal.last_fsynced_frame_count(), 0);
4802
4803 let page = vec![0xABu8; PAGE_SIZE as usize];
4804 let frames = [frame_ref(1, &page, 1)];
4805 wal.append_frames(&cx, &frames).expect("append");
4806 assert_eq!(wal.frame_count(), 1);
4807 assert_eq!(wal.last_fsynced_frame_count(), 0);
4808
4809 wal.durable_sync(&cx, fsqlite_vfs::SyncKind::FullDurable)
4810 .expect("durable_sync");
4811 assert_eq!(wal.last_fsynced_frame_count(), 1);
4812 wal.close(&cx).expect("close");
4813 }
4814
4815 #[test]
4816 fn assert_publish_safe_passes_after_durable_sync() {
4817 let cx = test_cx();
4818 let vfs = MemoryVfs::new();
4819 let file = open_wal_file(&vfs, &cx);
4820 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create");
4821
4822 let page = vec![0xCDu8; PAGE_SIZE as usize];
4823 let frames = [frame_ref(1, &page, 1)];
4824 wal.append_frames(&cx, &frames).expect("append");
4825 wal.durable_sync(&cx, fsqlite_vfs::SyncKind::FullDurable)
4826 .expect("durable_sync");
4827 wal.assert_publish_safe(1)
4828 .expect("should be safe after fsync");
4829 wal.close(&cx).expect("close");
4830 }
4831
4832 #[test]
4833 fn assert_publish_safe_passes_for_zero_frames() {
4834 let cx = test_cx();
4835 let vfs = MemoryVfs::new();
4836 let file = open_wal_file(&vfs, &cx);
4837 let wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create");
4838 wal.assert_publish_safe(0).expect("zero frames always safe");
4839 wal.close(&cx).expect("close");
4840 }
4841
4842 #[test]
4853 fn two_phase_publish_gate_blocks_commit_until_fsync() {
4854 let cx = test_cx();
4855 let vfs = MemoryVfs::new();
4856 let file = open_wal_file(&vfs, &cx);
4857 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create WAL");
4858
4859 assert_eq!(wal.frame_count(), 0);
4861 assert_eq!(wal.last_commit_frame(&cx).expect("query"), None);
4862 assert_eq!(wal.last_fsynced_frame_count(), 0);
4863
4864 wal.append_frame(&cx, 1, &sample_page(0x11), 1)
4867 .expect("append commit frame");
4868
4869 assert_eq!(wal.frame_count(), 1);
4871 assert_eq!(
4872 wal.last_commit_frame(&cx).expect("query"),
4873 Some(0),
4874 "append cursor (last_commit_frame) must advance during append"
4875 );
4876 assert_eq!(
4878 wal.last_fsynced_frame_count(),
4879 0,
4880 "durable watermark must NOT advance on append (only on a successful sync)"
4881 );
4882
4883 assert!(
4889 wal.last_fsynced_frame_count() < 1,
4890 "watermark below publish_frame_count means the commit is not yet publishable"
4891 );
4892 if cfg!(debug_assertions) {
4893 let tripped = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
4898 let _ = wal.assert_publish_safe(1);
4899 }));
4900 assert!(
4901 tripped.is_err(),
4902 "assert_publish_safe(1) must trip the durability guard when the \
4903 just-appended commit has not been fsynced"
4904 );
4905 }
4906
4907 wal.durable_sync(&cx, fsqlite_vfs::SyncKind::FullDurable)
4910 .expect("durable_sync");
4911 assert_eq!(
4912 wal.last_fsynced_frame_count(),
4913 1,
4914 "durable watermark equals the frame count after a successful fsync"
4915 );
4916 wal.assert_publish_safe(1)
4917 .expect("publish must be safe once the commit frame is fsynced");
4918
4919 wal.close(&cx).expect("close WAL");
4920 }
4921
4922 #[test]
4923 fn durable_sync_with_data_only_kind() {
4924 let cx = test_cx();
4925 let vfs = MemoryVfs::new();
4926 let file = open_wal_file(&vfs, &cx);
4927 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create");
4928
4929 let page = vec![0xEFu8; PAGE_SIZE as usize];
4930 let frames = [frame_ref(2, &page, 1)];
4931 wal.append_frames(&cx, &frames).expect("append");
4932 wal.durable_sync(&cx, fsqlite_vfs::SyncKind::DataOnly)
4933 .expect("data-only sync");
4934 assert_eq!(wal.last_fsynced_frame_count(), 1);
4935 wal.close(&cx).expect("close");
4936 }
4937
4938 #[test]
4939 fn raw_sync_reset_clears_fsynced_count() {
4940 let cx = test_cx();
4941 let vfs = MemoryVfs::new();
4942 let file = open_wal_file(&vfs, &cx);
4943 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create");
4944
4945 let page = vec![0x11u8; PAGE_SIZE as usize];
4946 let frames = [frame_ref(1, &page, 1)];
4947 wal.append_frames(&cx, &frames).expect("append");
4948 wal.sync(&cx, SyncFlags::NORMAL).expect("sync");
4949 assert_eq!(wal.last_fsynced_frame_count(), 1);
4950
4951 let new_salts = WalSalts {
4952 salt1: 0x1111_1111,
4953 salt2: 0x2222_2222,
4954 };
4955 wal.reset(&cx, 1, new_salts, true).expect("reset");
4956 assert_eq!(wal.last_fsynced_frame_count(), 0);
4957 wal.close(&cx).expect("close");
4958 }
4959
4960 #[test]
4961 fn opened_wal_has_fsynced_count_equal_to_frame_count() {
4962 let cx = test_cx();
4963 let vfs = MemoryVfs::new();
4964 let file = open_wal_file(&vfs, &cx);
4965 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create");
4966
4967 let page = vec![0x22u8; PAGE_SIZE as usize];
4968 let frames = [frame_ref(1, &page, 1)];
4969 wal.append_frames(&cx, &frames).expect("append");
4970 wal.durable_sync(&cx, fsqlite_vfs::SyncKind::FullDurable)
4971 .expect("durable_sync");
4972 wal.close(&cx).expect("close");
4973
4974 let file2 = open_wal_file(&vfs, &cx);
4975 let wal2 = WalFile::open(&cx, file2).expect("open");
4976 assert_eq!(wal2.frame_count(), 1);
4977 assert_eq!(
4978 wal2.last_fsynced_frame_count(),
4979 wal2.frame_count(),
4980 "opened WAL assumes existing frames are durable"
4981 );
4982 wal2.close(&cx).expect("close");
4983 }
4984
4985 #[test]
4986 fn crash_before_wal_frame_append_preserves_existing_frames() {
4987 let _guard = FAULT_TEST_LOCK.lock().unwrap();
4988 let cx = test_cx();
4989 let vfs = MemoryVfs::new();
4990 let file = open_wal_file(&vfs, &cx);
4991 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create");
4992
4993 let p1 = sample_page(0xAA);
4994 let first_frames = [frame_ref(1, &p1, 1)];
4995 wal.append_frames(&cx, &first_frames).expect("first append");
4996 wal.durable_sync(&cx, SyncKind::FullDurable)
4997 .expect("sync first frame");
4998 assert_eq!(wal.frame_count(), 1);
4999
5000 crate::fault_hooks::arm_crash_boundary(
5001 crate::fault_hooks::CrashBoundary::BeforeWalFrameAppend,
5002 crate::fault_hooks::FaultHookArm::new(
5003 "crash-before-append",
5004 "WAL-FRAME-APPEND-CRASH",
5005 "test_crash_before_append",
5006 ),
5007 );
5008
5009 let p2 = sample_page(0xBB);
5010 let second_frames = [frame_ref(2, &p2, 2)];
5011 let err = wal
5012 .append_frames(&cx, &second_frames)
5013 .expect_err("should fail at crash boundary");
5014 assert!(
5015 err.to_string().contains("fault_inject"),
5016 "error identifies the fault hook: {err}"
5017 );
5018
5019 crate::fault_hooks::clear_crash_boundary();
5020
5021 wal.close(&cx).expect("close after crash");
5022
5023 let file2 = open_wal_file(&vfs, &cx);
5024 let recovered = WalFile::open(&cx, file2).expect("reopen");
5025 assert_eq!(
5026 recovered.frame_count(),
5027 1,
5028 "only the first committed frame survives; the second was never written"
5029 );
5030 recovered.close(&cx).expect("close recovered");
5031 }
5032
5033 #[test]
5034 fn crash_after_fsync_before_publish_leaves_frames_durable() {
5035 let _guard = FAULT_TEST_LOCK.lock().unwrap();
5036 let cx = test_cx();
5037 let vfs = MemoryVfs::new();
5038 let file = open_wal_file(&vfs, &cx);
5039 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create");
5040
5041 let p1 = sample_page(0xCC);
5042 let frames = [frame_ref(1, &p1, 1)];
5043 wal.append_frames(&cx, &frames).expect("append");
5044
5045 crate::fault_hooks::arm_crash_boundary(
5046 crate::fault_hooks::CrashBoundary::AfterFsyncBeforePublish,
5047 crate::fault_hooks::FaultHookArm::new(
5048 "crash-after-fsync",
5049 "WAL-FSYNC-PUBLISH-CRASH",
5050 "test_crash_after_fsync",
5051 ),
5052 );
5053
5054 let err = wal
5055 .durable_sync(&cx, SyncKind::FullDurable)
5056 .expect_err("should fail at crash boundary");
5057 assert!(
5058 err.to_string().contains("fault_inject"),
5059 "error identifies the fault hook: {err}"
5060 );
5061
5062 crate::fault_hooks::clear_crash_boundary();
5063
5064 wal.close(&cx).expect("close after crash");
5065
5066 let file2 = open_wal_file(&vfs, &cx);
5067 let recovered = WalFile::open(&cx, file2).expect("reopen");
5068 assert_eq!(
5069 recovered.frame_count(),
5070 1,
5071 "frame was fsynced before crash so it survives recovery"
5072 );
5073 recovered.close(&cx).expect("close recovered");
5074 }
5075
5076 #[test]
5077 fn crash_before_wal_header_write_on_reset_preserves_old_generation() {
5078 let _guard = FAULT_TEST_LOCK.lock().unwrap();
5079 let cx = test_cx();
5080 let vfs = MemoryVfs::new();
5081 let file = open_wal_file(&vfs, &cx);
5082 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create");
5083
5084 let p1 = sample_page(0xDD);
5085 let frames = [frame_ref(1, &p1, 1)];
5086 wal.append_frames(&cx, &frames).expect("append");
5087 wal.durable_sync(&cx, SyncKind::FullDurable).expect("sync");
5088 assert_eq!(wal.frame_count(), 1);
5089
5090 let original_salts = wal.generation_identity().salts;
5091
5092 crate::fault_hooks::arm_crash_boundary(
5093 crate::fault_hooks::CrashBoundary::BeforeWalHeaderWrite,
5094 crate::fault_hooks::FaultHookArm::new(
5095 "crash-before-header",
5096 "WAL-HEADER-WRITE-CRASH",
5097 "test_crash_before_header",
5098 ),
5099 );
5100
5101 let new_salts = WalSalts {
5102 salt1: original_salts.salt1.wrapping_add(1),
5103 salt2: original_salts.salt2.wrapping_add(1),
5104 };
5105 let err = wal
5106 .reset(&cx, 1, new_salts, true)
5107 .expect_err("should fail at crash boundary");
5108 assert!(
5109 err.to_string().contains("fault_inject"),
5110 "error identifies the fault hook: {err}"
5111 );
5112
5113 crate::fault_hooks::clear_crash_boundary();
5114
5115 wal.close(&cx).expect("close after crash");
5116
5117 let file2 = open_wal_file(&vfs, &cx);
5118 let recovered = WalFile::open(&cx, file2).expect("reopen");
5119 assert_eq!(
5120 recovered.frame_count(),
5121 1,
5122 "header was never rewritten so old generation with 1 frame persists"
5123 );
5124 assert_eq!(
5125 recovered.generation_identity().salts,
5126 original_salts,
5127 "salts unchanged — reset never wrote new header"
5128 );
5129 recovered.close(&cx).expect("close recovered");
5130 }
5131
5132 #[test]
5133 fn crash_after_frame_append_before_fsync_frames_on_disk_but_not_durable() {
5134 let _guard = FAULT_TEST_LOCK.lock().unwrap();
5135 let cx = test_cx();
5136 let vfs = MemoryVfs::new();
5137 let file = open_wal_file(&vfs, &cx);
5138 let mut wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create");
5139
5140 crate::fault_hooks::arm_crash_boundary(
5141 crate::fault_hooks::CrashBoundary::AfterWalFrameAppendBeforeFsync,
5142 crate::fault_hooks::FaultHookArm::new(
5143 "crash-after-append",
5144 "WAL-APPEND-FSYNC-CRASH",
5145 "test_crash_after_append_before_fsync",
5146 ),
5147 );
5148
5149 let p1 = sample_page(0xEE);
5150 let frames = [frame_ref(1, &p1, 1)];
5151 let err = wal
5152 .append_frames(&cx, &frames)
5153 .expect_err("should fail after append but before fsync");
5154 assert!(
5155 err.to_string().contains("fault_inject"),
5156 "error identifies the fault hook: {err}"
5157 );
5158
5159 crate::fault_hooks::clear_crash_boundary();
5160
5161 assert_eq!(
5162 wal.last_fsynced_frame_count(),
5163 0,
5164 "no fsync happened so fsynced count is still zero"
5165 );
5166
5167 wal.close(&cx).expect("close after crash");
5168
5169 let file2 = open_wal_file(&vfs, &cx);
5170 let recovered = WalFile::open(&cx, file2).expect("reopen");
5171 assert!(
5172 recovered.frame_count() <= 1,
5173 "frame may or may not survive recovery depending on MemoryVfs behavior (data written but not fsynced)"
5174 );
5175 recovered.close(&cx).expect("close recovered");
5176 }
5177}