1use std::collections::{HashMap, HashSet};
12use std::path::{Path, PathBuf};
13use std::sync::Arc;
14
15use fsqlite_error::{FrankenError, Result};
16use fsqlite_pager::traits::{
17 PreparedWalChecksumSeed, PreparedWalFinalizationState, PreparedWalFrameBatch,
18 PreparedWalFrameMeta, WalFrameRef, WalFuture, WalLogicalReadSnapshot,
19};
20use fsqlite_pager::{
21 CheckpointMode, CheckpointPageWriter, CheckpointResult, ParallelWalCommitReconciliation,
22 WalBackend, WalPublicationSnapshot,
23};
24use fsqlite_types::cx::Cx;
25use fsqlite_types::flags::{AccessFlags, SyncFlags, VfsOpenFlags};
26use fsqlite_types::{CommitSeq, PageNumber, PageSize};
27#[cfg(all(feature = "native", any(unix, windows)))]
28use fsqlite_vfs::DatabaseNamespaceBinding;
29use fsqlite_vfs::{SyncKind, Vfs, VfsFile, VfsWriteCompletion};
30use fsqlite_wal::checkpoint_executor::CheckpointTargetFuture;
31use fsqlite_wal::checksum::{SqliteWalChecksum, WAL_FRAME_HEADER_SIZE, WalChecksumTransform};
32use fsqlite_wal::wal::WalAppendFrameRef;
33use fsqlite_wal::{
34 CheckpointMode as WalCheckpointMode, CheckpointState, CheckpointTarget,
35 PARALLEL_WAL_DURABLE_CERTIFICATE_MAGIC, PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE,
36 ParallelWalCommitCertificate, ParallelWalDurableCertificateRecord,
37 ParallelWalFramePayloadDigestBuilder, TransactionConflictPageBaseline,
38 TransactionConflictSnapshot, WAL_HEADER_SIZE, WalFile, WalGenerationIdentity, WalHeader,
39 WalSalts, execute_checkpoint, validate_wal_header_checksum,
40};
41use tracing::debug;
42#[cfg(all(not(target_arch = "wasm32"), feature = "native"))]
43use tracing::warn;
44
45#[cfg(all(not(target_arch = "wasm32"), feature = "native"))]
46use crate::wal_fec_adapter::{FecCommitHook, FecCommitResult};
47
48#[cfg(test)]
49mod test_support {
50 use std::fmt::Debug;
51 use std::future::Future;
52
53 std::thread_local! {
54 static TEST_RUNTIME: asupersync::runtime::Runtime =
55 asupersync::runtime::RuntimeBuilder::current_thread()
56 .blocking_threads(1, 2)
57 .build()
58 .expect("WAL adapter test runtime should build");
59 }
60
61 fn block_on<F: Future>(future: F) -> F::Output {
62 TEST_RUNTIME.with(|runtime| runtime.block_on(future))
63 }
64
65 pub(super) trait FutureResultTestExt<T, E>:
66 Future<Output = std::result::Result<T, E>> + Sized
67 {
68 fn wait(self) -> std::result::Result<T, E> {
69 block_on(self)
70 }
71
72 fn expect(self, message: &str) -> T
73 where
74 E: Debug,
75 {
76 block_on(self).expect(message)
77 }
78
79 fn expect_err(self, message: &str) -> E
80 where
81 T: Debug,
82 {
83 block_on(self).expect_err(message)
84 }
85 }
86
87 impl<F, T, E> FutureResultTestExt<T, E> for F where
88 F: Future<Output = std::result::Result<T, E>> + Sized
89 {
90 }
91}
92
93#[cfg(test)]
94use self::test_support::FutureResultTestExt;
95
96struct WalWriteCompletionPreflight<'a> {
103 completion: Option<&'a VfsWriteCompletion>,
104}
105
106impl<'a> WalWriteCompletionPreflight<'a> {
107 const fn new(completion: Option<&'a VfsWriteCompletion>) -> Self {
108 Self { completion }
109 }
110
111 fn hand_off(&mut self) {
112 self.completion = None;
113 }
114}
115
116impl Drop for WalWriteCompletionPreflight<'_> {
117 fn drop(&mut self) {
118 if let Some(completion) = self.completion {
119 completion.complete_error();
120 }
121 }
122}
123
124const PAGE_INDEX_MAX_ENTRIES: usize = usize::MAX;
135
136fn sqlite_database_header_page_size(page_one: &[u8]) -> Option<u32> {
137 if page_one.len() < 18 || !page_one.starts_with(b"SQLite format 3\0") {
138 return None;
139 }
140 let encoded = u16::from_be_bytes([page_one[16], page_one[17]]);
141 let decoded = if encoded == 1 {
142 65_536
143 } else {
144 u32::from(encoded)
145 };
146 PageSize::new(decoded).map(PageSize::get)
147}
148
149#[derive(Debug, Clone, Copy, PartialEq, Eq)]
156enum WalPageLookupResolution {
157 AuthoritativeHit { frame_index: usize },
158 AuthoritativeMiss,
159 PartialIndexFallbackHit { frame_index: usize },
160 PartialIndexFallbackMiss,
161}
162
163impl WalPageLookupResolution {
164 #[must_use]
165 const fn frame_index(self) -> Option<usize> {
166 match self {
167 Self::AuthoritativeHit { frame_index }
168 | Self::PartialIndexFallbackHit { frame_index } => Some(frame_index),
169 Self::AuthoritativeMiss | Self::PartialIndexFallbackMiss => None,
170 }
171 }
172
173 #[must_use]
174 const fn lookup_mode(self) -> &'static str {
175 match self {
176 Self::AuthoritativeHit { .. } | Self::AuthoritativeMiss => "authoritative_index",
177 Self::PartialIndexFallbackHit { .. } | Self::PartialIndexFallbackMiss => {
178 "partial_index_fallback"
179 }
180 }
181 }
182
183 #[must_use]
184 const fn fallback_reason(self) -> &'static str {
185 match self {
186 Self::AuthoritativeHit { .. } | Self::AuthoritativeMiss => "none",
187 Self::PartialIndexFallbackHit { .. } | Self::PartialIndexFallbackMiss => {
188 "partial_index_cap"
189 }
190 }
191 }
192}
193
194#[derive(Debug, Clone)]
200struct WalPublishedSnapshot {
201 publication_seq: u64,
202 generation: WalGenerationIdentity,
203 last_commit_frame: Option<usize>,
204 commit_count: u64,
205 page_index: Arc<HashMap<u32, usize>>,
206 index_is_partial: bool,
207}
208
209impl WalPublishedSnapshot {
210 #[must_use]
211 fn empty(publication_seq: u64, generation: WalGenerationIdentity) -> Self {
212 Self {
213 publication_seq,
214 generation,
215 last_commit_frame: None,
216 commit_count: 0,
217 page_index: Arc::new(HashMap::new()),
218 index_is_partial: false,
219 }
220 }
221}
222
223#[must_use]
224fn wal_publication_snapshot_from_published(
225 snapshot: &WalPublishedSnapshot,
226) -> WalPublicationSnapshot {
227 WalPublicationSnapshot {
228 publication_seq: snapshot.publication_seq,
229 generation: snapshot.generation,
230 last_commit_frame: snapshot.last_commit_frame,
231 commit_count: snapshot.commit_count,
232 latest_frame_entries: snapshot.page_index.len(),
233 index_is_partial: snapshot.index_is_partial,
234 }
235}
236
237#[derive(Debug, Clone, Copy)]
238struct PendingPublicationFrame {
239 page_number: u32,
240 frame_index: usize,
241 is_commit: bool,
242}
243
244pub struct WalBackendAdapter<F: VfsFile> {
245 wal: WalFile<F>,
246 refresh_before_append: bool,
248 published_snapshot: WalPublishedSnapshot,
250 next_publication_seq: u64,
252 read_snapshot: Option<WalPublishedSnapshot>,
254 pending_publication_frames: Vec<PendingPublicationFrame>,
256 pending_publication_commit: Option<usize>,
262 pending_publication_generation: Option<WalGenerationIdentity>,
267 #[cfg(all(not(target_arch = "wasm32"), feature = "native"))]
269 fec_hook: Option<FecCommitHook>,
270 #[cfg(all(not(target_arch = "wasm32"), feature = "native"))]
272 fec_pending: Vec<FecCommitResult>,
273 page_index_cap: usize,
277}
278
279impl<F: VfsFile> WalBackendAdapter<F> {
280 #[must_use]
282 pub fn new(wal: WalFile<F>) -> Self {
283 let generation = wal.generation_identity();
284 Self {
285 wal,
286 refresh_before_append: true,
287 published_snapshot: WalPublishedSnapshot::empty(0, generation),
288 next_publication_seq: 1,
289 read_snapshot: None,
290 pending_publication_frames: Vec::new(),
291 pending_publication_commit: None,
292 pending_publication_generation: None,
293 #[cfg(all(not(target_arch = "wasm32"), feature = "native"))]
294 fec_hook: None,
295 #[cfg(all(not(target_arch = "wasm32"), feature = "native"))]
296 fec_pending: Vec::new(),
297 page_index_cap: PAGE_INDEX_MAX_ENTRIES,
298 }
299 }
300
301 #[must_use]
303 #[cfg(all(not(target_arch = "wasm32"), feature = "native"))]
304 pub fn with_fec_hook(wal: WalFile<F>, hook: FecCommitHook) -> Self {
305 let generation = wal.generation_identity();
306 Self {
307 wal,
308 refresh_before_append: true,
309 published_snapshot: WalPublishedSnapshot::empty(0, generation),
310 next_publication_seq: 1,
311 read_snapshot: None,
312 pending_publication_frames: Vec::new(),
313 pending_publication_commit: None,
314 pending_publication_generation: None,
315 fec_hook: Some(hook),
316 fec_pending: Vec::new(),
317 page_index_cap: PAGE_INDEX_MAX_ENTRIES,
318 }
319 }
320
321 #[must_use]
331 pub fn has_pending_publication(&self) -> bool {
332 self.pending_publication_commit.is_some() || !self.pending_publication_frames.is_empty()
333 }
334
335 pub fn into_inner(self) -> Result<WalFile<F>> {
345 if self.has_pending_publication() {
346 return Err(FrankenError::Busy);
347 }
348 Ok(self.wal)
349 }
350
351 #[must_use]
353 pub fn inner(&self) -> &WalFile<F> {
354 &self.wal
355 }
356
357 pub fn inner_mut(&mut self) -> Result<&mut WalFile<F>> {
366 if self.has_pending_publication() {
367 return Err(FrankenError::Busy);
368 }
369 self.invalidate_publication();
370 Ok(&mut self.wal)
371 }
372
373 #[must_use]
379 pub fn published_snapshot(&self) -> WalPublicationSnapshot {
380 wal_publication_snapshot_from_published(&self.published_snapshot)
381 }
382
383 #[must_use]
385 pub fn pinned_read_snapshot(&self) -> Option<WalPublicationSnapshot> {
386 self.read_snapshot
387 .as_ref()
388 .map(wal_publication_snapshot_from_published)
389 }
390
391 pub async fn refresh_published_snapshot(&mut self, cx: &Cx) -> Result<WalPublicationSnapshot> {
394 self.wal.refresh(cx).await?;
395 self.publish_latest_committed_snapshot(cx, "refresh_published_snapshot")
396 .await?;
397 Ok(self.published_snapshot())
398 }
399
400 fn invalidate_publication(&mut self) {
402 self.read_snapshot = None;
403 self.discard_pending_publication();
404 self.published_snapshot = WalPublishedSnapshot::empty(
405 self.published_snapshot.publication_seq,
406 self.published_snapshot.generation,
407 );
408 }
409
410 async fn publish_visible_snapshot(
416 &mut self,
417 cx: &Cx,
418 last_commit_frame: Option<usize>,
419 scenario_id: &'static str,
420 ) -> Result<()> {
421 let generation = self.wal.generation_identity();
422 if self.published_snapshot.generation == generation
423 && self.published_snapshot.last_commit_frame == last_commit_frame
424 {
425 return Ok(());
426 }
427
428 let previous_generation = self.published_snapshot.generation;
429 let previous_last_commit = self.published_snapshot.last_commit_frame;
430 let previous_commit_count = if previous_generation == generation {
431 self.published_snapshot.commit_count
432 } else {
433 0
434 };
435 let mut page_index = if previous_generation == generation {
436 std::mem::replace(
437 &mut self.published_snapshot.page_index,
438 Arc::new(HashMap::new()),
439 )
440 } else {
441 Arc::new(HashMap::new())
442 };
443 let mut index_is_partial = if previous_generation == generation {
444 self.published_snapshot.index_is_partial
445 } else {
446 false
447 };
448
449 let frame_delta_count = match (previous_last_commit, last_commit_frame) {
450 (Some(prev), Some(curr)) if curr >= prev => curr.saturating_sub(prev),
451 (Some(_) | None, Some(curr)) => curr.saturating_add(1),
452 (Some(prev), None) => prev.saturating_add(1),
453 (None, None) => 0,
454 };
455
456 let scan_result = match last_commit_frame {
457 None => {
458 Arc::make_mut(&mut page_index).clear();
459 index_is_partial = false;
460 Ok(0)
461 }
462 Some(current_last_commit) => {
463 let (start, base_commit_count) =
464 match (previous_generation == generation, previous_last_commit) {
465 (true, Some(previous_last_commit))
466 if previous_last_commit < current_last_commit =>
467 {
468 (
469 previous_last_commit.saturating_add(1),
470 previous_commit_count,
471 )
472 }
473 (true, Some(previous_last_commit))
474 if previous_last_commit == current_last_commit =>
475 {
476 (current_last_commit.saturating_add(1), previous_commit_count)
477 }
478 _ => {
479 Arc::make_mut(&mut page_index).clear();
480 index_is_partial = false;
481 (0, 0)
482 }
483 };
484 if start <= current_last_commit {
485 self.index_range_and_count_commits(
486 cx,
487 Arc::make_mut(&mut page_index),
488 &mut index_is_partial,
489 start,
490 current_last_commit,
491 )
492 .await
493 .map(|delta| base_commit_count.saturating_add(delta))
494 } else {
495 Ok(base_commit_count)
496 }
497 }
498 };
499 let commit_count = match scan_result {
500 Ok(commit_count) => commit_count,
501 Err(error) => {
502 if previous_generation == generation {
503 self.published_snapshot.page_index = page_index;
504 }
505 return Err(error);
506 }
507 };
508
509 let publication_seq = self.next_publication_seq;
510 self.next_publication_seq = self.next_publication_seq.saturating_add(1);
511 let latest_frame_entries = page_index.len();
512 self.published_snapshot = WalPublishedSnapshot {
513 publication_seq,
514 generation,
515 last_commit_frame,
516 commit_count,
517 page_index,
518 index_is_partial,
519 };
520
521 tracing::trace!(
522 target: "fsqlite.wal_publication",
523 trace_id = cx.trace_id(),
524 run_id = "wal-publication",
525 scenario_id,
526 wal_generation = generation.checkpoint_seq,
527 wal_salt1 = generation.salts.salt1,
528 wal_salt2 = generation.salts.salt2,
529 publication_seq,
530 frame_delta_count,
531 latest_frame_entries,
532 snapshot_age = 0_u64,
533 lookup_mode = "published_visibility_map",
534 fallback_reason = if index_is_partial {
535 "partial_index_cap"
536 } else {
537 "none"
538 },
539 "published WAL visibility snapshot"
540 );
541
542 Ok(())
543 }
544
545 async fn resolve_visible_frame(
552 &self,
553 cx: &Cx,
554 snapshot: &WalPublishedSnapshot,
555 page_number: u32,
556 ) -> Result<WalPageLookupResolution> {
557 match snapshot.page_index.get(&page_number) {
558 Some(&frame_index) => Ok(WalPageLookupResolution::AuthoritativeHit { frame_index }),
559 None if !snapshot.index_is_partial => Ok(WalPageLookupResolution::AuthoritativeMiss),
560 None => match snapshot.last_commit_frame {
561 Some(last_commit_frame) => {
562 match self
563 .scan_backwards_for_page(cx, page_number, last_commit_frame)
564 .await?
565 {
566 Some(frame_index) => {
567 Ok(WalPageLookupResolution::PartialIndexFallbackHit { frame_index })
568 }
569 None => Ok(WalPageLookupResolution::PartialIndexFallbackMiss),
570 }
571 }
572 None => Ok(WalPageLookupResolution::AuthoritativeMiss),
573 },
574 }
575 }
576
577 async fn index_range_and_count_commits(
583 &self,
584 cx: &Cx,
585 page_index: &mut HashMap<u32, usize>,
586 index_is_partial: &mut bool,
587 start: usize,
588 end: usize,
589 ) -> Result<u64> {
590 if start > end {
591 return Ok(0);
592 }
593
594 let mut commit_count = 0_u64;
595 for frame_index in start..=end {
596 let header = self.wal.read_frame_header(cx, frame_index).await?;
597 if page_index.len() < self.page_index_cap
600 || page_index.contains_key(&header.page_number)
601 {
602 page_index.insert(header.page_number, frame_index);
603 } else {
604 *index_is_partial = true;
608 }
609 if header.is_commit() {
610 commit_count = commit_count.saturating_add(1);
611 }
612 }
613 Ok(commit_count)
614 }
615
616 async fn scan_backwards_for_page(
623 &self,
624 cx: &Cx,
625 page_number: u32,
626 last_commit_frame: usize,
627 ) -> Result<Option<usize>> {
628 for frame_index in (0..=last_commit_frame).rev() {
629 let header = self.wal.read_frame_header(cx, frame_index).await?;
630 if header.page_number == page_number {
631 return Ok(Some(frame_index));
632 }
633 }
634 Ok(None)
635 }
636
637 #[cfg(all(not(target_arch = "wasm32"), feature = "native"))]
639 pub fn take_fec_pending(&mut self) -> Vec<FecCommitResult> {
640 std::mem::take(&mut self.fec_pending)
641 }
642
643 #[must_use]
645 #[cfg(all(not(target_arch = "wasm32"), feature = "native"))]
646 pub fn fec_enabled(&self) -> bool {
647 self.fec_hook
648 .as_ref()
649 .is_some_and(FecCommitHook::is_enabled)
650 }
651
652 #[cfg(all(not(target_arch = "wasm32"), feature = "native"))]
654 pub fn fec_discard(&mut self) {
655 if let Some(hook) = &mut self.fec_hook {
656 hook.discard_buffered();
657 }
658 }
659
660 #[cfg(test)]
662 fn set_page_index_cap(&mut self, cap: usize) {
663 self.page_index_cap = cap;
664 self.invalidate_publication();
666 }
667
668 #[must_use]
669 fn current_prepared_finalization_state(&self) -> PreparedWalFinalizationState {
670 let generation = self.wal.generation_identity();
671 let seed = self.wal.running_checksum();
672 PreparedWalFinalizationState {
673 checkpoint_seq: generation.checkpoint_seq,
674 salt1: generation.salts.salt1,
675 salt2: generation.salts.salt2,
676 start_frame_index: self.wal.frame_count(),
677 seed: PreparedWalChecksumSeed {
678 s1: seed.s1,
679 s2: seed.s2,
680 },
681 }
682 }
683
684 #[must_use]
685 fn prepared_batch_matches_current_state(&self, prepared: &PreparedWalFrameBatch) -> bool {
686 prepared
687 .finalized_for
688 .is_some_and(|state| state == self.current_prepared_finalization_state())
689 }
690
691 async fn prepared_batch_matches_disk_state(
692 &self,
693 cx: &Cx,
694 prepared: &PreparedWalFrameBatch,
695 ) -> Result<bool> {
696 let Some(state) = prepared.finalized_for else {
697 return Ok(false);
698 };
699 let generation = WalGenerationIdentity {
700 checkpoint_seq: state.checkpoint_seq,
701 salts: fsqlite_wal::checksum::WalSalts {
702 salt1: state.salt1,
703 salt2: state.salt2,
704 },
705 };
706 self.wal
707 .prepared_append_window_still_current(cx, generation, state.start_frame_index)
708 .await
709 }
710
711 fn checksum_transforms_for_prepared(
712 prepared: &PreparedWalFrameBatch,
713 ) -> Vec<WalChecksumTransform> {
714 prepared
715 .checksum_transforms
716 .iter()
717 .map(|transform| WalChecksumTransform {
718 a11: transform.a11,
719 a12: transform.a12,
720 a21: transform.a21,
721 a22: transform.a22,
722 c1: transform.c1,
723 c2: transform.c2,
724 })
725 .collect()
726 }
727
728 fn finalize_prepared_batch_against_current_state(
729 &self,
730 prepared: &mut PreparedWalFrameBatch,
731 ) -> Result<()> {
732 let checksum_transforms = Self::checksum_transforms_for_prepared(prepared);
733 let final_running_checksum = self
734 .wal
735 .finalize_prepared_frame_bytes(&mut prepared.frame_bytes, &checksum_transforms)?;
736 prepared.finalized_for = Some(self.current_prepared_finalization_state());
737 prepared.finalized_running_checksum = Some(PreparedWalChecksumSeed {
738 s1: final_running_checksum.s1,
739 s2: final_running_checksum.s2,
740 });
741 Ok(())
742 }
743
744 fn finalized_running_checksum(prepared: &PreparedWalFrameBatch) -> Result<SqliteWalChecksum> {
745 let Some(checksum) = prepared.finalized_running_checksum else {
746 return Err(FrankenError::internal(
747 "prepared WAL batch missing finalized running checksum",
748 ));
749 };
750 Ok(SqliteWalChecksum {
751 s1: checksum.s1,
752 s2: checksum.s2,
753 })
754 }
755
756 async fn publish_latest_committed_snapshot(
757 &mut self,
758 cx: &Cx,
759 scenario_id: &'static str,
760 ) -> Result<()> {
761 let last_commit_frame = self.wal.last_commit_frame(cx)?;
762 let last_commit_frame = if self.pending_publication_commit.is_some() {
781 let durable_frames = self.wal.last_fsynced_frame_count();
782 last_commit_frame.filter(|frame| {
783 frame
784 .checked_add(1)
785 .is_some_and(|frame_count| frame_count <= durable_frames)
786 })
787 } else {
788 last_commit_frame
789 };
790 self.publish_visible_snapshot(cx, last_commit_frame, scenario_id)
791 .await
792 }
793
794 async fn synchronize_publication_before_append(
795 &mut self,
796 cx: &Cx,
797 scenario_id: &'static str,
798 ) -> Result<()> {
799 if self.has_pending_publication() {
807 return Err(FrankenError::Busy);
808 }
809 self.wal.refresh(cx).await?;
810 self.discard_pending_publication();
811 self.publish_latest_committed_snapshot(cx, scenario_id)
812 .await
813 }
814
815 fn discard_pending_publication(&mut self) {
817 self.pending_publication_frames.clear();
818 self.pending_publication_commit = None;
819 self.pending_publication_generation = None;
820 }
821
822 fn stage_pending_commit_publication(&mut self, last_commit_frame: usize) -> Result<()> {
829 let generation = self.wal.generation_identity();
830 if self
834 .pending_publication_generation
835 .is_some_and(|staged| staged != generation)
836 {
837 return Err(FrankenError::WalCorrupt {
838 detail: "cannot stage a commit horizon across differing WAL generations".to_owned(),
839 });
840 }
841 let staged = self
842 .pending_publication_commit
843 .map_or(last_commit_frame, |staged| staged.max(last_commit_frame));
844 self.pending_publication_commit = Some(staged);
845 self.pending_publication_generation = Some(generation);
846 Ok(())
847 }
848
849 fn assert_pending_horizon_matches_wal(
857 &mut self,
858 cx: &Cx,
859 last_commit_frame: usize,
860 ) -> Result<()> {
861 let generation = self.wal.generation_identity();
862 if self
863 .pending_publication_generation
864 .is_some_and(|staged| staged != generation)
865 {
866 return Err(FrankenError::WalCorrupt {
867 detail: "WAL generation changed before the staged commit horizon was published"
868 .to_owned(),
869 });
870 }
871
872 let frame_count = self.wal.frame_count();
873 if last_commit_frame >= frame_count {
874 return Err(FrankenError::WalCorrupt {
875 detail: format!(
876 "staged commit horizon {last_commit_frame} exceeds WAL frame count {frame_count}"
877 ),
878 });
879 }
880
881 let live_last_commit = self.wal.last_commit_frame(cx)?;
882 if live_last_commit.is_none_or(|live| live < last_commit_frame) {
883 return Err(FrankenError::WalCorrupt {
884 detail: format!(
885 "WAL does not report staged commit horizon {last_commit_frame} as committed"
886 ),
887 });
888 }
889
890 if self
891 .pending_publication_frames
892 .iter()
893 .any(|frame| frame.frame_index >= frame_count)
894 {
895 return Err(FrankenError::WalCorrupt {
896 detail: format!(
897 "a staged publication frame lies beyond WAL frame count {frame_count}"
898 ),
899 });
900 }
901
902 Ok(())
903 }
904
905 fn assert_publish_safe(&mut self, cx: &Cx, last_commit_frame: usize) -> Result<()> {
906 self.assert_pending_horizon_matches_wal(cx, last_commit_frame)?;
907
908 let publish_frame_count =
911 last_commit_frame
912 .checked_add(1)
913 .ok_or_else(|| FrankenError::WalCorrupt {
914 detail: "staged commit horizon overflows the publishable frame count"
915 .to_owned(),
916 })?;
917 self.wal.assert_publish_safe(publish_frame_count)?;
918
919 Ok(())
920 }
921
922 fn publish_authorized_deferred_commit(&mut self, cx: &Cx) -> Result<()> {
929 let Some(last_commit_frame) = self.pending_publication_commit else {
930 return Ok(());
931 };
932 self.assert_pending_horizon_matches_wal(cx, last_commit_frame)?;
933 self.publish_pending_commit_snapshot(cx, last_commit_frame, "authorized_deferred_commit");
934 self.pending_publication_commit = None;
935 self.pending_publication_generation = None;
936 Ok(())
937 }
938
939 fn publish_pending_after_sync(&mut self, cx: &Cx) -> Result<()> {
946 let Some(last_commit_frame) = self.pending_publication_commit else {
947 return Ok(());
948 };
949 self.assert_publish_safe(cx, last_commit_frame)?;
950 self.publish_pending_commit_snapshot(cx, last_commit_frame, "sync_publish_commit");
951 self.pending_publication_commit = None;
952 self.pending_publication_generation = None;
953 Ok(())
954 }
955
956 fn record_appended_frames<I>(&mut self, start_frame_index: usize, frames: I) -> Option<usize>
957 where
958 I: IntoIterator<Item = (u32, u32)>,
959 {
960 let mut last_commit_frame = None;
961 for (offset, (page_number, db_size_if_commit)) in frames.into_iter().enumerate() {
962 let frame_index = start_frame_index.saturating_add(offset);
963 self.pending_publication_frames
964 .push(PendingPublicationFrame {
965 page_number,
966 frame_index,
967 is_commit: db_size_if_commit != 0,
968 });
969 if db_size_if_commit != 0 {
970 last_commit_frame = Some(frame_index);
971 }
972 }
973 last_commit_frame
974 }
975
976 fn publish_pending_commit_snapshot(
983 &mut self,
984 cx: &Cx,
985 last_commit_frame: usize,
986 scenario_id: &'static str,
987 ) {
988 let generation = self.wal.generation_identity();
989 let previous_last_commit = self.published_snapshot.last_commit_frame;
990 let can_extend_previous = self.published_snapshot.generation == generation
991 && self
992 .published_snapshot
993 .last_commit_frame
994 .is_none_or(|previous_last_commit| previous_last_commit < last_commit_frame);
995 let mut page_index = if can_extend_previous {
996 std::mem::replace(
997 &mut self.published_snapshot.page_index,
998 Arc::new(HashMap::new()),
999 )
1000 } else {
1001 Arc::new(HashMap::new())
1002 };
1003 let mut index_is_partial = if can_extend_previous {
1004 self.published_snapshot.index_is_partial
1005 } else {
1006 false
1007 };
1008 let previous_last_commit = if can_extend_previous {
1009 previous_last_commit
1010 } else {
1011 None
1012 };
1013 let previous_commit_count = if can_extend_previous {
1014 self.published_snapshot.commit_count
1015 } else {
1016 0
1017 };
1018
1019 let mut frame_delta_count = 0_usize;
1020 let mut commit_delta_count = 0_u64;
1021 for frame in &self.pending_publication_frames {
1022 if previous_last_commit
1023 .is_some_and(|previous_last_commit| frame.frame_index <= previous_last_commit)
1024 || frame.frame_index > last_commit_frame
1025 {
1026 continue;
1027 }
1028
1029 frame_delta_count = frame_delta_count.saturating_add(1);
1030 let page_index_map = Arc::make_mut(&mut page_index);
1031 if page_index_map.len() < self.page_index_cap
1032 || page_index_map.contains_key(&frame.page_number)
1033 {
1034 page_index_map.insert(frame.page_number, frame.frame_index);
1035 } else {
1036 index_is_partial = true;
1037 }
1038 if frame.is_commit {
1039 commit_delta_count = commit_delta_count.saturating_add(1);
1040 }
1041 }
1042
1043 if frame_delta_count == 0 {
1044 if can_extend_previous {
1051 self.published_snapshot.page_index = page_index;
1052 }
1053 self.pending_publication_frames.clear();
1054 return;
1055 }
1056
1057 let publication_seq = self.next_publication_seq;
1058 self.next_publication_seq = self.next_publication_seq.saturating_add(1);
1059 let latest_frame_entries = page_index.len();
1060 self.published_snapshot = WalPublishedSnapshot {
1061 publication_seq,
1062 generation,
1063 last_commit_frame: Some(last_commit_frame),
1064 commit_count: previous_commit_count.saturating_add(commit_delta_count),
1065 page_index,
1066 index_is_partial,
1067 };
1068 self.pending_publication_frames.clear();
1069
1070 tracing::trace!(
1071 target: "fsqlite.wal_publication",
1072 trace_id = cx.trace_id(),
1073 run_id = "wal-publication",
1074 scenario_id,
1075 wal_generation = generation.checkpoint_seq,
1076 wal_salt1 = generation.salts.salt1,
1077 wal_salt2 = generation.salts.salt2,
1078 publication_seq,
1079 frame_delta_count,
1080 latest_frame_entries,
1081 snapshot_age = 0_u64,
1082 lookup_mode = "published_visibility_map",
1083 fallback_reason = if index_is_partial {
1084 "partial_index_cap"
1085 } else {
1086 "none"
1087 },
1088 "published WAL visibility snapshot from commit path"
1089 );
1090 }
1091}
1092
1093fn to_wal_mode(mode: CheckpointMode) -> WalCheckpointMode {
1095 match mode {
1096 CheckpointMode::Passive => WalCheckpointMode::Passive,
1097 CheckpointMode::Full => WalCheckpointMode::Full,
1098 CheckpointMode::Restart => WalCheckpointMode::Restart,
1099 CheckpointMode::Truncate => WalCheckpointMode::Truncate,
1100 }
1101}
1102
1103impl<F: VfsFile> WalBackend for WalBackendAdapter<F> {
1104 fn begin_transaction<'a>(&'a mut self, cx: &'a Cx) -> WalFuture<'a, ()> {
1105 Box::pin(async move {
1106 if self.has_pending_publication() {
1112 return Err(FrankenError::Busy);
1113 }
1114 self.wal.refresh(cx).await?;
1117 self.publish_latest_committed_snapshot(cx, "begin_transaction")
1118 .await?;
1119 self.read_snapshot = Some(self.published_snapshot.clone());
1120 self.refresh_before_append = true;
1121 Ok(())
1122 })
1123 }
1124
1125 fn published_snapshot(&self) -> Option<WalPublicationSnapshot> {
1126 Some(Self::published_snapshot(self))
1127 }
1128
1129 fn pinned_read_snapshot(&self) -> Option<WalPublicationSnapshot> {
1130 Self::pinned_read_snapshot(self)
1131 }
1132
1133 fn refresh_published_snapshot<'a>(
1134 &'a mut self,
1135 cx: &'a Cx,
1136 ) -> WalFuture<'a, Option<WalPublicationSnapshot>> {
1137 Box::pin(async move { Self::refresh_published_snapshot(self, cx).await.map(Some) })
1138 }
1139
1140 fn publish_authorized_deferred_commit<'a>(&'a mut self, cx: &'a Cx) -> WalFuture<'a, ()> {
1141 Box::pin(async move { Self::publish_authorized_deferred_commit(self, cx) })
1142 }
1143
1144 fn append_frame<'a>(
1145 &'a mut self,
1146 cx: &'a Cx,
1147 page_number: u32,
1148 page_data: &'a [u8],
1149 db_size_if_commit: u32,
1150 ) -> WalFuture<'a, ()> {
1151 Box::pin(async move {
1152 if self.refresh_before_append {
1153 self.synchronize_publication_before_append(cx, "append_frame_pre_refresh")
1157 .await?;
1158 }
1159 let start_frame_index = self.wal.frame_count();
1160 self.wal
1161 .append_frame(cx, page_number, page_data, db_size_if_commit)
1162 .await?;
1163 self.refresh_before_append = false;
1164 let last_commit_frame =
1165 self.record_appended_frames(start_frame_index, [(page_number, db_size_if_commit)]);
1166
1167 #[cfg(all(not(target_arch = "wasm32"), feature = "native"))]
1170 if let Some(hook) = &mut self.fec_hook {
1171 match hook.on_frame(cx, page_number, page_data, db_size_if_commit) {
1172 Ok(Some(result)) => {
1173 debug!(
1174 pages = result.page_numbers.len(),
1175 k_source = result.k_source,
1176 symbols = result.symbols.len(),
1177 "FEC commit group encoded"
1178 );
1179 self.fec_pending.push(result);
1180 }
1181 Ok(None) => {}
1182 Err(e) => {
1183 warn!(error = %e, "FEC encoding failed; commit proceeds without repair symbols");
1185 }
1186 }
1187 }
1188
1189 if let Some(last_commit_frame) = last_commit_frame {
1190 self.stage_pending_commit_publication(last_commit_frame)?;
1193 }
1194
1195 Ok(())
1196 })
1197 }
1198
1199 fn append_frames<'a>(
1200 &'a mut self,
1201 cx: &'a Cx,
1202 frames: &'a [WalFrameRef<'a>],
1203 ) -> WalFuture<'a, ()> {
1204 Box::pin(async move {
1205 if frames.is_empty() {
1206 return Ok(());
1207 }
1208
1209 if self.refresh_before_append {
1210 self.synchronize_publication_before_append(cx, "append_frames_pre_refresh")
1211 .await?;
1212 }
1213
1214 let start_frame_index = self.wal.frame_count();
1215 let mut wal_frames = Vec::with_capacity(frames.len());
1216 for frame in frames {
1217 wal_frames.push(WalAppendFrameRef {
1218 page_number: frame.page_number,
1219 page_data: frame.page_data,
1220 db_size_if_commit: frame.db_size_if_commit,
1221 });
1222 }
1223 self.wal.append_frames(cx, &wal_frames).await?;
1224 self.refresh_before_append = false;
1225 let last_commit_frame = self.record_appended_frames(
1226 start_frame_index,
1227 frames
1228 .iter()
1229 .map(|frame| (frame.page_number, frame.db_size_if_commit)),
1230 );
1231
1232 #[cfg(all(not(target_arch = "wasm32"), feature = "native"))]
1233 if let Some(hook) = &mut self.fec_hook {
1234 for frame in frames {
1235 match hook.on_frame(
1236 cx,
1237 frame.page_number,
1238 frame.page_data,
1239 frame.db_size_if_commit,
1240 ) {
1241 Ok(Some(result)) => {
1242 debug!(
1243 pages = result.page_numbers.len(),
1244 k_source = result.k_source,
1245 symbols = result.symbols.len(),
1246 "FEC commit group encoded"
1247 );
1248 self.fec_pending.push(result);
1249 }
1250 Ok(None) => {}
1251 Err(e) => {
1252 warn!(
1253 error = %e,
1254 "FEC encoding failed; commit proceeds without repair symbols"
1255 );
1256 }
1257 }
1258 }
1259 }
1260
1261 if let Some(last_commit_frame) = last_commit_frame {
1262 self.stage_pending_commit_publication(last_commit_frame)?;
1264 }
1265
1266 Ok(())
1267 })
1268 }
1269
1270 fn append_frames_tracked<'a>(
1271 &'a mut self,
1272 cx: &'a Cx,
1273 frames: &'a [WalFrameRef<'a>],
1274 completion: VfsWriteCompletion,
1275 ) -> WalFuture<'a, ()> {
1276 Box::pin(async move {
1277 let mut preflight = WalWriteCompletionPreflight::new(Some(&completion));
1278 if frames.is_empty() {
1279 completion.complete_success();
1280 preflight.hand_off();
1281 return Ok(());
1282 }
1283
1284 if self.refresh_before_append {
1285 self.synchronize_publication_before_append(cx, "append_frames_pre_refresh")
1286 .await?;
1287 }
1288
1289 let start_frame_index = self.wal.frame_count();
1290 let mut wal_frames = Vec::with_capacity(frames.len());
1291 for frame in frames {
1292 wal_frames.push(WalAppendFrameRef {
1293 page_number: frame.page_number,
1294 page_data: frame.page_data,
1295 db_size_if_commit: frame.db_size_if_commit,
1296 });
1297 }
1298 preflight.hand_off();
1299 drop(preflight);
1300 self.wal
1301 .append_frames_tracked(cx, &wal_frames, completion)
1302 .await?;
1303 self.refresh_before_append = false;
1304 let last_commit_frame = self.record_appended_frames(
1305 start_frame_index,
1306 frames
1307 .iter()
1308 .map(|frame| (frame.page_number, frame.db_size_if_commit)),
1309 );
1310
1311 #[cfg(all(not(target_arch = "wasm32"), feature = "native"))]
1312 if let Some(hook) = &mut self.fec_hook {
1313 for frame in frames {
1314 match hook.on_frame(
1315 cx,
1316 frame.page_number,
1317 frame.page_data,
1318 frame.db_size_if_commit,
1319 ) {
1320 Ok(Some(result)) => {
1321 debug!(
1322 pages = result.page_numbers.len(),
1323 k_source = result.k_source,
1324 symbols = result.symbols.len(),
1325 "FEC commit group encoded"
1326 );
1327 self.fec_pending.push(result);
1328 }
1329 Ok(None) => {}
1330 Err(e) => {
1331 warn!(
1332 error = %e,
1333 "FEC encoding failed; commit proceeds without repair symbols"
1334 );
1335 }
1336 }
1337 }
1338 }
1339
1340 if let Some(last_commit_frame) = last_commit_frame {
1341 self.stage_pending_commit_publication(last_commit_frame)?;
1343 }
1344
1345 Ok(())
1346 })
1347 }
1348
1349 fn prepare_append_frames(
1350 &self,
1351 frames: &[WalFrameRef<'_>],
1352 ) -> Result<Option<PreparedWalFrameBatch>> {
1353 if frames.is_empty() {
1354 return Ok(None);
1355 }
1356
1357 let mut frame_bytes = Vec::new();
1358 let mut checksum_transforms = Vec::new();
1359 let last_commit_frame_offset = self.wal.prepare_frame_bytes_with_transforms_into(
1360 frames.len(),
1361 frames.iter().map(|frame| WalAppendFrameRef {
1362 page_number: frame.page_number,
1363 page_data: frame.page_data,
1364 db_size_if_commit: frame.db_size_if_commit,
1365 }),
1366 &mut frame_bytes,
1367 &mut checksum_transforms,
1368 )?;
1369 let frame_metas = frames
1370 .iter()
1371 .map(|frame| PreparedWalFrameMeta {
1372 page_number: frame.page_number,
1373 db_size_if_commit: frame.db_size_if_commit,
1374 })
1375 .collect();
1376
1377 Ok(Some(PreparedWalFrameBatch {
1378 frame_size: self.wal.frame_size(),
1379 page_data_offset: WAL_FRAME_HEADER_SIZE,
1380 big_endian_checksum: self.wal.big_endian_checksum(),
1381 frame_metas,
1382 checksum_transforms,
1383 frame_bytes,
1384 last_commit_frame_offset,
1385 finalized_for: None,
1386 finalized_running_checksum: None,
1387 }))
1388 }
1389
1390 fn finalize_prepared_frames(
1391 &self,
1392 _cx: &Cx,
1393 prepared: &mut PreparedWalFrameBatch,
1394 ) -> Result<()> {
1395 if prepared.frame_count() == 0 {
1396 return Ok(());
1397 }
1398 self.finalize_prepared_batch_against_current_state(prepared)
1402 }
1403
1404 fn append_prepared_frames<'a>(
1405 &'a mut self,
1406 cx: &'a Cx,
1407 prepared: &'a mut PreparedWalFrameBatch,
1408 ) -> WalFuture<'a, ()> {
1409 Box::pin(async move {
1410 if prepared.frame_count() == 0 {
1411 return Ok(());
1412 }
1413
1414 let can_reuse_prelock_finalize = self.refresh_before_append
1415 && self.prepared_batch_matches_current_state(prepared)
1416 && self.prepared_batch_matches_disk_state(cx, prepared).await?;
1417 if self.refresh_before_append && !can_reuse_prelock_finalize {
1418 self.synchronize_publication_before_append(cx, "append_prepared_pre_refresh")
1419 .await?;
1420 }
1421
1422 if !self.prepared_batch_matches_current_state(prepared) {
1423 self.finalize_prepared_batch_against_current_state(prepared)?;
1424 }
1425
1426 let start_frame_index = self.wal.frame_count();
1427 self.wal
1428 .append_finalized_prepared_frame_bytes(
1429 cx,
1430 &prepared.frame_bytes,
1431 prepared.frame_count(),
1432 Self::finalized_running_checksum(prepared)?,
1433 prepared.last_commit_frame_offset,
1434 )
1435 .await?;
1436 self.refresh_before_append = false;
1437 let last_commit_frame = self.record_appended_frames(
1438 start_frame_index,
1439 prepared
1440 .frame_metas
1441 .iter()
1442 .map(|frame| (frame.page_number, frame.db_size_if_commit)),
1443 );
1444
1445 #[cfg(all(not(target_arch = "wasm32"), feature = "native"))]
1446 if let Some(hook) = &mut self.fec_hook {
1447 for (index, frame) in prepared.frame_metas.iter().enumerate() {
1448 match hook.on_frame(
1449 cx,
1450 frame.page_number,
1451 prepared.page_data(index),
1452 frame.db_size_if_commit,
1453 ) {
1454 Ok(Some(result)) => {
1455 debug!(
1456 pages = result.page_numbers.len(),
1457 k_source = result.k_source,
1458 symbols = result.symbols.len(),
1459 "FEC commit group encoded"
1460 );
1461 self.fec_pending.push(result);
1462 }
1463 Ok(None) => {}
1464 Err(e) => {
1465 warn!(
1466 error = %e,
1467 "FEC encoding failed; commit proceeds without repair symbols"
1468 );
1469 }
1470 }
1471 }
1472 }
1473
1474 if let Some(last_commit_frame) = last_commit_frame {
1475 self.stage_pending_commit_publication(last_commit_frame)?;
1477 }
1478
1479 Ok(())
1480 })
1481 }
1482
1483 fn append_prepared_frames_tracked<'a>(
1484 &'a mut self,
1485 cx: &'a Cx,
1486 prepared: &'a mut PreparedWalFrameBatch,
1487 completion: VfsWriteCompletion,
1488 ) -> WalFuture<'a, ()> {
1489 Box::pin(async move {
1490 let mut preflight = WalWriteCompletionPreflight::new(Some(&completion));
1491 if prepared.frame_count() == 0 {
1492 completion.complete_success();
1493 preflight.hand_off();
1494 return Ok(());
1495 }
1496
1497 let can_reuse_prelock_finalize = self.refresh_before_append
1498 && self.prepared_batch_matches_current_state(prepared)
1499 && self.prepared_batch_matches_disk_state(cx, prepared).await?;
1500 if self.refresh_before_append && !can_reuse_prelock_finalize {
1501 self.synchronize_publication_before_append(cx, "append_prepared_pre_refresh")
1502 .await?;
1503 }
1504
1505 if !self.prepared_batch_matches_current_state(prepared) {
1506 self.finalize_prepared_batch_against_current_state(prepared)?;
1507 }
1508
1509 let start_frame_index = self.wal.frame_count();
1510 let final_running_checksum = Self::finalized_running_checksum(prepared)?;
1511 preflight.hand_off();
1512 drop(preflight);
1513 self.wal
1514 .append_finalized_prepared_frame_bytes_tracked(
1515 cx,
1516 &prepared.frame_bytes,
1517 prepared.frame_count(),
1518 final_running_checksum,
1519 prepared.last_commit_frame_offset,
1520 completion,
1521 )
1522 .await?;
1523 self.refresh_before_append = false;
1524 let last_commit_frame = self.record_appended_frames(
1525 start_frame_index,
1526 prepared
1527 .frame_metas
1528 .iter()
1529 .map(|frame| (frame.page_number, frame.db_size_if_commit)),
1530 );
1531
1532 #[cfg(all(not(target_arch = "wasm32"), feature = "native"))]
1533 if let Some(hook) = &mut self.fec_hook {
1534 for (index, frame) in prepared.frame_metas.iter().enumerate() {
1535 match hook.on_frame(
1536 cx,
1537 frame.page_number,
1538 prepared.page_data(index),
1539 frame.db_size_if_commit,
1540 ) {
1541 Ok(Some(result)) => {
1542 debug!(
1543 pages = result.page_numbers.len(),
1544 k_source = result.k_source,
1545 symbols = result.symbols.len(),
1546 "FEC commit group encoded"
1547 );
1548 self.fec_pending.push(result);
1549 }
1550 Ok(None) => {}
1551 Err(e) => {
1552 warn!(
1553 error = %e,
1554 "FEC encoding failed; commit proceeds without repair symbols"
1555 );
1556 }
1557 }
1558 }
1559 }
1560
1561 if let Some(last_commit_frame) = last_commit_frame {
1562 self.stage_pending_commit_publication(last_commit_frame)?;
1564 }
1565
1566 Ok(())
1567 })
1568 }
1569
1570 fn read_page<'a>(&'a mut self, cx: &'a Cx, page_number: u32) -> WalFuture<'a, Option<Vec<u8>>> {
1571 Box::pin(async move {
1572 let snapshot = if let Some(snapshot) = self.read_snapshot.clone() {
1573 snapshot
1574 } else {
1575 self.publish_latest_committed_snapshot(cx, "read_page_unpinned")
1576 .await?;
1577 self.published_snapshot.clone()
1578 };
1579 if snapshot.last_commit_frame.is_none() {
1580 return Ok(None);
1581 }
1582 let snapshot_age = self
1583 .published_snapshot
1584 .publication_seq
1585 .saturating_sub(snapshot.publication_seq);
1586
1587 let resolution = self
1588 .resolve_visible_frame(cx, &snapshot, page_number)
1589 .await?;
1590 let Some(frame_index) = resolution.frame_index() else {
1591 debug!(
1592 page_number,
1593 wal_checkpoint_seq = snapshot.generation.checkpoint_seq,
1594 wal_salt1 = snapshot.generation.salts.salt1,
1595 wal_salt2 = snapshot.generation.salts.salt2,
1596 publication_seq = snapshot.publication_seq,
1597 snapshot_age,
1598 lookup_mode = resolution.lookup_mode(),
1599 fallback_reason = resolution.fallback_reason(),
1600 "WAL adapter: page absent from current generation"
1601 );
1602 return Ok(None);
1603 };
1604
1605 let mut frame_buf = vec![0u8; self.wal.frame_size()];
1607 let header = self
1608 .wal
1609 .read_frame_into(cx, frame_index, &mut frame_buf)
1610 .await?;
1611
1612 if header.page_number != page_number {
1615 return Err(FrankenError::WalCorrupt {
1616 detail: format!(
1617 "WAL page index integrity failure: expected page {page_number} \
1618 at frame {frame_index}, found page {}",
1619 header.page_number
1620 ),
1621 });
1622 }
1623
1624 let header_size = fsqlite_wal::checksum::WAL_FRAME_HEADER_SIZE;
1636 let page_size = self.wal.page_size();
1637 frame_buf.copy_within(header_size.., 0);
1638 frame_buf.truncate(page_size);
1639 debug!(
1640 page_number,
1641 frame_index,
1642 wal_checkpoint_seq = snapshot.generation.checkpoint_seq,
1643 wal_salt1 = snapshot.generation.salts.salt1,
1644 wal_salt2 = snapshot.generation.salts.salt2,
1645 publication_seq = snapshot.publication_seq,
1646 snapshot_age,
1647 lookup_mode = resolution.lookup_mode(),
1648 fallback_reason = resolution.fallback_reason(),
1649 "WAL adapter: resolved page from current WAL generation"
1650 );
1651 Ok(Some(frame_buf))
1652 })
1653 }
1654
1655 fn read_page_at_appended_tail<'a>(
1663 &'a mut self,
1664 cx: &'a Cx,
1665 page_number: u32,
1666 ) -> WalFuture<'a, Option<Vec<u8>>> {
1667 Box::pin(async move {
1668 let frame_count = self.wal.frame_count();
1669 let Some(tail_frame) = frame_count.checked_sub(1) else {
1670 return Ok(None);
1671 };
1672 let Some(frame_index) = self
1673 .scan_backwards_for_page(cx, page_number, tail_frame)
1674 .await?
1675 else {
1676 return Ok(None);
1677 };
1678 let mut frame_buf = vec![0u8; self.wal.frame_size()];
1679 let header = self
1680 .wal
1681 .read_frame_into(cx, frame_index, &mut frame_buf)
1682 .await?;
1683 if header.page_number != page_number {
1684 return Err(FrankenError::WalCorrupt {
1685 detail: format!(
1686 "WAL appended-tail scan integrity failure: expected page \
1687 {page_number} at frame {frame_index}, found page {}",
1688 header.page_number
1689 ),
1690 });
1691 }
1692 let header_size = fsqlite_wal::checksum::WAL_FRAME_HEADER_SIZE;
1693 let page_size = self.wal.page_size();
1694 frame_buf.copy_within(header_size.., 0);
1695 frame_buf.truncate(page_size);
1696 Ok(Some(frame_buf))
1697 })
1698 }
1699
1700 fn read_page_pinned<'a>(
1702 &'a self,
1703 cx: &'a Cx,
1704 page_number: u32,
1705 ) -> WalFuture<'a, Option<Vec<u8>>> {
1706 Box::pin(async move {
1707 let snapshot = self.read_snapshot.as_ref().ok_or_else(|| {
1708 FrankenError::internal(
1709 "read_page_pinned called without a pinned read snapshot; \
1710 use read_page(&mut self) or call begin_transaction first",
1711 )
1712 })?;
1713 if snapshot.last_commit_frame.is_none() {
1714 return Ok(None);
1715 }
1716
1717 let resolution = self
1718 .resolve_visible_frame(cx, snapshot, page_number)
1719 .await?;
1720 let Some(frame_index) = resolution.frame_index() else {
1721 return Ok(None);
1722 };
1723
1724 let mut frame_buf = vec![0u8; self.wal.frame_size()];
1725 let header = self
1726 .wal
1727 .read_frame_into(cx, frame_index, &mut frame_buf)
1728 .await?;
1729
1730 if header.page_number != page_number {
1731 return Err(FrankenError::WalCorrupt {
1732 detail: format!(
1733 "WAL page index integrity failure: expected page {page_number} \
1734 at frame {frame_index}, found page {}",
1735 header.page_number
1736 ),
1737 });
1738 }
1739
1740 let header_size = fsqlite_wal::checksum::WAL_FRAME_HEADER_SIZE;
1752 let page_size = self.wal.page_size();
1753 frame_buf.copy_within(header_size.., 0);
1754 frame_buf.truncate(page_size);
1755 Ok(Some(frame_buf))
1756 })
1757 }
1758
1759 fn supports_pinned_reads(&self) -> bool {
1760 self.read_snapshot.is_some()
1761 }
1762
1763 fn committed_txns_since_page<'a>(
1764 &'a mut self,
1765 cx: &'a Cx,
1766 page_number: u32,
1767 ) -> WalFuture<'a, u64> {
1768 Box::pin(async move {
1769 let snapshot = if let Some(snapshot) = self.read_snapshot.clone() {
1770 snapshot
1771 } else {
1772 self.publish_latest_committed_snapshot(cx, "committed_txns_since_page")
1773 .await?;
1774 self.published_snapshot.clone()
1775 };
1776 let Some(last_commit_frame) = snapshot.last_commit_frame else {
1777 return Ok(0);
1778 };
1779
1780 let resolution = self
1781 .resolve_visible_frame(cx, &snapshot, page_number)
1782 .await?;
1783 let Some(last_page_frame) = resolution.frame_index() else {
1784 let mut total_commits = 0_u64;
1785 for frame_index in 0..=last_commit_frame {
1786 if self
1787 .wal
1788 .read_frame_header(cx, frame_index)
1789 .await?
1790 .is_commit()
1791 {
1792 total_commits = total_commits.saturating_add(1);
1793 }
1794 }
1795 return Ok(total_commits);
1796 };
1797
1798 let mut page_commit_frame = None;
1799 for frame_index in last_page_frame..=last_commit_frame {
1800 if self
1801 .wal
1802 .read_frame_header(cx, frame_index)
1803 .await?
1804 .is_commit()
1805 {
1806 page_commit_frame = Some(frame_index);
1807 break;
1808 }
1809 }
1810
1811 let Some(page_commit_frame) = page_commit_frame else {
1812 return Ok(0);
1813 };
1814
1815 let mut committed_txns_after_page = 0_u64;
1816 for frame_index in page_commit_frame.saturating_add(1)..=last_commit_frame {
1817 if self
1818 .wal
1819 .read_frame_header(cx, frame_index)
1820 .await?
1821 .is_commit()
1822 {
1823 committed_txns_after_page = committed_txns_after_page.saturating_add(1);
1824 }
1825 }
1826
1827 Ok(committed_txns_after_page)
1828 })
1829 }
1830
1831 fn conflicting_pages_since_snapshot<'a>(
1832 &'a mut self,
1833 cx: &'a Cx,
1834 snapshot: TransactionConflictSnapshot,
1835 page_numbers: &'a [u32],
1836 _page_baselines: &'a [TransactionConflictPageBaseline],
1837 ) -> WalFuture<'a, Vec<u32>> {
1838 Box::pin(async move {
1839 if page_numbers.is_empty() {
1840 return Ok(Vec::new());
1841 }
1842
1843 let mut candidates = page_numbers
1844 .iter()
1845 .copied()
1846 .filter(|page| *page != 0)
1847 .collect::<Vec<_>>();
1848 candidates.sort_unstable();
1849 candidates.dedup();
1850 if candidates.is_empty() {
1851 return Ok(Vec::new());
1852 }
1853
1854 self.wal.refresh(cx).await?;
1855 self.publish_latest_committed_snapshot(cx, "conflicting_pages_since_snapshot")
1856 .await?;
1857 let latest = self.published_snapshot();
1858
1859 let mut conflicts = HashSet::<u32>::new();
1860
1861 if !(latest.commit_count <= snapshot.commit_count
1880 && latest.generation == snapshot.generation
1881 && latest.last_commit_frame <= snapshot.last_commit_frame)
1882 {
1883 if latest.generation != snapshot.generation {
1884 for &page in &candidates {
1885 conflicts.insert(page);
1886 }
1887 } else if let Some(latest_last_commit_frame) = latest.last_commit_frame {
1888 let start_frame = snapshot
1889 .last_commit_frame
1890 .map_or(0, |frame| frame.saturating_add(1));
1891 if start_frame <= latest_last_commit_frame {
1892 let candidate_set = candidates.iter().copied().collect::<HashSet<_>>();
1893 for frame_index in start_frame..=latest_last_commit_frame {
1894 let header = self.wal.read_frame_header(cx, frame_index).await?;
1895 if candidate_set.contains(&header.page_number) {
1896 conflicts.insert(header.page_number);
1897 }
1898 }
1899 }
1900 }
1901 }
1902
1903 if snapshot.snapshot_db_size > 0
1919 && latest.generation == snapshot.generation
1920 && candidates
1921 .iter()
1922 .any(|page| *page > snapshot.snapshot_db_size)
1923 {
1924 let published = self.published_snapshot.clone();
1925 for &page in &candidates {
1926 if page > snapshot.snapshot_db_size
1927 && !matches!(
1928 self.resolve_visible_frame(cx, &published, page).await?,
1929 WalPageLookupResolution::AuthoritativeMiss
1930 | WalPageLookupResolution::PartialIndexFallbackMiss
1931 )
1932 {
1933 tracing::debug!(
1934 target: "fsqlite.wal.conflict",
1935 page,
1936 allocation_base_db_size = snapshot.snapshot_db_size,
1937 latest_commit_frame = ?latest.last_commit_frame,
1938 "fresh EOF allocation aliases a committed page; failing \
1939 closed with BusySnapshot (bd-o81ov)"
1940 );
1941 conflicts.insert(page);
1942 }
1943 }
1944 }
1945
1946 let mut conflicts = conflicts.into_iter().collect::<Vec<_>>();
1947 conflicts.sort_unstable();
1948 Ok(conflicts)
1949 })
1950 }
1951
1952 fn committed_txn_count<'a>(&'a mut self, cx: &'a Cx) -> WalFuture<'a, u64> {
1953 Box::pin(async move {
1954 let snapshot = if let Some(snapshot) = self.read_snapshot.clone() {
1955 snapshot
1956 } else {
1957 self.publish_latest_committed_snapshot(cx, "committed_txn_count")
1958 .await?;
1959 self.published_snapshot.clone()
1960 };
1961 Ok(snapshot.commit_count)
1962 })
1963 }
1964
1965 fn sync(&mut self, cx: &Cx) -> Result<()> {
1966 self.wal.sync(cx, SyncFlags::NORMAL)?;
1976 self.publish_pending_after_sync(cx)?;
1977 if !self.has_pending_publication() {
1987 self.refresh_before_append = true;
1988 }
1989 Ok(())
1990 }
1991
1992 fn frame_count(&self) -> usize {
1993 self.wal.frame_count()
1994 }
1995
1996 fn checkpoint<'a>(
1997 &'a mut self,
1998 cx: &'a Cx,
1999 mode: CheckpointMode,
2000 writer: &'a mut dyn CheckpointPageWriter,
2001 backfilled_frames: u32,
2002 oldest_reader_frame: Option<u32>,
2003 ) -> WalFuture<'a, CheckpointResult> {
2004 Box::pin(async move {
2005 if self.has_pending_publication() {
2012 return Err(FrankenError::CheckpointFailed {
2013 detail: "staged, unpublished frames remain; a successful commit sync must \
2014 drain them before checkpointing"
2015 .to_owned(),
2016 });
2017 }
2018 self.wal.refresh(cx).await?;
2020 self.refresh_before_append = true;
2021 let total_frames = u32::try_from(self.wal.frame_count()).unwrap_or(u32::MAX);
2022
2023 let state = CheckpointState {
2025 total_frames,
2026 backfilled_frames,
2027 oldest_reader_frame,
2028 };
2029
2030 let mut target = CheckpointTargetAdapterRef { writer };
2032
2033 let result =
2035 execute_checkpoint(cx, &mut self.wal, to_wal_mode(mode), state, &mut target)
2036 .await?;
2037
2038 #[cfg(all(not(target_arch = "wasm32"), feature = "native"))]
2042 if result.frames_backfilled > 0 {
2043 let drained = self.fec_pending.len();
2044 self.fec_pending.clear();
2045 if drained > 0 {
2046 debug!(
2047 drained_groups = drained,
2048 frames_backfilled = result.frames_backfilled,
2049 "FEC symbols reclaimed after checkpoint"
2050 );
2051 }
2052 }
2053
2054 #[cfg(all(not(target_arch = "wasm32"), feature = "native"))]
2057 if result.wal_was_reset {
2058 self.fec_discard();
2059 }
2060 if result.wal_was_reset {
2061 self.invalidate_publication();
2062 }
2063
2064 self.publish_latest_committed_snapshot(cx, "checkpoint")
2065 .await?;
2066
2067 Ok(CheckpointResult {
2068 total_frames,
2069 frames_backfilled: result.frames_backfilled,
2070 completed: result.plan.completes_checkpoint(),
2071 wal_was_reset: result.wal_was_reset,
2072 requested_mode: mode,
2073 effective_mode: mode,
2074 })
2075 })
2076 }
2077}
2078
2079const MIN_DURABLE_CERTIFICATE_RECORD_SIZE: usize =
2080 ParallelWalDurableCertificateRecord::MIN_ENCODED_SIZE;
2081const DURABLE_CERTIFICATE_RECORD_HEADER_SIZE: usize = 14;
2082const MAX_ORPHAN_CERTIFICATE_LOOKBACK: usize = 64;
2083
2084fn durable_certificate_declared_len(bytes: &[u8]) -> Option<usize> {
2085 let length_bytes = bytes.get(10..DURABLE_CERTIFICATE_RECORD_HEADER_SIZE)?;
2086 usize::try_from(u32::from_le_bytes([
2087 length_bytes[0],
2088 length_bytes[1],
2089 length_bytes[2],
2090 length_bytes[3],
2091 ]))
2092 .ok()
2093}
2094
2095fn durable_certificate_declares_len(bytes: &[u8], expected: usize) -> bool {
2096 durable_certificate_declared_len(bytes).is_some_and(|actual| actual.cmp(&expected).is_eq())
2097}
2098
2099fn decode_durable_certificate_record(
2100 bytes: &[u8],
2101 location: &str,
2102) -> Result<ParallelWalDurableCertificateRecord> {
2103 ParallelWalDurableCertificateRecord::from_bytes(bytes).map_err(|error| {
2104 FrankenError::WalCorrupt {
2105 detail: format!("parallel WAL certificate {location} is invalid: {error}"),
2106 }
2107 })
2108}
2109
2110fn validate_incomplete_certificate_suffix(bytes: &[u8], anchored: bool) -> Result<()> {
2111 if bytes.is_empty() {
2112 return Ok(());
2113 }
2114 if bytes.len() > PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE {
2115 return Err(FrankenError::WalCorrupt {
2116 detail: format!(
2117 "parallel WAL certificate torn suffix exceeds {} bytes",
2118 PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE
2119 ),
2120 });
2121 }
2122
2123 if bytes.len() < PARALLEL_WAL_DURABLE_CERTIFICATE_MAGIC.len() {
2124 if anchored || PARALLEL_WAL_DURABLE_CERTIFICATE_MAGIC.starts_with(bytes) {
2129 return Ok(());
2130 }
2131 return Err(FrankenError::WalCorrupt {
2132 detail: "parallel WAL certificate sidecar starts with non-record garbage".to_owned(),
2133 });
2134 }
2135 if !bytes.starts_with(&PARALLEL_WAL_DURABLE_CERTIFICATE_MAGIC) {
2136 return Err(FrankenError::WalCorrupt {
2137 detail: "parallel WAL certificate suffix does not start at a record boundary"
2138 .to_owned(),
2139 });
2140 }
2141 if bytes.len() < 10 {
2142 return Ok(());
2143 }
2144 let version = u16::from_le_bytes([bytes[8], bytes[9]]);
2145 if version != fsqlite_wal::PARALLEL_WAL_DURABLE_CERTIFICATE_RECORD_VERSION {
2146 return Err(FrankenError::WalCorrupt {
2147 detail: format!(
2148 "parallel WAL certificate suffix has unsupported record version {version}"
2149 ),
2150 });
2151 }
2152 if bytes.len() < DURABLE_CERTIFICATE_RECORD_HEADER_SIZE {
2153 return Ok(());
2154 }
2155 let declared_len =
2156 durable_certificate_declared_len(bytes).ok_or_else(|| FrankenError::WalCorrupt {
2157 detail: "parallel WAL certificate suffix length exceeds usize".to_owned(),
2158 })?;
2159 if !(MIN_DURABLE_CERTIFICATE_RECORD_SIZE..=PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE)
2160 .contains(&declared_len)
2161 {
2162 return Err(FrankenError::WalCorrupt {
2163 detail: format!(
2164 "parallel WAL certificate suffix declares invalid record length {declared_len}"
2165 ),
2166 });
2167 }
2168 if bytes.len() < declared_len {
2169 return Ok(());
2170 }
2171
2172 decode_durable_certificate_record(&bytes[..declared_len], "suffix")?;
2177 Err(FrankenError::WalCorrupt {
2178 detail:
2179 "parallel WAL certificate sidecar contains a complete record outside the footer chain"
2180 .to_owned(),
2181 })
2182}
2183
2184fn combine_sidecar_io_results<const N: usize>(
2185 context: &str,
2186 results: [(&str, Result<()>); N],
2187) -> Result<()> {
2188 let failures = results
2189 .into_iter()
2190 .filter_map(|(stage, result)| result.err().map(|error| (stage, error)))
2191 .collect::<Vec<_>>();
2192 if failures.is_empty() {
2193 return Ok(());
2194 }
2195 if failures.len() == 1 {
2196 return failures
2197 .into_iter()
2198 .next()
2199 .map_or(Ok(()), |(_, error)| Err(error));
2200 }
2201 let details = failures
2202 .iter()
2203 .map(|(stage, error)| format!("{stage}={error}"))
2204 .collect::<Vec<_>>()
2205 .join("; ");
2206 Err(FrankenError::internal(format!("{context}: {details}")))
2207}
2208
2209pub struct PathRefreshingWalBackend<V: Vfs>
2219where
2220 V::File: Send + Sync + 'static,
2221{
2222 vfs: V,
2223 db_path: PathBuf,
2224 wal_path: PathBuf,
2225 page_size: u32,
2226 create_missing: bool,
2227 #[cfg(all(feature = "native", any(unix, windows)))]
2228 namespace_binding: Option<Arc<DatabaseNamespaceBinding>>,
2229 cached_verification_db: Option<V::File>,
2248 cached_certificate_read: std::sync::Mutex<Option<V::File>>,
2265 inner: WalBackendAdapter<V::File>,
2266}
2267
2268impl<V> PathRefreshingWalBackend<V>
2269where
2270 V: Vfs + 'static,
2271 V::File: Send + Sync + 'static,
2272{
2273 #[must_use]
2274 pub fn new(
2275 vfs: V,
2276 db_path: impl AsRef<Path>,
2277 wal_path: impl AsRef<Path>,
2278 page_size: u32,
2279 wal: WalFile<V::File>,
2280 create_missing: bool,
2281 #[cfg(all(feature = "native", any(unix, windows)))] namespace_binding: Option<
2282 Arc<DatabaseNamespaceBinding>,
2283 >,
2284 ) -> Self {
2285 Self {
2286 vfs,
2287 db_path: db_path.as_ref().to_path_buf(),
2288 wal_path: wal_path.as_ref().to_path_buf(),
2289 page_size,
2290 create_missing,
2291 #[cfg(all(feature = "native", any(unix, windows)))]
2292 namespace_binding,
2293 cached_verification_db: None,
2294 cached_certificate_read: std::sync::Mutex::new(None),
2295 inner: WalBackendAdapter::new(wal),
2296 }
2297 }
2298
2299 #[must_use]
2300 pub fn into_inner(self) -> WalBackendAdapter<V::File> {
2301 self.inner
2302 }
2303
2304 fn replace_inner(&mut self, cx: &Cx, wal: WalFile<V::File>) -> Result<()> {
2312 if self.inner.has_pending_publication() {
2313 let cleanup_cx = cx.create_child();
2314 let _cleanup_mask = cleanup_cx.masked();
2315 let _ = wal.close(&cleanup_cx);
2316 return Err(FrankenError::Busy);
2317 }
2318 let old = std::mem::replace(&mut self.inner, WalBackendAdapter::new(wal));
2319 if let Some(mut stale) = self
2323 .cached_certificate_read
2324 .get_mut()
2325 .unwrap_or_else(std::sync::PoisonError::into_inner)
2326 .take()
2327 {
2328 let _ = stale.close(cx);
2329 }
2330 let old_wal = old.into_inner()?;
2331 let _ = old_wal.close(cx);
2332 Ok(())
2333 }
2334
2335 async fn create_replacement_wal(&self, cx: &Cx) -> Result<WalFile<V::File>> {
2336 let flags = VfsOpenFlags::READWRITE | VfsOpenFlags::CREATE | VfsOpenFlags::WAL;
2337 let (file, _) = self.vfs.open(cx, Some(&self.wal_path), flags)?;
2338 let wal = WalFile::create(cx, file, self.page_size, 0, WalSalts::generate()).await?;
2341 if let Err(error) = self.vfs.sync_parent_directory(cx, &self.wal_path) {
2342 let cleanup_cx = cx.create_child();
2343 let _cleanup_mask = cleanup_cx.masked();
2344 let _ = wal.close(&cleanup_cx);
2345 return Err(error);
2346 }
2347 Ok(wal)
2348 }
2349
2350 async fn replace_with_created_wal(&mut self, cx: &Cx) -> Result<()> {
2351 let wal = self.create_replacement_wal(cx).await?;
2352 self.replace_inner(cx, wal)
2353 }
2354
2355 async fn open_replacement_wal(&self, cx: &Cx, path_file: V::File) -> Result<WalFile<V::File>> {
2356 let wal = WalFile::open(cx, path_file).await?;
2357 if u32::try_from(wal.page_size()).ok() != Some(self.page_size) {
2358 let actual_page_size = wal.page_size();
2359 let expected_page_size = self.page_size;
2360 let _ = wal.close(cx);
2361 return Err(FrankenError::WalCorrupt {
2362 detail: format!(
2363 "WAL page size {actual_page_size} does not match database page size {expected_page_size} during path refresh"
2364 ),
2365 });
2366 }
2367 Ok(wal)
2368 }
2369
2370 async fn path_header_matches_current_handle(
2371 &self,
2372 cx: &Cx,
2373 path_file: &V::File,
2374 ) -> Result<bool> {
2375 let mut header_buf = [0_u8; WAL_HEADER_SIZE];
2376 let bytes_read = path_file.read(cx, &mut header_buf, 0).await?;
2377 if bytes_read < WAL_HEADER_SIZE {
2378 return Ok(false);
2379 }
2380
2381 let path_header = WalHeader::from_bytes(&header_buf)?;
2382 if !validate_wal_header_checksum(&header_buf, path_header.big_endian_checksum())? {
2383 return Err(FrankenError::WalCorrupt {
2384 detail: "WAL header checksum mismatch during path refresh".to_owned(),
2385 });
2386 }
2387
2388 let current_header = self.inner.inner().header();
2389 Ok(path_header.magic == current_header.magic
2390 && path_header.format_version == current_header.format_version
2391 && path_header.page_size == current_header.page_size
2392 && path_header.checkpoint_seq == current_header.checkpoint_seq
2393 && path_header.salts == current_header.salts)
2394 }
2395
2396 async fn conflicts_after_generation_change(
2412 &mut self,
2413 cx: &Cx,
2414 page_numbers: &[u32],
2415 page_baselines: &[TransactionConflictPageBaseline],
2416 ) -> Vec<u32> {
2417 let mut candidates = page_numbers
2418 .iter()
2419 .copied()
2420 .filter(|page| *page != 0)
2421 .collect::<Vec<_>>();
2422 candidates.sort_unstable();
2423 candidates.dedup();
2424 if candidates.is_empty() {
2425 return Vec::new();
2426 }
2427
2428 let mut baselines = HashMap::<u32, [u8; 32]>::new();
2429 let mut ambiguous_baselines = HashSet::<u32>::new();
2430 for baseline in page_baselines {
2431 if baseline.page_number == 0 {
2432 continue;
2433 }
2434 if let Some(previous) = baselines.insert(baseline.page_number, baseline.page_hash)
2435 && previous != baseline.page_hash
2436 {
2437 ambiguous_baselines.insert(baseline.page_number);
2438 }
2439 }
2440
2441 let mut db_file = match self.cached_verification_db.take() {
2448 Some(cached) => cached,
2449 None => {
2450 let main_db_flags = VfsOpenFlags::READWRITE | VfsOpenFlags::MAIN_DB;
2451 match self.vfs.open(cx, Some(&self.db_path), main_db_flags) {
2452 Ok((file, _)) => file,
2453 Err(_) => return candidates,
2454 }
2455 }
2456 };
2457 let page_size = match usize::try_from(self.page_size) {
2458 Ok(page_size) if page_size > 0 => page_size,
2459 _ => {
2460 let _ = db_file.close(cx);
2461 return candidates;
2462 }
2463 };
2464
2465 let mut page_one = vec![0_u8; page_size];
2468 let page_one_read = match db_file.read(cx, &mut page_one, 0).await {
2469 Ok(bytes_read) => bytes_read,
2470 Err(_) => {
2471 let _ = db_file.close(cx);
2472 return candidates;
2473 }
2474 };
2475 let header_page_size =
2476 (page_one_read == page_size).then(|| sqlite_database_header_page_size(&page_one));
2477 if header_page_size.flatten() != Some(self.page_size) {
2478 let _ = db_file.close(cx);
2479 return candidates;
2480 }
2481 let committed_page_count =
2492 u32::from_be_bytes([page_one[28], page_one[29], page_one[30], page_one[31]]);
2493
2494 let mut conflicts = Vec::new();
2495 for &page_number in &candidates {
2496 let Some(expected_hash) = baselines.get(&page_number).copied() else {
2497 if committed_page_count > 0 && page_number > committed_page_count {
2498 continue;
2499 }
2500 conflicts.push(page_number);
2501 continue;
2502 };
2503 if ambiguous_baselines.contains(&page_number) {
2504 conflicts.push(page_number);
2505 continue;
2506 }
2507
2508 let current_page = match self.inner.read_page(cx, page_number).await {
2509 Ok(Some(page)) if page.len() == page_size => page,
2510 Ok(Some(_)) | Err(_) => {
2511 conflicts.push(page_number);
2512 continue;
2513 }
2514 Ok(None) => {
2515 let mut page = vec![0_u8; page_size];
2516 let page_offset = u64::from(page_number.saturating_sub(1))
2517 .saturating_mul(u64::from(self.page_size));
2518 match db_file.read(cx, &mut page, page_offset).await {
2519 Ok(bytes_read) if bytes_read == page_size => page,
2520 Ok(_) | Err(_) => {
2521 conflicts.push(page_number);
2522 continue;
2523 }
2524 }
2525 }
2526 };
2527 let current_hash = *blake3::hash(¤t_page).as_bytes();
2528 if current_hash != expected_hash {
2529 conflicts.push(page_number);
2530 }
2531 }
2532
2533 self.cached_verification_db = Some(db_file);
2538 conflicts.sort_unstable();
2539 conflicts.dedup();
2540 conflicts
2541 }
2542
2543 async fn ensure_current_wal_path(&mut self, cx: &Cx) -> Result<()> {
2544 #[cfg(all(feature = "native", any(unix, windows)))]
2545 if let Some(binding) = &self.namespace_binding {
2546 binding.validate_path_identity()?;
2547 }
2548 if !self.vfs.access(cx, &self.wal_path, AccessFlags::EXISTS)? {
2549 if self.create_missing {
2550 return self.replace_with_created_wal(cx).await;
2551 }
2552 return Ok(());
2553 }
2554
2555 let flags = VfsOpenFlags::READWRITE | VfsOpenFlags::WAL;
2556 let (mut path_file, _) = self.vfs.open(cx, Some(&self.wal_path), flags)?;
2557 let path_size = path_file.file_size(cx)?;
2558 if path_size < u64::try_from(WAL_HEADER_SIZE).unwrap_or(32) {
2559 let _ = path_file.close(cx);
2560 if self.create_missing {
2561 return self.replace_with_created_wal(cx).await;
2562 }
2563 return Ok(());
2564 }
2565
2566 let current_size = self.inner.inner().file().file_size(cx).unwrap_or(u64::MAX);
2567 let path_matches_current = if path_size == current_size {
2568 match self
2569 .path_header_matches_current_handle(cx, &path_file)
2570 .await
2571 {
2572 Ok(matches) => matches,
2573 Err(err) => {
2574 let _ = path_file.close(cx);
2575 return Err(err);
2576 }
2577 }
2578 } else {
2579 false
2580 };
2581 if !path_matches_current {
2582 let wal = self.open_replacement_wal(cx, path_file).await?;
2583 self.replace_inner(cx, wal)?;
2584 } else {
2585 let _ = path_file.close(cx);
2586 }
2587 Ok(())
2588 }
2589
2590 fn certificate_sidecar_path(&self) -> PathBuf {
2591 let mut path = self.wal_path.as_os_str().to_owned();
2592 path.push("-cert");
2593 PathBuf::from(path)
2594 }
2595
2596 fn certificate_checkpoint_handoff_path(&self) -> PathBuf {
2597 let mut path = self.wal_path.as_os_str().to_owned();
2598 path.push("-cert-head");
2599 PathBuf::from(path)
2600 }
2601
2602 async fn read_certificate_sidecar_exact(
2603 file: &V::File,
2604 cx: &Cx,
2605 offset: u64,
2606 len: usize,
2607 location: &str,
2608 ) -> Result<Vec<u8>> {
2609 let mut bytes = vec![0_u8; len];
2610 let bytes_read = file.read(cx, &mut bytes, offset).await?;
2611 if bytes_read != len {
2612 return Err(FrankenError::WalCorrupt {
2613 detail: format!(
2614 "parallel WAL certificate {location} at offset {offset} was short-read: got {bytes_read} of {len}"
2615 ),
2616 });
2617 }
2618 Ok(bytes)
2619 }
2620
2621 async fn read_certificate_record_ending_at(
2622 file: &V::File,
2623 cx: &Cx,
2624 record_end: u64,
2625 ) -> Result<(u64, ParallelWalDurableCertificateRecord)> {
2626 let footer_size =
2627 u64::try_from(ParallelWalDurableCertificateRecord::LENGTH_FOOTER_SIZE).unwrap_or(4);
2628 let footer_offset = record_end.checked_sub(footer_size).ok_or_else(|| {
2629 FrankenError::WalCorrupt {
2630 detail: format!(
2631 "parallel WAL certificate record ending at {record_end} has no length footer"
2632 ),
2633 }
2634 })?;
2635 let footer = Self::read_certificate_sidecar_exact(
2636 file,
2637 cx,
2638 footer_offset,
2639 ParallelWalDurableCertificateRecord::LENGTH_FOOTER_SIZE,
2640 "length footer",
2641 )
2642 .await?;
2643 let record_len = usize::try_from(u32::from_le_bytes([
2644 footer[0], footer[1], footer[2], footer[3],
2645 ]))
2646 .map_err(|_| FrankenError::WalCorrupt {
2647 detail: "parallel WAL certificate footer length exceeds usize".to_owned(),
2648 })?;
2649 if !(MIN_DURABLE_CERTIFICATE_RECORD_SIZE..=PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE)
2650 .contains(&record_len)
2651 {
2652 return Err(FrankenError::WalCorrupt {
2653 detail: format!(
2654 "parallel WAL certificate footer declares invalid record length {record_len}"
2655 ),
2656 });
2657 }
2658 let record_len_u64 = u64::try_from(record_len).map_err(|_| FrankenError::WalCorrupt {
2659 detail: "parallel WAL certificate record length exceeds u64".to_owned(),
2660 })?;
2661 let record_start =
2662 record_end
2663 .checked_sub(record_len_u64)
2664 .ok_or_else(|| FrankenError::WalCorrupt {
2665 detail: format!(
2666 "parallel WAL certificate record length {record_len} exceeds end offset {record_end}"
2667 ),
2668 })?;
2669 let bytes =
2670 Self::read_certificate_sidecar_exact(file, cx, record_start, record_len, "record")
2671 .await?;
2672 let record = decode_durable_certificate_record(&bytes, "record")?;
2673 Ok((record_start, record))
2674 }
2675
2676 async fn prepare_certificate_sidecar_for_append(file: &mut V::File, cx: &Cx) -> Result<u64> {
2683 let file_size = file.file_size(cx)?;
2684 if file_size == 0 {
2685 return Ok(0);
2686 }
2687
2688 let footer_size =
2689 u64::try_from(ParallelWalDurableCertificateRecord::LENGTH_FOOTER_SIZE).unwrap_or(4);
2690 if file_size >= footer_size {
2691 let footer = Self::read_certificate_sidecar_exact(
2692 file,
2693 cx,
2694 file_size - footer_size,
2695 ParallelWalDurableCertificateRecord::LENGTH_FOOTER_SIZE,
2696 "append-boundary length footer",
2697 )
2698 .await?;
2699 let record_len = usize::try_from(u32::from_le_bytes([
2700 footer[0], footer[1], footer[2], footer[3],
2701 ]))
2702 .unwrap_or(usize::MAX);
2703 let record_len_u64 = u64::try_from(record_len).unwrap_or(u64::MAX);
2704 if (MIN_DURABLE_CERTIFICATE_RECORD_SIZE
2705 ..=PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE)
2706 .contains(&record_len)
2707 && record_len_u64 <= file_size
2708 {
2709 let record_start = file_size - record_len_u64;
2710 let bytes = Self::read_certificate_sidecar_exact(
2711 file,
2712 cx,
2713 record_start,
2714 record_len,
2715 "append-boundary record",
2716 )
2717 .await?;
2718 if bytes.starts_with(&PARALLEL_WAL_DURABLE_CERTIFICATE_MAGIC)
2719 || durable_certificate_declares_len(&bytes, record_len)
2720 {
2721 decode_durable_certificate_record(&bytes, "append-boundary record")?;
2722 return Ok(file_size);
2723 }
2724 }
2725 }
2726
2727 let recovery_window_size =
2731 PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE.saturating_mul(2);
2732 let recovery_window_size_u64 = u64::try_from(recovery_window_size).unwrap_or(u64::MAX);
2733 let tail_offset = file_size.saturating_sub(recovery_window_size_u64);
2734 let tail_len =
2735 usize::try_from(file_size - tail_offset).map_err(|_| FrankenError::WalCorrupt {
2736 detail: "parallel WAL certificate append-repair window exceeds usize".to_owned(),
2737 })?;
2738 let tail = Self::read_certificate_sidecar_exact(
2739 file,
2740 cx,
2741 tail_offset,
2742 tail_len,
2743 "append-repair window",
2744 )
2745 .await?;
2746 let minimum_candidate_end = tail
2747 .len()
2748 .saturating_sub(PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE)
2749 .max(ParallelWalDurableCertificateRecord::LENGTH_FOOTER_SIZE);
2750 let mut anchor_end = None;
2751 for candidate_end in (minimum_candidate_end..tail.len()).rev() {
2752 let footer_start =
2753 candidate_end - ParallelWalDurableCertificateRecord::LENGTH_FOOTER_SIZE;
2754 let footer = &tail[footer_start..candidate_end];
2755 let record_len = usize::try_from(u32::from_le_bytes([
2756 footer[0], footer[1], footer[2], footer[3],
2757 ]))
2758 .unwrap_or(usize::MAX);
2759 if !(MIN_DURABLE_CERTIFICATE_RECORD_SIZE
2760 ..=PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE)
2761 .contains(&record_len)
2762 || record_len > candidate_end
2763 {
2764 continue;
2765 }
2766 let record_start = candidate_end - record_len;
2767 let record_bytes = &tail[record_start..candidate_end];
2768 if !record_bytes.starts_with(&PARALLEL_WAL_DURABLE_CERTIFICATE_MAGIC)
2769 || !durable_certificate_declares_len(record_bytes, record_len)
2770 {
2771 continue;
2772 }
2773 if ParallelWalDurableCertificateRecord::from_bytes(record_bytes).is_ok() {
2774 anchor_end = Some(candidate_end);
2775 break;
2776 }
2777 }
2778
2779 let safe_end = if let Some(anchor_end) = anchor_end {
2780 validate_incomplete_certificate_suffix(&tail[anchor_end..], true)?;
2781 tail_offset
2782 .checked_add(u64::try_from(anchor_end).unwrap_or(u64::MAX))
2783 .ok_or_else(|| FrankenError::WalCorrupt {
2784 detail: "parallel WAL certificate append-repair boundary overflow".to_owned(),
2785 })?
2786 } else {
2787 if file_size
2788 > u64::try_from(PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE)
2789 .unwrap_or(u64::MAX)
2790 {
2791 return Err(FrankenError::WalCorrupt {
2792 detail: format!(
2793 "parallel WAL certificate sidecar has no valid append boundary within its bounded {}-byte recovery suffix",
2794 PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE
2795 ),
2796 });
2797 }
2798 validate_incomplete_certificate_suffix(&tail, false)?;
2799 0
2800 };
2801
2802 if safe_end < file_size {
2803 file.truncate(cx, safe_end)?;
2804 }
2805 Ok(safe_end)
2806 }
2807
2808 async fn append_durable_certificate_record(
2809 &self,
2810 cx: &Cx,
2811 certificate: &ParallelWalCommitCertificate,
2812 wal_frame_start: u64,
2813 wal_frame_end: u64,
2814 sync: bool,
2815 ) -> Result<()> {
2816 self.append_durable_certificate_record_with_completion(
2817 cx,
2818 certificate,
2819 wal_frame_start,
2820 wal_frame_end,
2821 sync,
2822 None,
2823 )
2824 .await
2825 }
2826
2827 async fn append_durable_certificate_record_with_completion(
2828 &self,
2829 cx: &Cx,
2830 certificate: &ParallelWalCommitCertificate,
2831 wal_frame_start: u64,
2832 wal_frame_end: u64,
2833 sync: bool,
2834 completion: Option<&VfsWriteCompletion>,
2835 ) -> Result<()> {
2836 let mut preflight = WalWriteCompletionPreflight::new(completion);
2837 let expected_frame_start = u64::try_from(self.inner.frame_count())
2838 .unwrap_or(u64::MAX)
2839 .saturating_add(1);
2840 if wal_frame_start != expected_frame_start {
2841 return Err(FrankenError::internal(format!(
2842 "parallel WAL certificate starts at frame {wal_frame_start}, expected {expected_frame_start}"
2843 )));
2844 }
2845 let record = ParallelWalDurableCertificateRecord::new(
2846 self.inner.inner().generation_identity(),
2847 wal_frame_start,
2848 wal_frame_end,
2849 certificate.clone(),
2850 )
2851 .map_err(|error| {
2852 FrankenError::internal(format!(
2853 "could not encode parallel WAL durability certificate: {error}"
2854 ))
2855 })?;
2856 let record_bytes = record.to_bytes();
2857 if record_bytes.len() > PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE {
2858 return Err(FrankenError::WalCorrupt {
2859 detail: format!(
2860 "parallel WAL certificate record is {} bytes; maximum is {}",
2861 record_bytes.len(),
2862 PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE
2863 ),
2864 });
2865 }
2866 let certificate_path = self.certificate_sidecar_path();
2867 let existed = self
2868 .vfs
2869 .access(cx, &certificate_path, AccessFlags::EXISTS)?;
2870 let flags = VfsOpenFlags::READWRITE | VfsOpenFlags::CREATE | VfsOpenFlags::WAL;
2871 let (mut file, _) = self.vfs.open(cx, Some(&certificate_path), flags)?;
2872 let append_offset = Self::prepare_certificate_sidecar_for_append(&mut file, cx).await?;
2873 preflight.hand_off();
2874 drop(preflight);
2875 let write_result = if let Some(completion) = completion {
2876 file.write_tracked(cx, &record_bytes, append_offset, completion.clone())
2877 .await
2878 } else {
2879 file.write(cx, &record_bytes, append_offset).await
2880 };
2881 if let Err(write_error) = write_result {
2882 let cleanup_cx = cx.create_child();
2887 let _cleanup_mask = cleanup_cx.masked();
2888 let cleanup_result = file.truncate(&cleanup_cx, append_offset);
2889 let close_result = file.close(&cleanup_cx);
2890 return combine_sidecar_io_results(
2891 "parallel WAL certificate append cleanup failed",
2892 [
2893 ("write", Err(write_error)),
2894 ("truncate", cleanup_result),
2895 ("close", close_result),
2896 ],
2897 );
2898 }
2899
2900 let finalization_cx = cx.create_child();
2904 let _finalization_mask = finalization_cx.masked();
2905 let sync_result = if sync {
2906 file.durable_sync(&finalization_cx, SyncKind::FullDurable)
2907 } else {
2908 Ok(())
2909 };
2910 let directory_sync_result = if sync && !existed && sync_result.is_ok() {
2911 self.vfs
2912 .sync_parent_directory(&finalization_cx, &certificate_path)
2913 } else {
2914 Ok(())
2915 };
2916 let close_result = file.close(&finalization_cx);
2917 combine_sidecar_io_results(
2918 "parallel WAL certificate append finalization failed",
2919 [
2920 ("file_sync", sync_result),
2921 ("directory_sync", directory_sync_result),
2922 ("close", close_result),
2923 ],
2924 )
2925 }
2926
2927 async fn reconcile_certificate_sidecar_record(
2928 &self,
2929 cx: &Cx,
2930 expected: &ParallelWalDurableCertificateRecord,
2931 remove_expected_orphan: bool,
2932 sync: bool,
2933 ) -> Result<bool> {
2934 let certificate_path = self.certificate_sidecar_path();
2935 if !self
2936 .vfs
2937 .access(cx, &certificate_path, AccessFlags::EXISTS)?
2938 {
2939 return Ok(false);
2940 }
2941
2942 let flags = VfsOpenFlags::READWRITE | VfsOpenFlags::WAL;
2943 let (mut file, _) = self.vfs.open(cx, Some(&certificate_path), flags)?;
2944 let reconciliation_result = async {
2945 let original_size = file.file_size(cx)?;
2946 let safe_end = Self::prepare_certificate_sidecar_for_append(&mut file, cx).await?;
2947 let latest = if safe_end == 0 {
2948 None
2949 } else {
2950 Some(Self::read_certificate_record_ending_at(&file, cx, safe_end).await?)
2951 };
2952 let latest_is_expected = latest
2953 .as_ref()
2954 .is_some_and(|(_, record)| record == expected);
2955 let sidecar_changed = if remove_expected_orphan
2956 && let Some((record_start, _)) = latest.as_ref()
2957 && latest_is_expected
2958 {
2959 file.truncate(cx, *record_start)?;
2960 true
2961 } else {
2962 safe_end != original_size
2963 };
2964 if sync && (latest_is_expected || sidecar_changed) {
2965 file.durable_sync(cx, SyncKind::FullDurable)?;
2966 }
2967 if sync && latest_is_expected && !remove_expected_orphan {
2968 self.vfs.sync_parent_directory(cx, &certificate_path)?;
2971 }
2972 Ok(latest_is_expected)
2973 }
2974 .await;
2975
2976 let cleanup_cx = cx.create_child();
2977 let _cleanup_mask = cleanup_cx.masked();
2978 let close_result = file.close(&cleanup_cx);
2979 match (reconciliation_result, close_result) {
2980 (Ok(latest_is_expected), Ok(())) => Ok(latest_is_expected),
2981 (Err(error), Ok(())) | (Ok(_), Err(error)) => Err(error),
2982 (Err(reconciliation_error), Err(close_error)) => Err(FrankenError::internal(format!(
2983 "parallel WAL certificate reconciliation failed and close also failed: reconciliation={reconciliation_error}; close={close_error}"
2984 ))),
2985 }
2986 }
2987
2988 async fn persist_checkpoint_certificate_handoff(
2989 &self,
2990 cx: &Cx,
2991 record: &ParallelWalDurableCertificateRecord,
2992 ) -> Result<()> {
2993 let handoff_path = self.certificate_checkpoint_handoff_path();
2994 let existed = self.vfs.access(cx, &handoff_path, AccessFlags::EXISTS)?;
2995 let flags = VfsOpenFlags::READWRITE | VfsOpenFlags::CREATE | VfsOpenFlags::WAL;
2996 let (mut file, _) = self.vfs.open(cx, Some(&handoff_path), flags)?;
2997 let record_bytes = record.to_bytes();
2998 if record_bytes.len() > PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE {
2999 let cleanup_cx = cx.create_child();
3000 let _cleanup_mask = cleanup_cx.masked();
3001 let close_result = file.close(&cleanup_cx);
3002 return combine_sidecar_io_results(
3003 "parallel WAL checkpoint certificate handoff is oversized",
3004 [
3005 (
3006 "record_size",
3007 Err(FrankenError::WalCorrupt {
3008 detail: format!(
3009 "parallel WAL checkpoint certificate handoff is {} bytes; maximum is {}",
3010 record_bytes.len(),
3011 PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE
3012 ),
3013 }),
3014 ),
3015 ("close", close_result),
3016 ],
3017 );
3018 }
3019 if existed {
3024 let file_size = file.file_size(cx)?;
3025 if file_size == u64::try_from(record_bytes.len()).unwrap_or(u64::MAX) {
3026 let mut current = vec![0_u8; record_bytes.len()];
3027 let unchanged = file
3028 .read(cx, &mut current, 0)
3029 .await
3030 .is_ok_and(|bytes_read| {
3031 bytes_read == record_bytes.len() && current == record_bytes
3032 });
3033 if unchanged {
3034 let cleanup_cx = cx.create_child();
3035 let _cleanup_mask = cleanup_cx.masked();
3036 return file.close(&cleanup_cx);
3037 }
3038 }
3039 }
3040 let mutation_cx = cx.create_child();
3045 let _mutation_mask = mutation_cx.masked();
3046 let truncate_result = file.truncate(&mutation_cx, 0);
3047 let write_result = if truncate_result.is_ok() {
3048 file.write(&mutation_cx, &record_bytes, 0).await
3049 } else {
3050 Ok(())
3051 };
3052 if let Err(write_error) = write_result {
3053 let cleanup_result = file.truncate(&mutation_cx, 0);
3054 let close_result = file.close(&mutation_cx);
3055 return combine_sidecar_io_results(
3056 "parallel WAL checkpoint certificate handoff cleanup failed",
3057 [
3058 ("truncate_before_write", truncate_result),
3059 ("write", Err(write_error)),
3060 ("truncate_after_write", cleanup_result),
3061 ("close", close_result),
3062 ],
3063 );
3064 }
3065 let sync_result = if truncate_result.is_ok() {
3066 file.durable_sync(&mutation_cx, SyncKind::FullDurable)
3067 } else {
3068 Ok(())
3069 };
3070 let directory_sync_result = if !existed && truncate_result.is_ok() && sync_result.is_ok() {
3071 self.vfs.sync_parent_directory(&mutation_cx, &handoff_path)
3072 } else {
3073 Ok(())
3074 };
3075 let close_result = file.close(&mutation_cx);
3076 combine_sidecar_io_results(
3077 "parallel WAL checkpoint certificate handoff finalization failed",
3078 [
3079 ("truncate", truncate_result),
3080 ("file_sync", sync_result),
3081 ("directory_sync", directory_sync_result),
3082 ("close", close_result),
3083 ],
3084 )
3085 }
3086
3087 async fn checkpoint_certificate_handoff(
3088 &self,
3089 cx: &Cx,
3090 ) -> Result<Option<ParallelWalCommitCertificate>> {
3091 let handoff_path = self.certificate_checkpoint_handoff_path();
3092 if !self.vfs.access(cx, &handoff_path, AccessFlags::EXISTS)? {
3093 return Ok(None);
3094 }
3095 let flags = VfsOpenFlags::READONLY | VfsOpenFlags::WAL;
3096 let (mut file, _) = self.vfs.open(cx, Some(&handoff_path), flags)?;
3097 let read_result = async {
3098 let file_size =
3099 usize::try_from(file.file_size(cx)?).map_err(|_| FrankenError::WalCorrupt {
3100 detail: "parallel WAL checkpoint certificate handoff exceeds usize".to_owned(),
3101 })?;
3102 if file_size == 0 || file_size > PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE {
3103 return Err(FrankenError::WalCorrupt {
3104 detail: format!(
3105 "parallel WAL checkpoint certificate handoff has invalid size {file_size}"
3106 ),
3107 });
3108 }
3109 let mut bytes = vec![0_u8; file_size];
3110 let bytes_read = file.read(cx, &mut bytes, 0).await?;
3111 if bytes_read != bytes.len() {
3112 return Err(FrankenError::WalCorrupt {
3113 detail: "parallel WAL checkpoint certificate handoff was short-read".to_owned(),
3114 });
3115 }
3116 let record =
3117 ParallelWalDurableCertificateRecord::from_bytes(&bytes).map_err(|error| {
3118 FrankenError::WalCorrupt {
3119 detail: format!(
3120 "parallel WAL checkpoint certificate handoff is invalid: {error}"
3121 ),
3122 }
3123 })?;
3124 Ok(Some(record.certificate))
3125 }
3126 .await;
3127 let cleanup_cx = cx.create_child();
3128 let _cleanup_mask = cleanup_cx.masked();
3129 let close_result = file.close(&cleanup_cx);
3130 match (read_result, close_result) {
3131 (Ok(certificate), Ok(())) => Ok(certificate),
3132 (Err(error), Ok(())) | (Ok(_), Err(error)) => Err(error),
3133 (Err(read_error), Err(close_error)) => Err(FrankenError::internal(format!(
3134 "parallel WAL checkpoint handoff read failed and close also failed: read={read_error}; close={close_error}"
3135 ))),
3136 }
3137 }
3138
3139 async fn wal_frame_payload_digest(
3140 &self,
3141 cx: &Cx,
3142 wal_frame_start: u64,
3143 wal_frame_end: u64,
3144 ) -> Result<[u8; 32]> {
3145 if wal_frame_start == 0 || wal_frame_end < wal_frame_start {
3146 return Err(FrankenError::WalCorrupt {
3147 detail: format!(
3148 "invalid parallel WAL digest interval {wal_frame_start}..={wal_frame_end}"
3149 ),
3150 });
3151 }
3152
3153 let mut digest = ParallelWalFramePayloadDigestBuilder::new();
3154 for frame_number in wal_frame_start..=wal_frame_end {
3155 let frame_index = usize::try_from(frame_number.saturating_sub(1)).map_err(|_| {
3156 FrankenError::WalCorrupt {
3157 detail: format!(
3158 "parallel WAL digest frame number {frame_number} exceeds usize"
3159 ),
3160 }
3161 })?;
3162 let (header, page_data) = self.inner.inner().read_frame(cx, frame_index).await?;
3163 let page_number =
3164 PageNumber::new(header.page_number).ok_or_else(|| FrankenError::WalCorrupt {
3165 detail: format!(
3166 "parallel WAL digest frame {frame_number} has invalid page number {}",
3167 header.page_number
3168 ),
3169 })?;
3170 digest.update(page_number, header.db_size, &page_data);
3171 }
3172 Ok(digest.finalize())
3173 }
3174
3175 async fn latest_authorized_durable_certificate_record(
3176 &self,
3177 cx: &Cx,
3178 ) -> Result<Option<ParallelWalDurableCertificateRecord>> {
3179 let certificate_path = self.certificate_sidecar_path();
3180 let mut file = match self
3186 .cached_certificate_read
3187 .lock()
3188 .unwrap_or_else(std::sync::PoisonError::into_inner)
3189 .take()
3190 {
3191 Some(cached) => cached,
3192 None => {
3193 if !self
3194 .vfs
3195 .access(cx, &certificate_path, AccessFlags::EXISTS)?
3196 {
3197 return Ok(None);
3198 }
3199 let flags = VfsOpenFlags::READONLY | VfsOpenFlags::WAL;
3200 let (file, _) = self.vfs.open(cx, Some(&certificate_path), flags)?;
3201 file
3202 }
3203 };
3204 let read_result = async {
3205 let file_size = file.file_size(cx)?;
3206 if file_size == 0 {
3207 return Ok(None);
3208 }
3209
3210 let footer_size =
3213 u64::try_from(ParallelWalDurableCertificateRecord::LENGTH_FOOTER_SIZE)
3214 .unwrap_or(4);
3215 let mut newest = None;
3216 if file_size >= footer_size {
3217 let footer_offset = file_size - footer_size;
3218 let footer = Self::read_certificate_sidecar_exact(
3219 &file,
3220 cx,
3221 footer_offset,
3222 ParallelWalDurableCertificateRecord::LENGTH_FOOTER_SIZE,
3223 "newest length footer",
3224 )
3225 .await?;
3226 let record_len = usize::try_from(u32::from_le_bytes([
3227 footer[0], footer[1], footer[2], footer[3],
3228 ]))
3229 .map_err(|_| FrankenError::WalCorrupt {
3230 detail: "parallel WAL certificate newest footer length exceeds usize"
3231 .to_owned(),
3232 })?;
3233 let record_len_u64 = u64::try_from(record_len).unwrap_or(u64::MAX);
3234 if (MIN_DURABLE_CERTIFICATE_RECORD_SIZE
3235 ..=PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE)
3236 .contains(&record_len)
3237 && record_len_u64 <= file_size
3238 {
3239 let record_start = file_size - record_len_u64;
3240 let bytes = Self::read_certificate_sidecar_exact(
3241 &file,
3242 cx,
3243 record_start,
3244 record_len,
3245 "newest record",
3246 )
3247 .await?;
3248 if bytes.starts_with(&PARALLEL_WAL_DURABLE_CERTIFICATE_MAGIC)
3253 || durable_certificate_declares_len(&bytes, record_len)
3254 {
3255 let record =
3256 decode_durable_certificate_record(&bytes, "newest record")?;
3257 newest = Some((record_start, record));
3258 }
3259 }
3260 }
3261
3262 if newest.is_none() {
3263 let recovery_window_size =
3270 PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE.saturating_mul(2);
3271 let recovery_window_size_u64 =
3272 u64::try_from(recovery_window_size).unwrap_or(u64::MAX);
3273 let tail_offset = file_size.saturating_sub(recovery_window_size_u64);
3274 let tail_len =
3275 usize::try_from(file_size - tail_offset).map_err(|_| {
3276 FrankenError::WalCorrupt {
3277 detail: "parallel WAL certificate recovery window exceeds usize"
3278 .to_owned(),
3279 }
3280 })?;
3281 let tail = Self::read_certificate_sidecar_exact(
3282 &file,
3283 cx,
3284 tail_offset,
3285 tail_len,
3286 "recovery window",
3287 )
3288 .await?;
3289 let minimum_candidate_end = tail
3290 .len()
3291 .saturating_sub(PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE)
3292 .max(ParallelWalDurableCertificateRecord::LENGTH_FOOTER_SIZE);
3293 let mut anchor = None;
3294 for candidate_end in (minimum_candidate_end..tail.len()).rev() {
3295 let footer_start = candidate_end
3296 - ParallelWalDurableCertificateRecord::LENGTH_FOOTER_SIZE;
3297 let footer = &tail[footer_start..candidate_end];
3298 let record_len = usize::try_from(u32::from_le_bytes([
3299 footer[0], footer[1], footer[2], footer[3],
3300 ]))
3301 .unwrap_or(usize::MAX);
3302 if !(MIN_DURABLE_CERTIFICATE_RECORD_SIZE
3303 ..=PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE)
3304 .contains(&record_len)
3305 || record_len > candidate_end
3306 {
3307 continue;
3308 }
3309 let record_start = candidate_end - record_len;
3310 let record_bytes = &tail[record_start..candidate_end];
3311 if !record_bytes.starts_with(&PARALLEL_WAL_DURABLE_CERTIFICATE_MAGIC)
3312 || !durable_certificate_declares_len(record_bytes, record_len)
3313 {
3314 continue;
3315 }
3316 if let Ok(record) =
3317 ParallelWalDurableCertificateRecord::from_bytes(record_bytes)
3318 {
3319 anchor = Some((record_start, candidate_end, record));
3320 break;
3321 }
3322 }
3323
3324 if let Some((record_start, record_end, record)) = anchor {
3325 validate_incomplete_certificate_suffix(&tail[record_end..], true)?;
3326 let absolute_start = tail_offset
3327 .checked_add(u64::try_from(record_start).unwrap_or(u64::MAX))
3328 .ok_or_else(|| FrankenError::WalCorrupt {
3329 detail:
3330 "parallel WAL certificate recovery anchor offset overflow"
3331 .to_owned(),
3332 })?;
3333 newest = Some((absolute_start, record));
3334 } else {
3335 if file_size
3336 > u64::try_from(PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE)
3337 .unwrap_or(u64::MAX)
3338 {
3339 return Err(FrankenError::WalCorrupt {
3340 detail: format!(
3341 "parallel WAL certificate sidecar has no valid record within its bounded {}-byte recovery suffix",
3342 PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE
3343 ),
3344 });
3345 }
3346 validate_incomplete_certificate_suffix(&tail, false)?;
3347 return Ok(None);
3348 }
3349 }
3350
3351 let valid_frame_count = u64::try_from(self.inner.frame_count()).unwrap_or(u64::MAX);
3352 let wal_generation = self.inner.inner().generation_identity();
3353 let (mut record_start, mut record) = newest.ok_or_else(|| {
3354 FrankenError::WalCorrupt {
3355 detail: "parallel WAL certificate recovery produced no record".to_owned(),
3356 }
3357 })?;
3358 let mut unauthorized_records = 0_usize;
3359 loop {
3360 if record.wal_generation != wal_generation {
3365 return Ok(None);
3366 }
3367 let frame_index =
3368 usize::try_from(record.wal_frame_end.saturating_sub(1)).map_err(|_| {
3369 FrankenError::WalCorrupt {
3370 detail: "parallel WAL certificate commit-marker index exceeds usize"
3371 .to_owned(),
3372 }
3373 })?;
3374 let commit_marker_frame = if frame_index < self.inner.frame_count()
3375 && self
3376 .inner
3377 .inner()
3378 .read_frame_header(cx, frame_index)
3379 .await?
3380 .is_commit()
3381 {
3382 record.wal_frame_end
3383 } else {
3384 0
3385 };
3386 let actual_wal_frame_payload_digest =
3387 if commit_marker_frame == record.wal_frame_end {
3388 Some(
3389 self.wal_frame_payload_digest(
3390 cx,
3391 record.wal_frame_start,
3392 record.wal_frame_end,
3393 )
3394 .await?,
3395 )
3396 } else {
3397 None
3398 };
3399 if actual_wal_frame_payload_digest.is_some_and(|actual_digest| {
3400 record.authorizes_wal_boundary(
3401 wal_generation,
3402 valid_frame_count,
3403 commit_marker_frame,
3404 actual_digest,
3405 )
3406 }) {
3407 return Ok(Some(record));
3408 }
3409
3410 if record.wal_frame_end > valid_frame_count {
3423 tracing::debug!(
3424 target: "fsqlite::wal::durability_combiner",
3425 future_certificate_epoch = record.certificate.certificate_epoch,
3426 future_commit_seq_hi = record.certificate.commit_seq_hi.get(),
3427 future_wal_frame_end = record.wal_frame_end,
3428 valid_frame_count,
3429 "skipped parallel WAL certificate newer than reader frame snapshot"
3430 );
3431 } else {
3432 unauthorized_records = unauthorized_records.saturating_add(1);
3433 if unauthorized_records > MAX_ORPHAN_CERTIFICATE_LOOKBACK {
3434 return Err(FrankenError::WalCorrupt {
3435 detail: format!(
3436 "parallel WAL certificate sidecar exceeded bounded orphan lookback {MAX_ORPHAN_CERTIFICATE_LOOKBACK}"
3437 ),
3438 });
3439 }
3440 tracing::debug!(
3441 target: "fsqlite::wal::durability_combiner",
3442 orphan_certificate_epoch = record.certificate.certificate_epoch,
3443 orphan_commit_seq_hi = record.certificate.commit_seq_hi.get(),
3444 orphan_wal_frame_end = record.wal_frame_end,
3445 lookback = unauthorized_records,
3446 "ignored unauthorized parallel WAL certificate tail"
3447 );
3448 }
3449 if record_start == 0 {
3450 return Ok(None);
3451 }
3452 (record_start, record) =
3453 Self::read_certificate_record_ending_at(&file, cx, record_start).await?;
3454 }
3455 }
3456 .await;
3457 match read_result {
3458 Ok(certificate) => {
3463 *self
3464 .cached_certificate_read
3465 .lock()
3466 .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(file);
3467 Ok(certificate)
3468 }
3469 Err(error) => {
3472 let cleanup_cx = cx.create_child();
3473 let _cleanup_mask = cleanup_cx.masked();
3474 let _ = file.close(&cleanup_cx);
3475 Err(error)
3476 }
3477 }
3478 }
3479}
3480
3481impl<V> WalBackend for PathRefreshingWalBackend<V>
3482where
3483 V: Vfs + 'static,
3484 V::File: Send + Sync + 'static,
3485{
3486 fn begin_transaction<'a>(&'a mut self, cx: &'a Cx) -> WalFuture<'a, ()> {
3487 Box::pin(async move {
3488 self.ensure_current_wal_path(cx).await?;
3489 self.inner.begin_transaction(cx).await
3490 })
3491 }
3492
3493 fn published_snapshot(&self) -> Option<WalPublicationSnapshot> {
3494 Some(self.inner.published_snapshot())
3495 }
3496
3497 fn pinned_read_snapshot(&self) -> Option<WalPublicationSnapshot> {
3498 self.inner.pinned_read_snapshot()
3499 }
3500
3501 fn pinned_logical_read_snapshot<'a>(
3502 &'a self,
3503 cx: &'a Cx,
3504 ) -> WalFuture<'a, Option<WalLogicalReadSnapshot>> {
3505 Box::pin(async move {
3506 let Some(pinned) = self.inner.pinned_read_snapshot() else {
3507 return Ok(None);
3508 };
3509 let Some(last_commit_frame) = pinned.last_commit_frame else {
3510 return Ok(None);
3511 };
3512 let Some(record) = self
3513 .latest_authorized_durable_certificate_record(cx)
3514 .await?
3515 else {
3516 return Ok(None);
3517 };
3518 if record.wal_generation != pinned.generation {
3519 return Err(FrankenError::WalCorrupt {
3520 detail: "current logical WAL certificate generation differs from pinned reader"
3521 .to_owned(),
3522 });
3523 }
3524 let certificate_commit_frame =
3525 usize::try_from(record.wal_frame_end.checked_sub(1).ok_or_else(|| {
3526 FrankenError::WalCorrupt {
3527 detail: "current logical WAL certificate ends at frame zero".to_owned(),
3528 }
3529 })?)
3530 .map_err(|_| FrankenError::WalCorrupt {
3531 detail: "current logical WAL certificate frame exceeds usize".to_owned(),
3532 })?;
3533 if certificate_commit_frame > last_commit_frame {
3534 return Err(FrankenError::WalCorrupt {
3535 detail: "current logical WAL certificate extends past pinned reader horizon"
3536 .to_owned(),
3537 });
3538 }
3539
3540 let first_tail_frame =
3541 usize::try_from(record.wal_frame_end).map_err(|_| FrankenError::WalCorrupt {
3542 detail: "logical WAL tail frame exceeds usize".to_owned(),
3543 })?;
3544 let mut tail_commit_count = 0_u64;
3545 if first_tail_frame <= last_commit_frame {
3546 for frame_index in first_tail_frame..=last_commit_frame {
3547 if self
3548 .inner
3549 .inner()
3550 .read_frame_header(cx, frame_index)
3551 .await?
3552 .is_commit()
3553 {
3554 tail_commit_count = tail_commit_count.checked_add(1).ok_or_else(|| {
3555 FrankenError::WalCorrupt {
3556 detail: "logical WAL tail commit count overflow".to_owned(),
3557 }
3558 })?;
3559 }
3560 }
3561 }
3562 let visible_commit_seq = CommitSeq::new(
3563 record
3564 .certificate
3565 .commit_seq_hi
3566 .get()
3567 .checked_add(tail_commit_count)
3568 .ok_or_else(|| FrankenError::WalCorrupt {
3569 detail: "logical WAL visible commit sequence overflow".to_owned(),
3570 })?,
3571 );
3572 Ok(Some(WalLogicalReadSnapshot {
3573 generation: pinned.generation,
3574 last_commit_frame: pinned.last_commit_frame,
3575 visible_commit_seq,
3576 }))
3577 })
3578 }
3579
3580 fn refresh_published_snapshot<'a>(
3581 &'a mut self,
3582 cx: &'a Cx,
3583 ) -> WalFuture<'a, Option<WalPublicationSnapshot>> {
3584 Box::pin(async move {
3585 self.ensure_current_wal_path(cx).await?;
3586 self.inner.refresh_published_snapshot(cx).await.map(Some)
3587 })
3588 }
3589
3590 fn publish_authorized_deferred_commit<'a>(&'a mut self, cx: &'a Cx) -> WalFuture<'a, ()> {
3591 Box::pin(async move { self.inner.publish_authorized_deferred_commit(cx) })
3592 }
3593
3594 fn append_frame<'a>(
3595 &'a mut self,
3596 cx: &'a Cx,
3597 page_number: u32,
3598 page_data: &'a [u8],
3599 db_size_if_commit: u32,
3600 ) -> WalFuture<'a, ()> {
3601 Box::pin(async move {
3602 self.ensure_current_wal_path(cx).await?;
3603 self.inner
3604 .append_frame(cx, page_number, page_data, db_size_if_commit)
3605 .await
3606 })
3607 }
3608
3609 fn append_frames<'a>(
3610 &'a mut self,
3611 cx: &'a Cx,
3612 frames: &'a [WalFrameRef<'a>],
3613 ) -> WalFuture<'a, ()> {
3614 Box::pin(async move {
3615 self.ensure_current_wal_path(cx).await?;
3616 self.inner.append_frames(cx, frames).await
3617 })
3618 }
3619
3620 fn append_frames_tracked<'a>(
3621 &'a mut self,
3622 cx: &'a Cx,
3623 frames: &'a [WalFrameRef<'a>],
3624 completion: VfsWriteCompletion,
3625 ) -> WalFuture<'a, ()> {
3626 Box::pin(async move {
3627 let mut preflight = WalWriteCompletionPreflight::new(Some(&completion));
3628 self.ensure_current_wal_path(cx).await?;
3629 preflight.hand_off();
3630 drop(preflight);
3631 self.inner
3632 .append_frames_tracked(cx, frames, completion)
3633 .await
3634 })
3635 }
3636
3637 fn prepare_append_frames(
3638 &self,
3639 frames: &[WalFrameRef<'_>],
3640 ) -> Result<Option<PreparedWalFrameBatch>> {
3641 self.inner.prepare_append_frames(frames)
3642 }
3643
3644 fn finalize_prepared_frames(
3645 &self,
3646 cx: &Cx,
3647 prepared: &mut PreparedWalFrameBatch,
3648 ) -> Result<()> {
3649 self.inner.finalize_prepared_frames(cx, prepared)
3650 }
3651
3652 fn append_prepared_frames<'a>(
3653 &'a mut self,
3654 cx: &'a Cx,
3655 prepared: &'a mut PreparedWalFrameBatch,
3656 ) -> WalFuture<'a, ()> {
3657 Box::pin(async move {
3658 self.ensure_current_wal_path(cx).await?;
3659 self.inner.append_prepared_frames(cx, prepared).await
3660 })
3661 }
3662
3663 fn append_prepared_frames_tracked<'a>(
3664 &'a mut self,
3665 cx: &'a Cx,
3666 prepared: &'a mut PreparedWalFrameBatch,
3667 completion: VfsWriteCompletion,
3668 ) -> WalFuture<'a, ()> {
3669 Box::pin(async move {
3670 let mut preflight = WalWriteCompletionPreflight::new(Some(&completion));
3671 self.ensure_current_wal_path(cx).await?;
3672 preflight.hand_off();
3673 drop(preflight);
3674 self.inner
3675 .append_prepared_frames_tracked(cx, prepared, completion)
3676 .await
3677 })
3678 }
3679
3680 fn persist_parallel_wal_commit_certificate<'a>(
3681 &'a mut self,
3682 cx: &'a Cx,
3683 certificate: &'a ParallelWalCommitCertificate,
3684 wal_frame_start: u64,
3685 wal_frame_end: u64,
3686 sync: bool,
3687 ) -> WalFuture<'a, ()> {
3688 Box::pin(async move {
3689 self.ensure_current_wal_path(cx).await?;
3690 self.append_durable_certificate_record(
3691 cx,
3692 certificate,
3693 wal_frame_start,
3694 wal_frame_end,
3695 sync,
3696 )
3697 .await
3698 })
3699 }
3700
3701 fn persist_parallel_wal_commit_certificate_tracked<'a>(
3702 &'a mut self,
3703 cx: &'a Cx,
3704 certificate: &'a ParallelWalCommitCertificate,
3705 wal_frame_start: u64,
3706 wal_frame_end: u64,
3707 sync: bool,
3708 completion: VfsWriteCompletion,
3709 ) -> WalFuture<'a, ()> {
3710 Box::pin(async move {
3711 let mut preflight = WalWriteCompletionPreflight::new(Some(&completion));
3712 self.ensure_current_wal_path(cx).await?;
3713 preflight.hand_off();
3714 drop(preflight);
3715 self.append_durable_certificate_record_with_completion(
3716 cx,
3717 certificate,
3718 wal_frame_start,
3719 wal_frame_end,
3720 sync,
3721 Some(&completion),
3722 )
3723 .await
3724 })
3725 }
3726
3727 fn reconcile_parallel_wal_commit<'a>(
3728 &'a mut self,
3729 cx: &'a Cx,
3730 certificate: &'a ParallelWalCommitCertificate,
3731 wal_frame_start: u64,
3732 wal_frame_end: u64,
3733 sync: bool,
3734 ) -> WalFuture<'a, ParallelWalCommitReconciliation> {
3735 Box::pin(async move {
3736 self.ensure_current_wal_path(cx).await?;
3737 self.inner.wal.refresh(cx).await?;
3738 let wal_generation = self.inner.wal.generation_identity();
3739 let expected_record = ParallelWalDurableCertificateRecord::new(
3740 wal_generation,
3741 wal_frame_start,
3742 wal_frame_end,
3743 certificate.clone(),
3744 )
3745 .map_err(|error| {
3746 FrankenError::internal(format!(
3747 "could not reconstruct in-doubt parallel WAL certificate: {error}"
3748 ))
3749 })?;
3750
3751 let valid_frame_count = u64::try_from(self.inner.wal.frame_count()).unwrap_or(u64::MAX);
3752 let target_commit_present = if valid_frame_count < wal_frame_end {
3753 false
3754 } else {
3755 let target_index =
3756 usize::try_from(wal_frame_end.saturating_sub(1)).map_err(|_| {
3757 FrankenError::WalCorrupt {
3758 detail: "in-doubt WAL commit-marker index exceeds usize".to_owned(),
3759 }
3760 })?;
3761 self.inner
3762 .wal
3763 .read_frame_header(cx, target_index)
3764 .await?
3765 .is_commit()
3766 };
3767
3768 if target_commit_present {
3769 if valid_frame_count != wal_frame_end {
3770 return Err(FrankenError::WalCorrupt {
3771 detail: format!(
3772 "in-doubt parallel WAL interval ends at frame {wal_frame_end}, but the retained writer gate observed committed frame count {valid_frame_count}"
3773 ),
3774 });
3775 }
3776 let actual_wal_frame_payload_digest = self
3777 .wal_frame_payload_digest(cx, wal_frame_start, wal_frame_end)
3778 .await?;
3779 if !expected_record.authorizes_wal_boundary(
3780 wal_generation,
3781 valid_frame_count,
3782 wal_frame_end,
3783 actual_wal_frame_payload_digest,
3784 ) {
3785 return Err(FrankenError::WalCorrupt {
3786 detail: format!(
3787 "in-doubt parallel WAL interval {wal_frame_start}..={wal_frame_end} does not match its content-bound certificate"
3788 ),
3789 });
3790 }
3791 let sidecar_is_exact = self
3792 .reconcile_certificate_sidecar_record(cx, &expected_record, false, sync)
3793 .await?;
3794 if !sidecar_is_exact {
3795 return Err(FrankenError::WalCorrupt {
3796 detail: format!(
3797 "parallel WAL commit marker at frame {wal_frame_end} has no exact durable certificate"
3798 ),
3799 });
3800 }
3801 if sync {
3802 self.inner.wal.sync(cx, SyncFlags::NORMAL)?;
3803 self.vfs.sync_parent_directory(cx, &self.wal_path)?;
3804 }
3805 return Ok(ParallelWalCommitReconciliation::Authorized);
3806 }
3807
3808 let committed_prefix_before =
3809 wal_frame_start
3810 .checked_sub(1)
3811 .ok_or_else(|| FrankenError::WalCorrupt {
3812 detail: "parallel WAL recovery interval starts at frame zero".to_owned(),
3813 })?;
3814 if valid_frame_count != committed_prefix_before {
3815 return Err(FrankenError::WalCorrupt {
3816 detail: format!(
3817 "in-doubt WAL interval {wal_frame_start}..={wal_frame_end} has unexpected committed prefix {valid_frame_count}"
3818 ),
3819 });
3820 }
3821 self.reconcile_certificate_sidecar_record(cx, &expected_record, true, sync)
3826 .await?;
3827 self.inner.wal.repair_uncommitted_tail(cx)?;
3828 if sync {
3829 self.inner.wal.sync(cx, SyncFlags::NORMAL)?;
3830 self.vfs.sync_parent_directory(cx, &self.wal_path)?;
3831 }
3832 Ok(ParallelWalCommitReconciliation::NotCommitted)
3833 })
3834 }
3835
3836 fn latest_authorized_parallel_wal_commit_certificate<'a>(
3837 &'a mut self,
3838 cx: &'a Cx,
3839 ) -> WalFuture<'a, Option<ParallelWalCommitCertificate>> {
3840 Box::pin(async move {
3841 self.ensure_current_wal_path(cx).await?;
3842 if let Some(record) = self
3843 .latest_authorized_durable_certificate_record(cx)
3844 .await?
3845 {
3846 return Ok(Some(record.certificate));
3847 }
3848 self.checkpoint_certificate_handoff(cx).await
3849 })
3850 }
3851
3852 fn read_page<'a>(&'a mut self, cx: &'a Cx, page_number: u32) -> WalFuture<'a, Option<Vec<u8>>> {
3853 Box::pin(async move {
3854 self.ensure_current_wal_path(cx).await?;
3855 self.inner.read_page(cx, page_number).await
3856 })
3857 }
3858
3859 fn read_page_at_appended_tail<'a>(
3863 &'a mut self,
3864 cx: &'a Cx,
3865 page_number: u32,
3866 ) -> WalFuture<'a, Option<Vec<u8>>> {
3867 Box::pin(async move {
3868 self.ensure_current_wal_path(cx).await?;
3869 self.inner.read_page_at_appended_tail(cx, page_number).await
3870 })
3871 }
3872
3873 fn read_page_pinned<'a>(
3874 &'a self,
3875 cx: &'a Cx,
3876 page_number: u32,
3877 ) -> WalFuture<'a, Option<Vec<u8>>> {
3878 Box::pin(async move { self.inner.read_page_pinned(cx, page_number).await })
3879 }
3880
3881 fn supports_pinned_reads(&self) -> bool {
3882 self.inner.supports_pinned_reads()
3883 }
3884
3885 fn committed_txns_since_page<'a>(
3886 &'a mut self,
3887 cx: &'a Cx,
3888 page_number: u32,
3889 ) -> WalFuture<'a, u64> {
3890 Box::pin(async move {
3891 self.ensure_current_wal_path(cx).await?;
3892 self.inner.committed_txns_since_page(cx, page_number).await
3893 })
3894 }
3895
3896 fn conflicting_pages_since_snapshot<'a>(
3897 &'a mut self,
3898 cx: &'a Cx,
3899 snapshot: TransactionConflictSnapshot,
3900 page_numbers: &'a [u32],
3901 page_baselines: &'a [TransactionConflictPageBaseline],
3902 ) -> WalFuture<'a, Vec<u32>> {
3903 Box::pin(async move {
3904 self.ensure_current_wal_path(cx).await?;
3905 let latest = self.inner.refresh_published_snapshot(cx).await?;
3906 if latest.generation != snapshot.generation {
3907 return Ok(self
3908 .conflicts_after_generation_change(cx, page_numbers, page_baselines)
3909 .await);
3910 }
3911 self.inner
3912 .conflicting_pages_since_snapshot(cx, snapshot, page_numbers, page_baselines)
3913 .await
3914 })
3915 }
3916
3917 fn committed_txn_count<'a>(&'a mut self, cx: &'a Cx) -> WalFuture<'a, u64> {
3918 Box::pin(async move {
3919 self.ensure_current_wal_path(cx).await?;
3920 self.inner.committed_txn_count(cx).await
3921 })
3922 }
3923
3924 fn sync(&mut self, cx: &Cx) -> Result<()> {
3925 #[cfg(all(feature = "native", any(unix, windows)))]
3926 if let Some(binding) = &self.namespace_binding {
3927 binding.validate_path_identity()?;
3928 }
3929 self.inner.sync(cx)
3930 }
3931
3932 fn frame_count(&self) -> usize {
3933 self.inner.frame_count()
3934 }
3935
3936 fn checkpoint<'a>(
3937 &'a mut self,
3938 cx: &'a Cx,
3939 mode: CheckpointMode,
3940 writer: &'a mut dyn CheckpointPageWriter,
3941 backfilled_frames: u32,
3942 oldest_reader_frame: Option<u32>,
3943 ) -> WalFuture<'a, CheckpointResult> {
3944 Box::pin(async move {
3945 self.ensure_current_wal_path(cx).await?;
3946 let checkpoint_handoff = self
3947 .latest_authorized_durable_certificate_record(cx)
3948 .await?;
3949 if let Some(record) = checkpoint_handoff.as_ref() {
3950 self.persist_checkpoint_certificate_handoff(cx, record)
3957 .await?;
3958 }
3959 let result = self
3960 .inner
3961 .checkpoint(cx, mode, writer, backfilled_frames, oldest_reader_frame)
3962 .await?;
3963 if let Some(mut stale) = self
3967 .cached_certificate_read
3968 .get_mut()
3969 .unwrap_or_else(std::sync::PoisonError::into_inner)
3970 .take()
3971 {
3972 let cleanup_cx = cx.create_child();
3973 let _cleanup_mask = cleanup_cx.masked();
3974 let _ = stale.close(&cleanup_cx);
3975 }
3976 Ok(result)
3977 })
3978 }
3979}
3980
3981struct CheckpointTargetAdapterRef<'a> {
3986 writer: &'a mut dyn CheckpointPageWriter,
3987}
3988
3989impl CheckpointTarget for CheckpointTargetAdapterRef<'_> {
3990 fn write_page<'a>(
3991 &'a mut self,
3992 cx: &'a Cx,
3993 page_no: PageNumber,
3994 data: &'a [u8],
3995 ) -> CheckpointTargetFuture<'a, ()> {
3996 Box::pin(async move { self.writer.write_page(cx, page_no, data).await })
3997 }
3998
3999 fn truncate_db<'a>(&'a mut self, cx: &'a Cx, n_pages: u32) -> CheckpointTargetFuture<'a, ()> {
4000 Box::pin(async move { self.writer.truncate(cx, n_pages).await })
4001 }
4002
4003 fn sync_db<'a>(&'a mut self, cx: &'a Cx) -> CheckpointTargetFuture<'a, ()> {
4004 Box::pin(async move { self.writer.sync(cx).await })
4005 }
4006}
4007
4008#[cfg(test)]
4013mod tests {
4014 use std::sync::Mutex;
4015
4016 use fsqlite_pager::MockCheckpointPageWriter;
4017 use fsqlite_pager::traits::WalFrameRef;
4018 use fsqlite_types::flags::VfsOpenFlags;
4019 use fsqlite_vfs::MemoryVfs;
4020 use fsqlite_vfs::traits::{Vfs, VfsFile};
4021 use fsqlite_wal::checksum::WalSalts;
4022
4023 use super::*;
4024
4025 const PAGE_SIZE: u32 = 4096;
4026 const CERTIFICATE_PATH: &str = "test.db-wal-cert";
4027 const CHECKPOINT_HANDOFF_PATH: &str = "test.db-wal-cert-head";
4028
4029 #[derive(Clone, Copy, Debug)]
4030 enum CheckpointHandoffWriteFault {
4031 Error,
4032 Pending,
4033 }
4034
4035 #[derive(Clone, Debug, Eq, PartialEq)]
4036 enum CertificateSyncObservation {
4037 Ordinary(PathBuf),
4038 Durable(PathBuf, SyncKind),
4039 }
4040
4041 #[derive(Debug, Default)]
4042 struct CheckpointHandoffFaultState {
4043 next_write: Option<CheckpointHandoffWriteFault>,
4044 fail_next_sync: bool,
4045 fail_next_wal_sync: bool,
4047 sync_observations: Vec<CertificateSyncObservation>,
4048 }
4049
4050 #[derive(Clone, Debug)]
4051 struct CheckpointHandoffFaultVfs {
4052 inner: MemoryVfs,
4053 faults: Arc<Mutex<CheckpointHandoffFaultState>>,
4054 }
4055
4056 impl CheckpointHandoffFaultVfs {
4057 fn new() -> Self {
4058 Self {
4059 inner: MemoryVfs::new(),
4060 faults: Arc::new(Mutex::new(CheckpointHandoffFaultState::default())),
4061 }
4062 }
4063
4064 fn fail_next_handoff_write(&self) {
4065 self.faults
4066 .lock()
4067 .unwrap_or_else(std::sync::PoisonError::into_inner)
4068 .next_write = Some(CheckpointHandoffWriteFault::Error);
4069 }
4070
4071 fn pend_next_handoff_write(&self) {
4072 self.faults
4073 .lock()
4074 .unwrap_or_else(std::sync::PoisonError::into_inner)
4075 .next_write = Some(CheckpointHandoffWriteFault::Pending);
4076 }
4077
4078 fn fail_next_handoff_sync(&self) {
4079 self.faults
4080 .lock()
4081 .unwrap_or_else(std::sync::PoisonError::into_inner)
4082 .fail_next_sync = true;
4083 }
4084
4085 fn fail_next_wal_sync(&self) {
4087 self.faults
4088 .lock()
4089 .unwrap_or_else(std::sync::PoisonError::into_inner)
4090 .fail_next_wal_sync = true;
4091 }
4092
4093 fn take_sync_observations(&self) -> Vec<CertificateSyncObservation> {
4094 std::mem::take(
4095 &mut self
4096 .faults
4097 .lock()
4098 .unwrap_or_else(std::sync::PoisonError::into_inner)
4099 .sync_observations,
4100 )
4101 }
4102 }
4103
4104 #[derive(Debug)]
4105 struct CheckpointHandoffFaultFile {
4106 inner: <MemoryVfs as Vfs>::File,
4107 faults: Arc<Mutex<CheckpointHandoffFaultState>>,
4108 path: Option<PathBuf>,
4109 is_checkpoint_handoff: bool,
4110 }
4111
4112 impl Vfs for CheckpointHandoffFaultVfs {
4113 type File = CheckpointHandoffFaultFile;
4114
4115 fn name(&self) -> &'static str {
4116 "checkpoint-handoff-fault"
4117 }
4118
4119 fn open(
4120 &self,
4121 cx: &Cx,
4122 path: Option<&Path>,
4123 flags: VfsOpenFlags,
4124 ) -> Result<(Self::File, VfsOpenFlags)> {
4125 let is_checkpoint_handoff =
4126 path.is_some_and(|candidate| candidate == Path::new(CHECKPOINT_HANDOFF_PATH));
4127 let (inner, actual_flags) = self.inner.open(cx, path, flags)?;
4128 Ok((
4129 CheckpointHandoffFaultFile {
4130 inner,
4131 faults: Arc::clone(&self.faults),
4132 path: path.map(Path::to_path_buf),
4133 is_checkpoint_handoff,
4134 },
4135 actual_flags,
4136 ))
4137 }
4138
4139 fn delete(&self, cx: &Cx, path: &Path, sync_dir: bool) -> Result<()> {
4140 self.inner.delete(cx, path, sync_dir)
4141 }
4142
4143 fn sync_parent_directory(&self, cx: &Cx, path: &Path) -> Result<()> {
4144 self.inner.sync_parent_directory(cx, path)
4145 }
4146
4147 fn access(&self, cx: &Cx, path: &Path, flags: AccessFlags) -> Result<bool> {
4148 self.inner.access(cx, path, flags)
4149 }
4150
4151 fn path_entry_exists(&self, cx: &Cx, path: &Path) -> Result<bool> {
4152 self.inner.path_entry_exists(cx, path)
4153 }
4154
4155 fn full_pathname(&self, cx: &Cx, path: &Path) -> Result<PathBuf> {
4156 self.inner.full_pathname(cx, path)
4157 }
4158
4159 fn randomness(&self, cx: &Cx, buf: &mut [u8]) {
4160 self.inner.randomness(cx, buf);
4161 }
4162
4163 fn current_time(&self, cx: &Cx) -> f64 {
4164 self.inner.current_time(cx)
4165 }
4166
4167 fn is_memory(&self) -> bool {
4168 true
4169 }
4170 }
4171
4172 impl VfsFile for CheckpointHandoffFaultFile {
4173 fn close(&mut self, cx: &Cx) -> Result<()> {
4174 self.inner.close(cx)
4175 }
4176
4177 fn file_identity(&self) -> Result<Option<fsqlite_vfs::FileIdentity>> {
4178 self.inner.file_identity()
4179 }
4180
4181 fn read<'a>(
4182 &'a self,
4183 cx: &'a Cx,
4184 buf: &'a mut [u8],
4185 offset: u64,
4186 ) -> impl std::future::Future<Output = Result<usize>> + Send + 'a {
4187 self.inner.read(cx, buf, offset)
4188 }
4189
4190 async fn write<'a>(&'a self, cx: &'a Cx, buf: &'a [u8], offset: u64) -> Result<()> {
4191 let fault = if self.is_checkpoint_handoff {
4192 self.faults
4193 .lock()
4194 .unwrap_or_else(std::sync::PoisonError::into_inner)
4195 .next_write
4196 .take()
4197 } else {
4198 None
4199 };
4200 match fault {
4201 Some(CheckpointHandoffWriteFault::Error) => Err(FrankenError::Io(
4202 std::io::Error::other("injected checkpoint handoff write failure"),
4203 )),
4204 Some(CheckpointHandoffWriteFault::Pending) => {
4205 std::future::pending::<Result<()>>().await
4206 }
4207 None => self.inner.write(cx, buf, offset).await,
4208 }
4209 }
4210
4211 fn truncate(&mut self, cx: &Cx, size: u64) -> Result<()> {
4212 self.inner.truncate(cx, size)
4213 }
4214
4215 fn sync(&mut self, cx: &Cx, flags: SyncFlags) -> Result<()> {
4216 let mut faults = self
4217 .faults
4218 .lock()
4219 .unwrap_or_else(std::sync::PoisonError::into_inner);
4220 if let Some(path) = self.path.as_ref().filter(|path| {
4221 path.as_path() == Path::new(CERTIFICATE_PATH)
4222 || path.as_path() == Path::new(CHECKPOINT_HANDOFF_PATH)
4223 }) {
4224 faults
4225 .sync_observations
4226 .push(CertificateSyncObservation::Ordinary(path.clone()));
4227 }
4228 let fail = self.is_checkpoint_handoff && std::mem::take(&mut faults.fail_next_sync);
4229 let fail_wal =
4230 !self.is_checkpoint_handoff && std::mem::take(&mut faults.fail_next_wal_sync);
4231 drop(faults);
4232 if fail {
4233 Err(FrankenError::Io(std::io::Error::other(
4234 "injected checkpoint handoff sync failure",
4235 )))
4236 } else if fail_wal {
4237 Err(FrankenError::Io(std::io::Error::other(
4238 "injected WAL sync failure",
4239 )))
4240 } else {
4241 self.inner.sync(cx, flags)
4242 }
4243 }
4244
4245 fn durable_sync(&mut self, cx: &Cx, kind: SyncKind) -> Result<()> {
4246 let mut faults = self
4247 .faults
4248 .lock()
4249 .unwrap_or_else(std::sync::PoisonError::into_inner);
4250 if let Some(path) = self.path.as_ref().filter(|path| {
4251 path.as_path() == Path::new(CERTIFICATE_PATH)
4252 || path.as_path() == Path::new(CHECKPOINT_HANDOFF_PATH)
4253 }) {
4254 faults
4255 .sync_observations
4256 .push(CertificateSyncObservation::Durable(path.clone(), kind));
4257 }
4258 let fail = self.is_checkpoint_handoff && std::mem::take(&mut faults.fail_next_sync);
4259 drop(faults);
4260 if fail {
4261 Err(FrankenError::Io(std::io::Error::other(
4262 "injected checkpoint handoff durable-sync failure",
4263 )))
4264 } else {
4265 self.inner.durable_sync(cx, kind)
4266 }
4267 }
4268
4269 fn file_size(&self, cx: &Cx) -> Result<u64> {
4270 self.inner.file_size(cx)
4271 }
4272
4273 fn lock(&mut self, cx: &Cx, level: fsqlite_types::LockLevel) -> Result<()> {
4274 self.inner.lock(cx, level)
4275 }
4276
4277 fn unlock(&mut self, cx: &Cx, level: fsqlite_types::LockLevel) -> Result<()> {
4278 self.inner.unlock(cx, level)
4279 }
4280
4281 fn lock_external_shared_snapshot(&mut self, cx: &Cx) -> Result<()> {
4282 self.inner.lock_external_shared_snapshot(cx)
4283 }
4284
4285 fn restore_external_shared_snapshot_attempt(&mut self, cx: &Cx) -> Result<()> {
4286 self.inner.restore_external_shared_snapshot_attempt(cx)
4287 }
4288
4289 fn lock_external_maintenance(&mut self, cx: &Cx, wal_mode: bool) -> Result<()> {
4290 self.inner.lock_external_maintenance(cx, wal_mode)
4291 }
4292
4293 fn restore_external_maintenance_attempt(&mut self, cx: &Cx) -> Result<()> {
4294 self.inner.restore_external_maintenance_attempt(cx)
4295 }
4296
4297 fn check_reserved_lock(&self, cx: &Cx) -> Result<bool> {
4298 self.inner.check_reserved_lock(cx)
4299 }
4300
4301 fn sector_size(&self) -> u32 {
4302 self.inner.sector_size()
4303 }
4304
4305 fn device_characteristics(&self) -> u32 {
4306 self.inner.device_characteristics()
4307 }
4308
4309 fn shm_map(
4310 &mut self,
4311 cx: &Cx,
4312 region: u32,
4313 size: u32,
4314 extend: bool,
4315 ) -> Result<fsqlite_vfs::ShmRegion> {
4316 self.inner.shm_map(cx, region, size, extend)
4317 }
4318
4319 fn shm_lock(&mut self, cx: &Cx, offset: u32, n: u32, flags: u32) -> Result<()> {
4320 self.inner.shm_lock(cx, offset, n, flags)
4321 }
4322
4323 fn shm_barrier(&self) {
4324 self.inner.shm_barrier();
4325 }
4326
4327 fn shm_unmap(&mut self, cx: &Cx, delete: bool) -> Result<()> {
4328 self.inner.shm_unmap(cx, delete)
4329 }
4330
4331 fn set_busy_timeout_ms(&mut self, ms: u64) {
4332 self.inner.set_busy_timeout_ms(ms);
4333 }
4334 }
4335
4336 fn init_wal_publication_test_tracing() {}
4352
4353 #[test]
4363 fn wal_publication_tracing_helper_installs_no_global_subscriber() {
4364 let before = tracing::dispatcher::has_been_set();
4365 init_wal_publication_test_tracing();
4366
4367 assert_eq!(
4368 before,
4369 tracing::dispatcher::has_been_set(),
4370 "init_wal_publication_test_tracing must not install or alter a global subscriber"
4371 );
4372 }
4373
4374 fn test_cx() -> Cx {
4375 Cx::default()
4376 }
4377
4378 fn test_salts() -> WalSalts {
4379 WalSalts {
4380 salt1: 0xDEAD_BEEF,
4381 salt2: 0xCAFE_BABE,
4382 }
4383 }
4384
4385 fn sample_page(seed: u8) -> Vec<u8> {
4386 let page_size = usize::try_from(PAGE_SIZE).expect("page size fits usize");
4387 let mut page = vec![0u8; page_size];
4388 for (i, byte) in page.iter_mut().enumerate() {
4389 let reduced = u8::try_from(i % 251).expect("modulo fits u8");
4390 *byte = reduced ^ seed;
4391 }
4392 page
4393 }
4394
4395 fn test_frame_payload_digest(
4396 page_number: u32,
4397 page_data: &[u8],
4398 db_size_if_commit: u32,
4399 ) -> [u8; 32] {
4400 let mut digest = ParallelWalFramePayloadDigestBuilder::new();
4401 digest.update(
4402 PageNumber::new(page_number).expect("test page number must be valid"),
4403 db_size_if_commit,
4404 page_data,
4405 );
4406 digest.finalize()
4407 }
4408
4409 fn sample_certificate(
4410 certificate_epoch: u64,
4411 commit_seq: u64,
4412 lane_record_counts: Vec<u32>,
4413 ) -> ParallelWalCommitCertificate {
4414 let lane_count = u16::try_from(lane_record_counts.len()).expect("test lane count fits u16");
4415 let mut certificate = ParallelWalCommitCertificate {
4416 format_version: fsqlite_wal::PARALLEL_WAL_COMMIT_CERTIFICATE_VERSION,
4417 residue: fsqlite_wal::ParallelWalOrderedResidue::CommitCertificateThenPublish,
4418 certificate_epoch,
4419 commit_seq_lo: fsqlite_types::CommitSeq::new(commit_seq),
4420 commit_seq_hi: fsqlite_types::CommitSeq::new(commit_seq),
4421 durable_segment_epoch: certificate_epoch,
4422 lane_count,
4423 lane_record_counts,
4424 db_size_pages: 1,
4425 page_set_size: 1,
4426 wal_frame_payload_digest: [0xA5; 32],
4427 certificate_crc32c: 0,
4428 fallback_active: false,
4429 };
4430 certificate.certificate_crc32c = certificate.computed_crc32c();
4431 certificate
4432 }
4433
4434 fn make_path_refreshing_backend(
4435 vfs: &MemoryVfs,
4436 cx: &Cx,
4437 ) -> PathRefreshingWalBackend<MemoryVfs> {
4438 let wal = WalFile::create(cx, open_wal_file(vfs, cx), PAGE_SIZE, 0, test_salts())
4439 .expect("create WAL");
4440 PathRefreshingWalBackend::new(
4441 vfs.clone(),
4442 std::path::Path::new("test.db"),
4443 std::path::Path::new("test.db-wal"),
4444 PAGE_SIZE,
4445 wal,
4446 true,
4447 #[cfg(all(feature = "native", any(unix, windows)))]
4448 None,
4449 )
4450 }
4451
4452 fn make_authorized_certificate_backend(
4453 vfs: &MemoryVfs,
4454 cx: &Cx,
4455 ) -> (
4456 PathRefreshingWalBackend<MemoryVfs>,
4457 ParallelWalCommitCertificate,
4458 ) {
4459 let mut backend = make_path_refreshing_backend(vfs, cx);
4460 let committed_page = sample_page(0x44);
4461 let mut certificate = sample_certificate(1, 1, vec![1]);
4462 certificate.wal_frame_payload_digest = test_frame_payload_digest(1, &committed_page, 1);
4463 certificate.certificate_crc32c = certificate.computed_crc32c();
4464 backend
4465 .persist_parallel_wal_commit_certificate(cx, &certificate, 1, 1, true)
4466 .expect("persist authorized certificate");
4467 backend
4468 .append_frame(cx, 1, &committed_page, 1)
4469 .expect("append matching commit marker");
4470 backend.sync(cx).expect("sync matching commit marker");
4471 (backend, certificate)
4472 }
4473
4474 struct AuthoritativeWalSnapshot {
4475 generation: WalGenerationIdentity,
4476 frame_count: usize,
4477 wal_bytes: Vec<u8>,
4478 certificate: ParallelWalCommitCertificate,
4479 committed_page: Vec<u8>,
4480 }
4481
4482 fn make_checkpoint_handoff_fault_backend(
4483 vfs: &CheckpointHandoffFaultVfs,
4484 cx: &Cx,
4485 ) -> (
4486 PathRefreshingWalBackend<CheckpointHandoffFaultVfs>,
4487 ParallelWalCommitCertificate,
4488 Vec<u8>,
4489 ) {
4490 let flags = VfsOpenFlags::READWRITE | VfsOpenFlags::CREATE | VfsOpenFlags::WAL;
4491 let (file, _) = vfs
4492 .open(cx, Some(Path::new("test.db-wal")), flags)
4493 .expect("open fault-injected WAL file");
4494 let wal = WalFile::create(cx, file, PAGE_SIZE, 0, test_salts())
4495 .expect("create fault-injected WAL");
4496 let mut backend = PathRefreshingWalBackend::new(
4497 vfs.clone(),
4498 Path::new("test.db"),
4499 Path::new("test.db-wal"),
4500 PAGE_SIZE,
4501 wal,
4502 true,
4503 #[cfg(all(feature = "native", any(unix, windows)))]
4504 None,
4505 );
4506 let committed_page = sample_page(0x47);
4507 let mut certificate = sample_certificate(1, 1, vec![1]);
4508 certificate.wal_frame_payload_digest = test_frame_payload_digest(1, &committed_page, 1);
4509 certificate.certificate_crc32c = certificate.computed_crc32c();
4510 backend
4511 .persist_parallel_wal_commit_certificate(cx, &certificate, 1, 1, true)
4512 .expect("persist authorized certificate");
4513 backend
4514 .append_frame(cx, 1, &committed_page, 1)
4515 .expect("append matching commit marker");
4516 backend.sync(cx).expect("sync matching commit marker");
4517 (backend, certificate, committed_page)
4518 }
4519
4520 fn read_fault_injected_wal(vfs: &CheckpointHandoffFaultVfs, cx: &Cx) -> Vec<u8> {
4521 let flags = VfsOpenFlags::READONLY | VfsOpenFlags::WAL;
4522 let (mut file, _) = vfs
4523 .open(cx, Some(Path::new("test.db-wal")), flags)
4524 .expect("open WAL snapshot");
4525 let len = usize::try_from(file.file_size(cx).expect("read WAL size"))
4526 .expect("WAL size fits usize");
4527 let mut bytes = vec![0_u8; len];
4528 assert_eq!(
4529 file.read(cx, &mut bytes, 0).expect("read WAL snapshot"),
4530 len
4531 );
4532 file.close(cx).expect("close WAL snapshot");
4533 bytes
4534 }
4535
4536 fn capture_authoritative_wal(
4537 backend: &PathRefreshingWalBackend<CheckpointHandoffFaultVfs>,
4538 vfs: &CheckpointHandoffFaultVfs,
4539 cx: &Cx,
4540 certificate: ParallelWalCommitCertificate,
4541 committed_page: Vec<u8>,
4542 ) -> AuthoritativeWalSnapshot {
4543 AuthoritativeWalSnapshot {
4544 generation: backend.inner.inner().generation_identity(),
4545 frame_count: backend.inner.frame_count(),
4546 wal_bytes: read_fault_injected_wal(vfs, cx),
4547 certificate,
4548 committed_page,
4549 }
4550 }
4551
4552 fn assert_authoritative_wal_unchanged(
4553 backend: &mut PathRefreshingWalBackend<CheckpointHandoffFaultVfs>,
4554 vfs: &CheckpointHandoffFaultVfs,
4555 cx: &Cx,
4556 before: &AuthoritativeWalSnapshot,
4557 ) {
4558 assert_eq!(
4559 backend.inner.inner().generation_identity(),
4560 before.generation,
4561 "checkpoint handoff failure must not reset the WAL generation"
4562 );
4563 assert_eq!(
4564 backend.inner.frame_count(),
4565 before.frame_count,
4566 "checkpoint handoff failure must not change the visible frame count"
4567 );
4568 assert_eq!(
4569 read_fault_injected_wal(vfs, cx),
4570 before.wal_bytes,
4571 "checkpoint handoff failure must leave the authoritative WAL byte-for-byte unchanged"
4572 );
4573 assert!(
4574 backend
4575 .inner
4576 .inner()
4577 .read_frame_header(cx, 0)
4578 .expect("read original commit frame")
4579 .is_commit(),
4580 "the original generation's commit marker must remain authoritative"
4581 );
4582 assert_eq!(
4583 backend
4584 .latest_authorized_parallel_wal_commit_certificate(cx)
4585 .expect("recover certificate from unchanged WAL generation"),
4586 Some(before.certificate.clone())
4587 );
4588 assert_eq!(
4589 backend
4590 .read_page(cx, 1)
4591 .expect("read committed page from unchanged WAL generation"),
4592 Some(before.committed_page.clone())
4593 );
4594 }
4595
4596 fn read_certificate_sidecar(vfs: &MemoryVfs, cx: &Cx) -> Vec<u8> {
4597 let path = std::path::Path::new("test.db-wal-cert");
4598 let (mut file, _) = vfs
4599 .open(cx, Some(path), VfsOpenFlags::READONLY | VfsOpenFlags::WAL)
4600 .expect("open certificate sidecar");
4601 let len = usize::try_from(file.file_size(cx).expect("read certificate sidecar size"))
4602 .expect("certificate sidecar size fits usize");
4603 let mut bytes = vec![0_u8; len];
4604 assert_eq!(
4605 file.read(cx, &mut bytes, 0)
4606 .expect("read certificate sidecar"),
4607 len
4608 );
4609 file.close(cx).expect("close certificate sidecar");
4610 bytes
4611 }
4612
4613 fn replace_certificate_sidecar(vfs: &MemoryVfs, cx: &Cx, bytes: &[u8]) {
4614 let path = std::path::Path::new("test.db-wal-cert");
4615 let flags = VfsOpenFlags::READWRITE | VfsOpenFlags::CREATE | VfsOpenFlags::WAL;
4616 let (mut file, _) = vfs
4617 .open(cx, Some(path), flags)
4618 .expect("open mutable certificate sidecar");
4619 file.truncate(cx, 0)
4620 .expect("truncate mutable certificate sidecar");
4621 file.write(cx, bytes, 0)
4622 .expect("replace certificate sidecar bytes");
4623 file.close(cx).expect("close mutable certificate sidecar");
4624 }
4625
4626 fn assert_wal_corrupt<T: std::fmt::Debug>(result: Result<T>, scenario: &str) {
4627 assert!(
4628 matches!(&result, Err(FrankenError::WalCorrupt { .. })),
4629 "{scenario} must fail closed with WalCorrupt, got {result:?}"
4630 );
4631 }
4632
4633 fn sqlite_page_one(encoded_page_size: u16) -> Vec<u8> {
4634 let mut page = sample_page(0x11);
4635 page[..16].copy_from_slice(b"SQLite format 3\0");
4636 page[16..18].copy_from_slice(&encoded_page_size.to_be_bytes());
4637 page
4638 }
4639
4640 fn write_main_db_pages(vfs: &MemoryVfs, cx: &Cx, pages: &[Vec<u8>]) {
4641 let flags = VfsOpenFlags::READWRITE | VfsOpenFlags::CREATE | VfsOpenFlags::MAIN_DB;
4642 let (mut file, _) = vfs
4643 .open(cx, Some(std::path::Path::new("test.db")), flags)
4644 .expect("open main database");
4645 file.truncate(cx, 0).expect("truncate main database");
4646 for (index, page) in pages.iter().enumerate() {
4647 let offset = u64::try_from(index)
4648 .expect("page index fits u64")
4649 .saturating_mul(u64::from(PAGE_SIZE));
4650 file.write(cx, page, offset).expect("write database page");
4651 }
4652 file.close(cx).expect("close main database");
4653 }
4654
4655 fn replacement_salts() -> WalSalts {
4656 WalSalts {
4657 salt1: 0x1234_5678,
4658 salt2: 0x9ABC_DEF0,
4659 }
4660 }
4661
4662 fn replace_path_visible_wal(vfs: &MemoryVfs, cx: &Cx) {
4663 let wal_path = std::path::Path::new("test.db-wal");
4664 vfs.delete(cx, wal_path, false)
4665 .expect("remove old path-visible WAL");
4666 let file = open_wal_file(vfs, cx);
4667 WalFile::create(cx, file, PAGE_SIZE, 1, replacement_salts())
4668 .expect("create replacement WAL")
4669 .close(cx)
4670 .expect("close replacement WAL");
4671 }
4672
4673 fn append_replacement_wal_page(
4674 vfs: &MemoryVfs,
4675 cx: &Cx,
4676 page_number: u32,
4677 page: &[u8],
4678 db_size_if_commit: u32,
4679 ) {
4680 let file = open_wal_file(vfs, cx);
4681 let wal = WalFile::open(cx, file).expect("open replacement WAL");
4682 let mut adapter = WalBackendAdapter::new(wal);
4683 adapter
4684 .append_frame(cx, page_number, page, db_size_if_commit)
4685 .expect("append replacement WAL page");
4686 adapter.sync(cx).expect("sync replacement WAL page");
4687 adapter
4688 .into_inner()
4689 .expect("sync drained the staged frames")
4690 .close(cx)
4691 .expect("close replacement WAL");
4692 }
4693
4694 fn make_generation_transition_backend(
4695 vfs: &MemoryVfs,
4696 cx: &Cx,
4697 ) -> (
4698 PathRefreshingWalBackend<MemoryVfs>,
4699 TransactionConflictSnapshot,
4700 Vec<u8>,
4701 ) {
4702 let page_one = sqlite_page_one(u16::try_from(PAGE_SIZE).expect("page size fits u16"));
4703 let page_two = sample_page(0x22);
4704 write_main_db_pages(vfs, cx, &[page_one.clone(), page_two.clone()]);
4705
4706 let file = open_wal_file(vfs, cx);
4707 let wal =
4708 WalFile::create(cx, file, PAGE_SIZE, 0, test_salts()).expect("create original WAL");
4709 let mut backend = PathRefreshingWalBackend::new(
4710 vfs.clone(),
4711 std::path::Path::new("test.db"),
4712 std::path::Path::new("test.db-wal"),
4713 PAGE_SIZE,
4714 wal,
4715 true,
4716 #[cfg(all(feature = "native", any(unix, windows)))]
4717 None,
4718 );
4719 backend
4720 .append_frame(cx, 1, &page_one, 0)
4721 .expect("append original page 1");
4722 backend
4723 .append_frame(cx, 2, &page_two, 2)
4724 .expect("append original commit");
4725 backend.sync(cx).expect("publish original commit");
4729 backend
4730 .begin_transaction(cx)
4731 .expect("pin original WAL generation");
4732 let pinned = backend.pinned_read_snapshot().expect("pinned WAL snapshot");
4733 let snapshot = TransactionConflictSnapshot {
4734 generation: pinned.generation,
4735 last_commit_frame: pinned.last_commit_frame,
4736 commit_count: pinned.commit_count,
4737 snapshot_db_size: 0,
4738 };
4739 replace_path_visible_wal(vfs, cx);
4740 (backend, snapshot, page_two)
4741 }
4742
4743 #[test]
4744 fn durable_certificate_sidecar_precedes_and_reconstructs_wal_commit() {
4745 let cx = test_cx();
4746 let vfs = MemoryVfs::new();
4747 let committed_page = sample_page(0x44);
4748 let file = open_wal_file(&vfs, &cx);
4749 let wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create WAL");
4750 let mut backend = PathRefreshingWalBackend::new(
4751 vfs.clone(),
4752 std::path::Path::new("test.db"),
4753 std::path::Path::new("test.db-wal"),
4754 PAGE_SIZE,
4755 wal,
4756 true,
4757 #[cfg(all(feature = "native", any(unix, windows)))]
4758 None,
4759 );
4760 let mut certificate = ParallelWalCommitCertificate {
4761 format_version: fsqlite_wal::PARALLEL_WAL_COMMIT_CERTIFICATE_VERSION,
4762 residue: fsqlite_wal::ParallelWalOrderedResidue::CommitCertificateThenPublish,
4763 certificate_epoch: 1,
4764 commit_seq_lo: fsqlite_types::CommitSeq::new(1),
4765 commit_seq_hi: fsqlite_types::CommitSeq::new(1),
4766 durable_segment_epoch: 1,
4767 lane_count: 1,
4768 lane_record_counts: vec![1],
4769 db_size_pages: 1,
4770 page_set_size: 1,
4771 wal_frame_payload_digest: test_frame_payload_digest(1, &committed_page, 1),
4772 certificate_crc32c: 0,
4773 fallback_active: false,
4774 };
4775 certificate.certificate_crc32c = certificate.computed_crc32c();
4776
4777 backend
4778 .persist_parallel_wal_commit_certificate(&cx, &certificate, 1, 1, true)
4779 .expect("persist certificate before WAL commit marker");
4780 assert_eq!(
4781 backend.inner.frame_count(),
4782 0,
4783 "certificate persistence must not itself expose a WAL commit marker"
4784 );
4785
4786 let certificate_path = std::path::Path::new("test.db-wal-cert");
4787 let (mut certificate_file, _) = vfs
4788 .open(
4789 &cx,
4790 Some(certificate_path),
4791 VfsOpenFlags::READONLY | VfsOpenFlags::WAL,
4792 )
4793 .expect("open certificate sidecar");
4794 let certificate_len = usize::try_from(
4795 certificate_file
4796 .file_size(&cx)
4797 .expect("certificate sidecar size"),
4798 )
4799 .expect("certificate sidecar size fits usize");
4800 let mut record_bytes = vec![0_u8; certificate_len];
4801 assert_eq!(
4802 certificate_file
4803 .read(&cx, &mut record_bytes, 0)
4804 .expect("read certificate sidecar"),
4805 certificate_len
4806 );
4807 certificate_file
4808 .close(&cx)
4809 .expect("close certificate sidecar");
4810 let reconstructed = ParallelWalDurableCertificateRecord::from_bytes(&record_bytes)
4811 .expect("reconstruct durable certificate record");
4812 assert_eq!(reconstructed.certificate, certificate);
4813 assert_eq!(reconstructed.wal_frame_start, 1);
4814 assert_eq!(reconstructed.wal_frame_end, 1);
4815 assert_eq!(
4816 reconstructed.wal_generation,
4817 backend.inner.inner().generation_identity()
4818 );
4819 assert!(
4820 !reconstructed.authorizes_wal_boundary(
4821 backend.inner.inner().generation_identity(),
4822 0,
4823 0,
4824 test_frame_payload_digest(1, &committed_page, 1),
4825 ),
4826 "orphan certificate must not authorize visibility before the matching commit marker"
4827 );
4828
4829 backend
4830 .append_frame(&cx, 1, &committed_page, 1)
4831 .expect("append matching WAL commit marker");
4832 backend.sync(&cx).expect("sync WAL commit marker");
4833 assert!(
4834 backend
4835 .inner
4836 .inner()
4837 .read_frame_header(&cx, 0)
4838 .expect("read matching WAL commit frame")
4839 .is_commit()
4840 );
4841 assert!(reconstructed.authorizes_wal_boundary(
4842 backend.inner.inner().generation_identity(),
4843 1,
4844 1,
4845 test_frame_payload_digest(1, &committed_page, 1),
4846 ));
4847
4848 let (mut certificate_file, _) = vfs
4849 .open(
4850 &cx,
4851 Some(certificate_path),
4852 VfsOpenFlags::READWRITE | VfsOpenFlags::WAL,
4853 )
4854 .expect("reopen certificate sidecar");
4855 let torn_offset = certificate_file
4856 .file_size(&cx)
4857 .expect("certificate sidecar size before torn tail");
4858 certificate_file
4859 .write(&cx, &[0xA5], torn_offset)
4860 .expect("append torn footer byte");
4861 certificate_file
4862 .close(&cx)
4863 .expect("close sidecar with torn tail");
4864 let recovered = backend
4865 .latest_authorized_parallel_wal_commit_certificate(&cx)
4866 .wait()
4867 .expect("torn certificate tail should recover the prior valid record")
4868 .expect("prior authorized certificate should remain discoverable");
4869 assert_eq!(recovered, certificate);
4870 }
4871
4872 #[test]
4873 fn content_mismatched_wal_interval_cannot_be_authorized_or_repaired() {
4874 let cx = test_cx();
4875 let vfs = MemoryVfs::new();
4876 let certified_page = sample_page(0x61);
4877 let actual_page = sample_page(0x62);
4878 let mut backend = make_path_refreshing_backend(&vfs, &cx);
4879 let mut certificate = sample_certificate(1, 1, vec![1]);
4880 certificate.wal_frame_payload_digest = test_frame_payload_digest(1, &certified_page, 1);
4881 certificate.certificate_crc32c = certificate.computed_crc32c();
4882
4883 backend
4884 .persist_parallel_wal_commit_certificate(&cx, &certificate, 1, 1, true)
4885 .expect("persist content-bound certificate");
4886 backend
4887 .append_frame(&cx, 1, &actual_page, 1)
4888 .expect("append differently valued commit frame");
4889 backend.sync(&cx).expect("sync mismatched commit frame");
4890
4891 let sidecar_before = read_certificate_sidecar(&vfs, &cx);
4892 assert!(
4893 backend
4894 .latest_authorized_parallel_wal_commit_certificate(&cx)
4895 .wait()
4896 .expect("content mismatch is a non-authorizing record")
4897 .is_none(),
4898 "matching generation and commit marker must not authorize different frame bytes"
4899 );
4900
4901 assert_wal_corrupt(
4902 backend
4903 .reconcile_parallel_wal_commit(&cx, &certificate, 1, 1, true)
4904 .wait(),
4905 "in-doubt content-bound reconciliation mismatch",
4906 );
4907 assert_eq!(
4908 read_certificate_sidecar(&vfs, &cx),
4909 sidecar_before,
4910 "digest mismatch must be diagnosed before sidecar repair"
4911 );
4912 assert_eq!(
4913 backend.inner.frame_count(),
4914 1,
4915 "digest mismatch must preserve the live WAL for diagnosis and retry"
4916 );
4917 }
4918
4919 #[test]
4920 fn absent_commit_marker_repairs_certificate_and_partial_wal_tail() {
4921 let cx = test_cx();
4922 let vfs = MemoryVfs::new();
4923 let mut backend = make_path_refreshing_backend(&vfs, &cx);
4924 let certificate = sample_certificate(1, 1, vec![1]);
4925 backend
4926 .persist_parallel_wal_commit_certificate(&cx, &certificate, 1, 1, true)
4927 .expect("persist orphan certificate");
4928
4929 let (mut tail_writer, _) = vfs
4930 .open(
4931 &cx,
4932 Some(std::path::Path::new("test.db-wal")),
4933 VfsOpenFlags::READWRITE | VfsOpenFlags::WAL,
4934 )
4935 .expect("open WAL for partial-tail injection");
4936 let committed_size = tail_writer.file_size(&cx).expect("read committed WAL size");
4937 tail_writer
4938 .write(&cx, &[0xA5; 7], committed_size)
4939 .expect("inject a partial physical frame");
4940 assert!(
4941 tail_writer.file_size(&cx).expect("read extended WAL size") > committed_size,
4942 "fault fixture must extend the physical WAL"
4943 );
4944 tail_writer.close(&cx).expect("close partial-tail injector");
4945
4946 assert_eq!(
4947 backend
4948 .reconcile_parallel_wal_commit(&cx, &certificate, 1, 1, true)
4949 .wait()
4950 .expect("missing commit marker must be exactly repairable"),
4951 ParallelWalCommitReconciliation::NotCommitted
4952 );
4953 assert!(
4954 read_certificate_sidecar(&vfs, &cx).is_empty(),
4955 "matching orphan certificate must be removed after NotCommitted proof"
4956 );
4957 let (mut repaired_wal, _) = vfs
4958 .open(
4959 &cx,
4960 Some(std::path::Path::new("test.db-wal")),
4961 VfsOpenFlags::READONLY | VfsOpenFlags::WAL,
4962 )
4963 .expect("open repaired WAL");
4964 assert_eq!(
4965 repaired_wal.file_size(&cx).expect("read repaired WAL size"),
4966 committed_size,
4967 "NotCommitted reconciliation must truncate the physical partial tail"
4968 );
4969 repaired_wal.close(&cx).expect("close repaired WAL");
4970 }
4971
4972 #[test]
4973 fn durable_certificate_recovery_accepts_every_truncated_record_prefix() {
4974 let cx = test_cx();
4975 let vfs = MemoryVfs::new();
4976 let (mut backend, authorized) = make_authorized_certificate_backend(&vfs, &cx);
4977 let authorized_bytes = read_certificate_sidecar(&vfs, &cx);
4978 let orphan = sample_certificate(2, 2, vec![1]);
4979 let orphan_bytes = ParallelWalDurableCertificateRecord::new(
4980 backend.inner.inner().generation_identity(),
4981 2,
4982 2,
4983 orphan,
4984 )
4985 .expect("construct orphan record")
4986 .to_bytes();
4987
4988 for prefix_len in 1..orphan_bytes.len() {
4989 let mut sidecar = authorized_bytes.clone();
4990 sidecar.extend_from_slice(&orphan_bytes[..prefix_len]);
4991 replace_certificate_sidecar(&vfs, &cx, &sidecar);
4992 let recovered_result = backend
4993 .latest_authorized_parallel_wal_commit_certificate(&cx)
4994 .wait();
4995 assert!(
4996 recovered_result.is_ok(),
4997 "truncated certificate prefix of {prefix_len} bytes must recover: {recovered_result:?}"
4998 );
4999 let recovered = recovered_result
5000 .expect("truncated certificate recovery was asserted successful")
5001 .expect("authorized record must remain discoverable");
5002 assert_eq!(recovered, authorized, "failed at prefix {prefix_len}");
5003 }
5004 }
5005
5006 #[test]
5007 fn durable_certificate_append_repairs_the_accepted_torn_suffix() {
5008 let cx = test_cx();
5009 let vfs = MemoryVfs::new();
5010 let (mut backend, authorized) = make_authorized_certificate_backend(&vfs, &cx);
5011 let authorized_bytes = read_certificate_sidecar(&vfs, &cx);
5012 let orphan = sample_certificate(2, 2, vec![1]);
5013 let orphan_bytes = ParallelWalDurableCertificateRecord::new(
5014 backend.inner.inner().generation_identity(),
5015 2,
5016 2,
5017 orphan.clone(),
5018 )
5019 .expect("construct orphan record")
5020 .to_bytes();
5021 for prefix_len in 1..orphan_bytes.len() {
5022 let mut torn_sidecar = authorized_bytes.clone();
5023 torn_sidecar.extend_from_slice(&orphan_bytes[..prefix_len]);
5024 replace_certificate_sidecar(&vfs, &cx, &torn_sidecar);
5025
5026 assert_eq!(
5027 backend
5028 .latest_authorized_parallel_wal_commit_certificate(&cx)
5029 .wait()
5030 .expect("one torn suffix should recover")
5031 .expect("authorized predecessor remains visible"),
5032 authorized,
5033 "read recovery failed for prefix {prefix_len}"
5034 );
5035
5036 backend
5037 .persist_parallel_wal_commit_certificate(&cx, &orphan, 2, 2, true)
5038 .expect("next append repairs the torn suffix first");
5039 let repaired_sidecar = read_certificate_sidecar(&vfs, &cx);
5040 assert_eq!(
5041 repaired_sidecar.len(),
5042 authorized_bytes.len() + orphan_bytes.len(),
5043 "replacement record did not start at the prior complete boundary for prefix {prefix_len}"
5044 );
5045 assert_eq!(
5046 backend
5047 .latest_authorized_parallel_wal_commit_certificate(&cx)
5048 .wait()
5049 .expect("orphan lookback crosses the repaired boundary")
5050 .expect("authorized predecessor remains discoverable"),
5051 authorized,
5052 "orphan lookback failed after repairing prefix {prefix_len}"
5053 );
5054 }
5055
5056 let mut corrupt_record = orphan_bytes;
5057 let envelope_crc_offset =
5058 corrupt_record.len() - ParallelWalDurableCertificateRecord::LENGTH_FOOTER_SIZE - 4;
5059 corrupt_record[envelope_crc_offset] ^= 0x80;
5060 let mut corrupt_sidecar = authorized_bytes;
5061 corrupt_sidecar.extend_from_slice(&corrupt_record);
5062 replace_certificate_sidecar(&vfs, &cx, &corrupt_sidecar);
5063 assert_wal_corrupt(
5064 backend
5065 .persist_parallel_wal_commit_certificate(&cx, &orphan, 2, 2, true)
5066 .wait(),
5067 "append-time complete record corruption",
5068 );
5069 }
5070
5071 #[test]
5072 fn durable_certificate_recovery_rejects_complete_corruption_and_garbage() {
5073 let cx = test_cx();
5074 let vfs = MemoryVfs::new();
5075 let (mut backend, _) = make_authorized_certificate_backend(&vfs, &cx);
5076 let authorized_bytes = read_certificate_sidecar(&vfs, &cx);
5077 let orphan = sample_certificate(2, 2, vec![1]);
5078 let orphan_bytes = ParallelWalDurableCertificateRecord::new(
5079 backend.inner.inner().generation_identity(),
5080 2,
5081 2,
5082 orphan,
5083 )
5084 .expect("construct orphan record")
5085 .to_bytes();
5086
5087 let mut bad_crc = orphan_bytes.clone();
5088 let envelope_crc_offset =
5089 bad_crc.len() - ParallelWalDurableCertificateRecord::LENGTH_FOOTER_SIZE - 4;
5090 bad_crc[envelope_crc_offset] ^= 0x80;
5091 let mut sidecar = authorized_bytes.clone();
5092 sidecar.extend_from_slice(&bad_crc);
5093 replace_certificate_sidecar(&vfs, &cx, &sidecar);
5094 assert_wal_corrupt(
5095 backend
5096 .latest_authorized_parallel_wal_commit_certificate(&cx)
5097 .wait(),
5098 "complete record with bad CRC",
5099 );
5100
5101 let mut bad_version = orphan_bytes.clone();
5102 bad_version[8] ^= 0x01;
5103 let mut sidecar = authorized_bytes.clone();
5104 sidecar.extend_from_slice(&bad_version);
5105 replace_certificate_sidecar(&vfs, &cx, &sidecar);
5106 assert_wal_corrupt(
5107 backend
5108 .latest_authorized_parallel_wal_commit_certificate(&cx)
5109 .wait(),
5110 "complete record with bad version",
5111 );
5112
5113 let mut bad_magic = orphan_bytes.clone();
5114 bad_magic[0] ^= 0x01;
5115 let mut sidecar = authorized_bytes.clone();
5116 sidecar.extend_from_slice(&bad_magic);
5117 replace_certificate_sidecar(&vfs, &cx, &sidecar);
5118 assert_wal_corrupt(
5119 backend
5120 .latest_authorized_parallel_wal_commit_certificate(&cx)
5121 .wait(),
5122 "complete record with bad magic",
5123 );
5124
5125 let mut bad_footer = orphan_bytes;
5126 let footer_offset =
5127 bad_footer.len() - ParallelWalDurableCertificateRecord::LENGTH_FOOTER_SIZE;
5128 bad_footer[footer_offset] ^= 0x80;
5129 let mut sidecar = authorized_bytes;
5130 sidecar.extend_from_slice(&bad_footer);
5131 replace_certificate_sidecar(&vfs, &cx, &sidecar);
5132 assert_wal_corrupt(
5133 backend
5134 .latest_authorized_parallel_wal_commit_certificate(&cx)
5135 .wait(),
5136 "complete record with bad footer",
5137 );
5138
5139 let garbage_vfs = MemoryVfs::new();
5140 let mut garbage_backend = make_path_refreshing_backend(&garbage_vfs, &cx);
5141 replace_certificate_sidecar(&garbage_vfs, &cx, &[0xA5; 128]);
5142 assert_wal_corrupt(
5143 garbage_backend
5144 .latest_authorized_parallel_wal_commit_certificate(&cx)
5145 .wait(),
5146 "nonempty garbage sidecar",
5147 );
5148
5149 let mut fake_magic = vec![0_u8; MIN_DURABLE_CERTIFICATE_RECORD_SIZE];
5150 fake_magic[..PARALLEL_WAL_DURABLE_CERTIFICATE_MAGIC.len()]
5151 .copy_from_slice(&PARALLEL_WAL_DURABLE_CERTIFICATE_MAGIC);
5152 fake_magic[8..10].copy_from_slice(
5153 &fsqlite_wal::PARALLEL_WAL_DURABLE_CERTIFICATE_RECORD_VERSION.to_le_bytes(),
5154 );
5155 let fake_record_len = u32::try_from(fake_magic.len()).expect("fake record length fits u32");
5156 fake_magic[10..14].copy_from_slice(&fake_record_len.to_le_bytes());
5157 let fake_footer_offset =
5158 fake_magic.len() - ParallelWalDurableCertificateRecord::LENGTH_FOOTER_SIZE;
5159 fake_magic[fake_footer_offset..].copy_from_slice(&fake_record_len.to_le_bytes());
5160 replace_certificate_sidecar(&garbage_vfs, &cx, &fake_magic);
5161 assert_wal_corrupt(
5162 garbage_backend
5163 .latest_authorized_parallel_wal_commit_certificate(&cx)
5164 .wait(),
5165 "fake magic and length without a valid envelope",
5166 );
5167 }
5168
5169 #[test]
5170 fn durable_certificate_maximum_size_is_shared_by_writer_and_reader() {
5171 let cx = test_cx();
5172 let vfs = MemoryVfs::new();
5173 let mut backend = make_path_refreshing_backend(&vfs, &cx);
5174 let certificate = sample_certificate(1, 1, vec![1; usize::from(u16::MAX)]);
5175 let record = ParallelWalDurableCertificateRecord::new(
5176 backend.inner.inner().generation_identity(),
5177 1,
5178 1,
5179 certificate.clone(),
5180 )
5181 .expect("construct maximum-size record");
5182 assert_eq!(
5183 record.to_bytes().len(),
5184 PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE
5185 );
5186 backend
5187 .persist_parallel_wal_commit_certificate(&cx, &certificate, 1, 1, true)
5188 .expect("writer accepts maximum-size record");
5189 assert!(
5190 backend
5191 .latest_authorized_parallel_wal_commit_certificate(&cx)
5192 .wait()
5193 .expect("reader accepts maximum-size record")
5194 .is_none(),
5195 "record remains unauthorized until its WAL commit marker exists"
5196 );
5197 }
5198
5199 #[test]
5200 fn durable_certificate_orphan_lookback_allows_exact_boundary_plus_torn_tail() {
5201 let cx = test_cx();
5202 let vfs = MemoryVfs::new();
5203 let (mut backend, authorized) = make_authorized_certificate_backend(&vfs, &cx);
5204 let mut sidecar = read_certificate_sidecar(&vfs, &cx);
5205 for orphan_index in 0..MAX_ORPHAN_CERTIFICATE_LOOKBACK {
5212 let epoch = u64::try_from(orphan_index).expect("orphan index fits u64") + 2;
5213 let orphan = sample_certificate(epoch, epoch, vec![1]);
5214 sidecar.extend_from_slice(
5215 &ParallelWalDurableCertificateRecord::new(
5216 backend.inner.inner().generation_identity(),
5217 1,
5218 1,
5219 orphan,
5220 )
5221 .expect("construct bounded orphan")
5222 .to_bytes(),
5223 );
5224 }
5225 sidecar.push(0xA5);
5226 replace_certificate_sidecar(&vfs, &cx, &sidecar);
5227 assert_eq!(
5228 backend
5229 .latest_authorized_parallel_wal_commit_certificate(&cx)
5230 .wait()
5231 .expect("64 orphans plus one torn suffix remain within bound")
5232 .expect("authorized predecessor is found"),
5233 authorized
5234 );
5235
5236 sidecar.pop();
5237 let overflow_epoch =
5238 u64::try_from(MAX_ORPHAN_CERTIFICATE_LOOKBACK).expect("lookback fits u64") + 2;
5239 let overflow = sample_certificate(overflow_epoch, overflow_epoch, vec![1]);
5240 sidecar.extend_from_slice(
5241 &ParallelWalDurableCertificateRecord::new(
5242 backend.inner.inner().generation_identity(),
5243 1,
5244 1,
5245 overflow,
5246 )
5247 .expect("construct overflow orphan")
5248 .to_bytes(),
5249 );
5250 replace_certificate_sidecar(&vfs, &cx, &sidecar);
5251 assert_wal_corrupt(
5252 backend
5253 .latest_authorized_parallel_wal_commit_certificate(&cx)
5254 .wait(),
5255 "65 unauthorized records",
5256 );
5257
5258 let mut future_sidecar = read_certificate_sidecar(&vfs, &cx);
5262 future_sidecar.truncate(
5263 future_sidecar.len()
5264 - (MAX_ORPHAN_CERTIFICATE_LOOKBACK + 1)
5265 * ParallelWalDurableCertificateRecord::new(
5266 backend.inner.inner().generation_identity(),
5267 1,
5268 1,
5269 sample_certificate(2, 2, vec![1]),
5270 )
5271 .expect("sizing record")
5272 .to_bytes()
5273 .len(),
5274 );
5275 for future_index in 0..=MAX_ORPHAN_CERTIFICATE_LOOKBACK {
5276 let epoch = u64::try_from(future_index).expect("future index fits u64") + 2;
5277 let future = sample_certificate(epoch, epoch, vec![1]);
5278 future_sidecar.extend_from_slice(
5279 &ParallelWalDurableCertificateRecord::new(
5280 backend.inner.inner().generation_identity(),
5281 2,
5282 2,
5283 future,
5284 )
5285 .expect("construct future record")
5286 .to_bytes(),
5287 );
5288 }
5289 future_sidecar.push(0xA5);
5290 replace_certificate_sidecar(&vfs, &cx, &future_sidecar);
5291 assert_eq!(
5292 backend
5293 .latest_authorized_parallel_wal_commit_certificate(&cx)
5294 .wait()
5295 .expect("future-boundary records are budget-exempt")
5296 .expect("authorized predecessor is found beneath futures"),
5297 authorized
5298 );
5299 }
5300
5301 #[test]
5302 fn certificate_and_handoff_fences_request_full_durability() {
5303 let cx = test_cx();
5304 let vfs = CheckpointHandoffFaultVfs::new();
5305 let (mut backend, certificate, _) = make_checkpoint_handoff_fault_backend(&vfs, &cx);
5306
5307 assert_eq!(
5308 vfs.take_sync_observations(),
5309 vec![CertificateSyncObservation::Durable(
5310 PathBuf::from(CERTIFICATE_PATH),
5311 SyncKind::FullDurable,
5312 )],
5313 "certificate append must use the strongest durability intent"
5314 );
5315
5316 assert_eq!(
5317 backend
5318 .reconcile_parallel_wal_commit(&cx, &certificate, 1, 1, true)
5319 .wait()
5320 .expect("reconcile committed certificate"),
5321 ParallelWalCommitReconciliation::Authorized
5322 );
5323 assert_eq!(
5324 vfs.take_sync_observations(),
5325 vec![CertificateSyncObservation::Durable(
5326 PathBuf::from(CERTIFICATE_PATH),
5327 SyncKind::FullDurable,
5328 )],
5329 "certificate reconciliation must preserve full durability intent"
5330 );
5331
5332 let record = backend
5333 .latest_authorized_durable_certificate_record(&cx)
5334 .wait()
5335 .expect("read authorized certificate record")
5336 .expect("authorized certificate record must exist");
5337 backend
5338 .persist_checkpoint_certificate_handoff(&cx, &record)
5339 .wait()
5340 .expect("persist checkpoint certificate handoff");
5341 assert_eq!(
5342 vfs.take_sync_observations(),
5343 vec![CertificateSyncObservation::Durable(
5344 PathBuf::from(CHECKPOINT_HANDOFF_PATH),
5345 SyncKind::FullDurable,
5346 )],
5347 "checkpoint handoff must use the strongest durability intent"
5348 );
5349 }
5350
5351 #[test]
5352 fn checkpoint_handoff_write_failure_preserves_authoritative_wal_generation() {
5353 let cx = test_cx();
5354 let vfs = CheckpointHandoffFaultVfs::new();
5355 let (mut backend, certificate, committed_page) =
5356 make_checkpoint_handoff_fault_backend(&vfs, &cx);
5357 let before = capture_authoritative_wal(&backend, &vfs, &cx, certificate, committed_page);
5358 vfs.fail_next_handoff_write();
5359
5360 let mut checkpoint_writer = MockCheckpointPageWriter;
5361 let error = backend
5362 .checkpoint(
5363 &cx,
5364 CheckpointMode::Truncate,
5365 &mut checkpoint_writer,
5366 0,
5367 None,
5368 )
5369 .expect_err("checkpoint must fail before reset when the handoff write fails");
5370 assert!(
5371 error
5372 .to_string()
5373 .contains("injected checkpoint handoff write failure"),
5374 "unexpected handoff write error: {error}"
5375 );
5376 assert_authoritative_wal_unchanged(&mut backend, &vfs, &cx, &before);
5377 }
5378
5379 #[test]
5380 fn checkpoint_handoff_durable_sync_failure_preserves_authoritative_wal_generation() {
5381 let cx = test_cx();
5382 let vfs = CheckpointHandoffFaultVfs::new();
5383 let (mut backend, certificate, committed_page) =
5384 make_checkpoint_handoff_fault_backend(&vfs, &cx);
5385 let before = capture_authoritative_wal(&backend, &vfs, &cx, certificate, committed_page);
5386 vfs.fail_next_handoff_sync();
5387
5388 let mut checkpoint_writer = MockCheckpointPageWriter;
5389 let error = backend
5390 .checkpoint(
5391 &cx,
5392 CheckpointMode::Truncate,
5393 &mut checkpoint_writer,
5394 0,
5395 None,
5396 )
5397 .expect_err("checkpoint must fail before reset when the handoff sync fails");
5398 assert!(
5399 error
5400 .to_string()
5401 .contains("injected checkpoint handoff durable-sync failure"),
5402 "unexpected handoff durable-sync error: {error}"
5403 );
5404 assert_authoritative_wal_unchanged(&mut backend, &vfs, &cx, &before);
5405 }
5406
5407 #[test]
5408 fn dropping_pending_checkpoint_handoff_write_preserves_authoritative_wal_generation() {
5409 let cx = test_cx();
5410 let vfs = CheckpointHandoffFaultVfs::new();
5411 let (mut backend, certificate, committed_page) =
5412 make_checkpoint_handoff_fault_backend(&vfs, &cx);
5413 let before = capture_authoritative_wal(&backend, &vfs, &cx, certificate, committed_page);
5414 vfs.pend_next_handoff_write();
5415
5416 let mut checkpoint_writer = MockCheckpointPageWriter;
5417 let reached_pending_handoff = {
5418 let mut checkpoint = backend.checkpoint(
5419 &cx,
5420 CheckpointMode::Truncate,
5421 &mut checkpoint_writer,
5422 0,
5423 None,
5424 );
5425 let mut task_cx = std::task::Context::from_waker(std::task::Waker::noop());
5426 matches!(
5427 std::future::Future::poll(checkpoint.as_mut(), &mut task_cx),
5428 std::task::Poll::Pending
5429 )
5430 };
5431 assert!(
5432 reached_pending_handoff,
5433 "checkpoint should remain pending inside the injected handoff write"
5434 );
5435 assert_authoritative_wal_unchanged(&mut backend, &vfs, &cx, &before);
5436 }
5437
5438 #[test]
5439 fn two_backend_instances_continue_authorized_certificate_clocks() {
5440 let cx = test_cx();
5441 let vfs = MemoryVfs::new();
5442 let wal = WalFile::create(&cx, open_wal_file(&vfs, &cx), PAGE_SIZE, 0, test_salts())
5443 .expect("create shared WAL");
5444 let mut first_backend = PathRefreshingWalBackend::new(
5445 vfs.clone(),
5446 std::path::Path::new("test.db"),
5447 std::path::Path::new("test.db-wal"),
5448 PAGE_SIZE,
5449 wal,
5450 true,
5451 #[cfg(all(feature = "native", any(unix, windows)))]
5452 None,
5453 );
5454 let request =
5455 |batch_id, wal_frame_payload_digest| fsqlite_wal::ParallelWalDurabilityRequest {
5456 trace_id: batch_id,
5457 scenario_id: "two-instance-continuity".to_owned(),
5458 certificate_epoch: 0,
5459 durable_segment_epoch: 0,
5460 batch_size: 1,
5461 batch_ids: vec![batch_id],
5462 lane_record_counts: vec![1],
5463 db_size_pages: 1,
5464 page_set_size: 1,
5465 control_mode: fsqlite_wal::ParallelWalOperatingMode::Auto,
5466 fallback_reason: None,
5467 checkpoint_active: false,
5468 wal_frame_payload_digest,
5469 };
5470
5471 let first_combiner = fsqlite_wal::ParallelWalDurabilityCombiner::default();
5472 let first_page = sample_page(0x51);
5473 let first_receipt = first_combiner
5474 .certify_and_publish(
5475 request(1, test_frame_payload_digest(1, &first_page, 1)),
5476 |certificate| {
5477 first_backend
5478 .persist_parallel_wal_commit_certificate(&cx, certificate, 1, 1, true)
5479 .wait()
5480 .and_then(|()| first_backend.append_frame(&cx, 1, &first_page, 1).wait())
5481 .and_then(|()| first_backend.sync(&cx))
5482 .map_err(|error| error.to_string())
5483 },
5484 )
5485 .expect("first backend publishes certificate");
5486
5487 let second_wal =
5488 WalFile::open(&cx, open_wal_file(&vfs, &cx)).expect("second backend opens shared WAL");
5489 let mut second_backend = PathRefreshingWalBackend::new(
5490 vfs.clone(),
5491 std::path::Path::new("test.db"),
5492 std::path::Path::new("test.db-wal"),
5493 PAGE_SIZE,
5494 second_wal,
5495 true,
5496 #[cfg(all(feature = "native", any(unix, windows)))]
5497 None,
5498 );
5499
5500 let orphan_combiner = fsqlite_wal::ParallelWalDurabilityCombiner::default();
5504 orphan_combiner
5505 .reconcile_authorized_seed(&first_receipt.certificate)
5506 .expect("seed orphan-producing process");
5507 let orphan_receipt = orphan_combiner
5508 .certify_and_publish(
5509 request(99, test_frame_payload_digest(1, &sample_page(0x52), 1)),
5510 |_| Ok(()),
5511 )
5512 .expect("construct deterministic orphan certificate");
5513 second_backend
5514 .persist_parallel_wal_commit_certificate(&cx, &orphan_receipt.certificate, 2, 2, true)
5515 .expect("persist well-formed orphan certificate tail");
5516 let authorized_seed = second_backend
5517 .latest_authorized_parallel_wal_commit_certificate(&cx)
5518 .expect("second backend performs bounded orphan lookback")
5519 .expect("preceding first certificate remains authorized");
5520 assert_eq!(authorized_seed, first_receipt.certificate);
5521
5522 let second_combiner = fsqlite_wal::ParallelWalDurabilityCombiner::default();
5523 second_combiner
5524 .reconcile_authorized_seed(&authorized_seed)
5525 .expect("seed second process-local combiner");
5526 let second_page = sample_page(0x52);
5527 let second_receipt = second_combiner
5528 .certify_and_publish(
5529 request(2, test_frame_payload_digest(1, &second_page, 1)),
5530 |certificate| {
5531 second_backend
5532 .persist_parallel_wal_commit_certificate(&cx, certificate, 2, 2, true)
5533 .wait()
5534 .and_then(|()| second_backend.append_frame(&cx, 1, &second_page, 1).wait())
5535 .and_then(|()| second_backend.sync(&cx))
5536 .map_err(|error| error.to_string())
5537 },
5538 )
5539 .expect("second backend publishes certificate");
5540
5541 assert_eq!(
5542 second_receipt.certificate.commit_seq_lo.get(),
5543 first_receipt.certificate.commit_seq_hi.get() + 1
5544 );
5545 assert_eq!(
5546 second_receipt.certificate.certificate_epoch,
5547 first_receipt.certificate.certificate_epoch + 1
5548 );
5549 assert_eq!(
5550 second_receipt.certificate, orphan_receipt.certificate,
5551 "continuation may reuse an orphan identity but must not overlap any authorized certificate"
5552 );
5553 let latest = second_backend
5554 .latest_authorized_parallel_wal_commit_certificate(&cx)
5555 .expect("read second bounded authorized tail")
5556 .expect("second certificate is authorized");
5557 assert_eq!(latest, second_receipt.certificate);
5558
5559 let generation_before_checkpoint = second_backend.inner.inner().generation_identity();
5560 let mut checkpoint_writer = MockCheckpointPageWriter;
5561 let checkpoint = second_backend
5562 .checkpoint(
5563 &cx,
5564 CheckpointMode::Truncate,
5565 &mut checkpoint_writer,
5566 0,
5567 None,
5568 )
5569 .expect("truncate checkpoint records certificate clock handoff");
5570 assert!(checkpoint.wal_was_reset);
5571 assert_ne!(
5572 second_backend.inner.inner().generation_identity(),
5573 generation_before_checkpoint
5574 );
5575 let checkpoint_seed = second_backend
5576 .latest_authorized_parallel_wal_commit_certificate(&cx)
5577 .expect("read checkpoint certificate clock handoff")
5578 .expect("reset generation retains the last consumed certificate clock");
5579 assert_eq!(checkpoint_seed, second_receipt.certificate);
5580 second_backend
5581 .begin_transaction(&cx)
5582 .expect("pin reset-generation reader snapshot");
5583 let reset_pinned = second_backend
5584 .pinned_read_snapshot()
5585 .expect("reset-generation reader snapshot");
5586 assert_eq!(
5587 reset_pinned.generation,
5588 second_backend.inner.inner().generation_identity(),
5589 "reader snapshot must bind the reset WAL generation"
5590 );
5591 assert_eq!(
5592 reset_pinned.last_commit_frame, None,
5593 "truncate checkpoint leaves no current-generation commit marker"
5594 );
5595 assert_eq!(
5596 second_backend
5597 .pinned_logical_read_snapshot(&cx)
5598 .expect("inspect reset-generation reader horizon"),
5599 None,
5600 "an earlier-generation checkpoint handoff is a clock seed, never reader visibility"
5601 );
5602
5603 let post_checkpoint_combiner = fsqlite_wal::ParallelWalDurabilityCombiner::default();
5604 post_checkpoint_combiner
5605 .reconcile_authorized_seed(&checkpoint_seed)
5606 .expect("seed fresh post-checkpoint combiner");
5607 let post_checkpoint_page = sample_page(0x53);
5608 let post_checkpoint_receipt = post_checkpoint_combiner
5609 .certify_and_publish(
5610 request(3, test_frame_payload_digest(1, &post_checkpoint_page, 1)),
5611 |certificate| {
5612 second_backend
5613 .persist_parallel_wal_commit_certificate(&cx, certificate, 1, 1, true)
5614 .wait()
5615 .and_then(|()| {
5616 second_backend
5617 .append_frame(&cx, 1, &post_checkpoint_page, 1)
5618 .wait()
5619 })
5620 .and_then(|()| second_backend.sync(&cx))
5621 .map_err(|error| error.to_string())
5622 },
5623 )
5624 .expect("publish first certificate in reset WAL generation");
5625 assert_eq!(
5626 post_checkpoint_receipt.certificate.commit_seq_lo.get(),
5627 second_receipt.certificate.commit_seq_hi.get() + 1
5628 );
5629 assert_eq!(
5630 post_checkpoint_receipt.certificate.certificate_epoch,
5631 second_receipt.certificate.certificate_epoch + 1
5632 );
5633 assert_eq!(
5634 second_backend
5635 .latest_authorized_parallel_wal_commit_certificate(&cx)
5636 .expect("read post-checkpoint current-generation certificate")
5637 .expect("post-checkpoint certificate is authorized"),
5638 post_checkpoint_receipt.certificate
5639 );
5640 second_backend
5641 .begin_transaction(&cx)
5642 .expect("pin post-checkpoint reader snapshot");
5643 let pinned = second_backend
5644 .pinned_read_snapshot()
5645 .expect("post-checkpoint reader snapshot");
5646 let logical = second_backend
5647 .pinned_logical_read_snapshot(&cx)
5648 .expect("inspect post-checkpoint reader horizon")
5649 .expect("current-generation certificate exposes a reader horizon");
5650 assert_eq!(logical.generation, pinned.generation);
5651 assert_eq!(logical.last_commit_frame, pinned.last_commit_frame);
5652 assert_eq!(
5653 logical.visible_commit_seq,
5654 post_checkpoint_receipt.certificate.commit_seq_hi
5655 );
5656 }
5657
5658 #[test]
5659 fn pinned_logical_reader_horizon_counts_physical_tail_after_current_certificate() {
5660 let cx = test_cx();
5661 let vfs = MemoryVfs::new();
5662 let (mut backend, certificate) = make_authorized_certificate_backend(&vfs, &cx);
5663
5664 backend
5665 .begin_transaction(&cx)
5666 .expect("pin certificate reader snapshot");
5667 let initial_pinned = backend
5668 .pinned_read_snapshot()
5669 .expect("initial reader snapshot");
5670 let initial_logical = backend
5671 .pinned_logical_read_snapshot(&cx)
5672 .expect("inspect certificate reader horizon")
5673 .expect("current certificate exposes reader horizon");
5674 assert_eq!(initial_logical.generation, initial_pinned.generation);
5675 assert_eq!(
5676 initial_logical.last_commit_frame,
5677 initial_pinned.last_commit_frame
5678 );
5679 assert_eq!(
5680 initial_logical.visible_commit_seq, certificate.commit_seq_hi,
5681 "certificate horizon is exact when no later physical commit exists"
5682 );
5683
5684 let tail_page = sample_page(0x45);
5685 backend
5686 .append_frame(&cx, 2, &tail_page, 2)
5687 .expect("append later ordinary commit marker");
5688 backend
5689 .sync(&cx)
5690 .expect("sync later ordinary commit marker");
5691 backend
5692 .begin_transaction(&cx)
5693 .expect("repin reader after ordinary tail commit");
5694 let pinned = backend
5695 .pinned_read_snapshot()
5696 .expect("reader snapshot includes ordinary tail commit");
5697 let logical = backend
5698 .pinned_logical_read_snapshot(&cx)
5699 .expect("inspect reader horizon with ordinary tail")
5700 .expect("current certificate remains reader-authoritative");
5701 assert_eq!(logical.generation, pinned.generation);
5702 assert_eq!(logical.last_commit_frame, pinned.last_commit_frame);
5703 assert_eq!(
5704 logical.visible_commit_seq.get(),
5705 certificate.commit_seq_hi.get() + 1
5706 );
5707 }
5708
5709 fn open_wal_file(vfs: &MemoryVfs, cx: &Cx) -> <MemoryVfs as Vfs>::File {
5710 let flags = VfsOpenFlags::READWRITE | VfsOpenFlags::CREATE | VfsOpenFlags::WAL;
5711 let (file, _) = vfs
5712 .open(cx, Some(std::path::Path::new("test.db-wal")), flags)
5713 .expect("open WAL file");
5714 file
5715 }
5716
5717 fn make_adapter(vfs: &MemoryVfs, cx: &Cx) -> WalBackendAdapter<<MemoryVfs as Vfs>::File> {
5718 let file = open_wal_file(vfs, cx);
5719 let wal = WalFile::create(cx, file, PAGE_SIZE, 0, test_salts()).expect("create WAL");
5720 WalBackendAdapter::new(wal)
5721 }
5722
5723 fn make_fault_adapter(
5725 vfs: &CheckpointHandoffFaultVfs,
5726 cx: &Cx,
5727 ) -> WalBackendAdapter<<CheckpointHandoffFaultVfs as Vfs>::File> {
5728 let flags = VfsOpenFlags::READWRITE | VfsOpenFlags::CREATE | VfsOpenFlags::WAL;
5729 let (file, _) = vfs
5730 .open(cx, Some(std::path::Path::new("test.db-wal")), flags)
5731 .expect("open fault WAL file");
5732 let wal = WalFile::create(cx, file, PAGE_SIZE, 0, test_salts()).expect("create fault WAL");
5733 WalBackendAdapter::new(wal)
5734 }
5735
5736 #[test]
5739 fn test_adapter_append_and_frame_count() {
5740 let cx = test_cx();
5741 let vfs = MemoryVfs::new();
5742 let mut adapter = make_adapter(&vfs, &cx);
5743
5744 assert_eq!(adapter.frame_count(), 0);
5745
5746 let page = sample_page(0x42);
5747 adapter
5748 .append_frame(&cx, 1, &page, 0)
5749 .expect("append frame");
5750 assert_eq!(adapter.frame_count(), 1);
5751
5752 adapter
5753 .append_frame(&cx, 2, &sample_page(0x43), 2)
5754 .expect("append commit frame");
5755 assert_eq!(adapter.frame_count(), 2);
5756 }
5757
5758 #[test]
5759 fn test_adapter_read_page_found() {
5760 let cx = test_cx();
5761 let vfs = MemoryVfs::new();
5762 let mut adapter = make_adapter(&vfs, &cx);
5763
5764 let page1 = sample_page(0x10);
5765 let page2 = sample_page(0x20);
5766 adapter.append_frame(&cx, 1, &page1, 0).expect("append");
5767 adapter
5768 .append_frame(&cx, 2, &page2, 2)
5769 .expect("append commit");
5770
5771 assert_eq!(
5774 adapter.read_page(&cx, 1).expect("read staged page 1"),
5775 None,
5776 "staged frames must stay invisible before publication"
5777 );
5778 adapter.sync(&cx).expect("publish staged frames");
5779
5780 let result = adapter.read_page(&cx, 1).expect("read page 1");
5781 assert_eq!(result, Some(page1));
5782
5783 let result = adapter.read_page(&cx, 2).expect("read page 2");
5784 assert_eq!(result, Some(page2));
5785 }
5786
5787 #[test]
5788 fn test_adapter_read_page_not_found() {
5789 let cx = test_cx();
5790 let vfs = MemoryVfs::new();
5791 let mut adapter = make_adapter(&vfs, &cx);
5792
5793 adapter
5794 .append_frame(&cx, 1, &sample_page(0x10), 1)
5795 .expect("append");
5796
5797 let result = adapter.read_page(&cx, 99).expect("read missing page");
5798 assert_eq!(result, None);
5799 }
5800
5801 #[test]
5802 fn test_adapter_read_page_returns_latest_version() {
5803 let cx = test_cx();
5804 let vfs = MemoryVfs::new();
5805 let mut adapter = make_adapter(&vfs, &cx);
5806
5807 let old_data = sample_page(0xAA);
5808 let new_data = sample_page(0xBB);
5809
5810 adapter
5812 .append_frame(&cx, 5, &old_data, 0)
5813 .expect("append old");
5814 adapter
5815 .append_frame(&cx, 5, &new_data, 1)
5816 .expect("append new (commit)");
5817
5818 adapter.sync(&cx).expect("publish staged frames");
5820
5821 let result = adapter.read_page(&cx, 5).expect("read page 5");
5822 assert_eq!(
5823 result,
5824 Some(new_data),
5825 "adapter should return the latest WAL version"
5826 );
5827 }
5828
5829 #[test]
5830 fn test_adapter_refreshes_cross_handle_visibility_and_append_position() {
5831 let cx = test_cx();
5832 let vfs = MemoryVfs::new();
5833
5834 let file1 = open_wal_file(&vfs, &cx);
5835 let wal1 = WalFile::create(&cx, file1, PAGE_SIZE, 0, test_salts()).expect("create WAL");
5836 let mut adapter1 = WalBackendAdapter::new(wal1);
5837
5838 let file2 = open_wal_file(&vfs, &cx);
5839 let wal2 = WalFile::open(&cx, file2).expect("open WAL");
5840 let mut adapter2 = WalBackendAdapter::new(wal2);
5841
5842 let page1 = sample_page(0x11);
5843 adapter1
5844 .append_frame(&cx, 1, &page1, 1)
5845 .expect("adapter1 append commit");
5846 adapter1.sync(&cx).expect("adapter1 sync");
5847 adapter2
5848 .begin_transaction(&cx)
5849 .expect("adapter2 begin transaction");
5850 assert_eq!(
5851 adapter2.read_page(&cx, 1).expect("adapter2 read page1"),
5852 Some(page1.clone()),
5853 "adapter2 should observe adapter1 commit at transaction begin"
5854 );
5855
5856 let page2 = sample_page(0x22);
5857 adapter2
5858 .append_frame(&cx, 2, &page2, 2)
5859 .expect("adapter2 append commit");
5860 adapter2.sync(&cx).expect("adapter2 sync");
5861 adapter1
5862 .begin_transaction(&cx)
5863 .expect("adapter1 begin transaction");
5864 assert_eq!(
5865 adapter1.read_page(&cx, 2).expect("adapter1 read page2"),
5866 Some(page2.clone()),
5867 "adapter1 should observe adapter2 commit at transaction begin"
5868 );
5869
5870 assert_eq!(
5872 adapter1.frame_count(),
5873 2,
5874 "shared WAL should contain both commit frames"
5875 );
5876 assert_eq!(
5877 adapter2.frame_count(),
5878 2,
5879 "shared WAL should contain both commit frames"
5880 );
5881 }
5882
5883 #[test]
5884 fn test_path_refresh_rejects_replacement_wal_page_size_mismatch() {
5885 let cx = test_cx();
5886 let vfs = MemoryVfs::new();
5887 let wal_path = std::path::Path::new("test.db-wal");
5888
5889 let file = open_wal_file(&vfs, &cx);
5890 let wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create WAL");
5891 let mut backend = PathRefreshingWalBackend::new(
5892 vfs.clone(),
5893 std::path::Path::new("test.db"),
5894 wal_path,
5895 PAGE_SIZE,
5896 wal,
5897 true,
5898 #[cfg(all(feature = "native", any(unix, windows)))]
5899 None,
5900 );
5901
5902 backend
5903 .append_frame(&cx, 1, &sample_page(0x31), 1)
5904 .expect("append through live backend");
5905 backend.sync(&cx).expect("sync live backend");
5906
5907 vfs.delete(&cx, wal_path, false)
5908 .expect("remove path-visible WAL");
5909 let flags = VfsOpenFlags::READWRITE | VfsOpenFlags::CREATE | VfsOpenFlags::WAL;
5910 let (replacement_file, _) = vfs
5911 .open(&cx, Some(wal_path), flags)
5912 .expect("open replacement WAL path");
5913 let replacement_page_size = PAGE_SIZE
5914 .checked_mul(2)
5915 .expect("test replacement page size fits u32");
5916 let replacement_wal = WalFile::create(
5917 &cx,
5918 replacement_file,
5919 replacement_page_size,
5920 0,
5921 test_salts(),
5922 )
5923 .expect("create mismatched replacement WAL");
5924 replacement_wal.close(&cx).expect("close replacement WAL");
5925
5926 let err = backend
5927 .begin_transaction(&cx)
5928 .expect_err("path refresh should reject mismatched WAL page size");
5929 assert!(
5930 matches!(
5931 err,
5932 FrankenError::WalCorrupt { ref detail }
5933 if detail.contains("does not match database page size")
5934 && detail.contains("during path refresh")
5935 ),
5936 "unexpected error: {err:?}"
5937 );
5938 }
5939
5940 #[test]
5941 fn test_generation_change_allows_identical_full_page_baseline() {
5942 let cx = test_cx();
5943 let vfs = MemoryVfs::new();
5944 let (mut backend, snapshot, page_two) = make_generation_transition_backend(&vfs, &cx);
5945 let baseline = TransactionConflictPageBaseline {
5946 page_number: 2,
5947 page_hash: *blake3::hash(&page_two).as_bytes(),
5948 };
5949
5950 let conflicts = backend
5951 .conflicting_pages_since_snapshot(&cx, snapshot, &[2], &[baseline])
5952 .expect("validate checkpoint-only generation transition");
5953 assert!(
5954 conflicts.is_empty(),
5955 "byte-identical checkpoint-only reset must not create a false conflict"
5956 );
5957 }
5958
5959 #[test]
5966 fn test_generation_change_cached_verification_fd_reads_live_main_db_content() {
5967 let cx = test_cx();
5968 let vfs = MemoryVfs::new();
5969 let (mut backend, snapshot, page_two) = make_generation_transition_backend(&vfs, &cx);
5970 let baseline = TransactionConflictPageBaseline {
5971 page_number: 2,
5972 page_hash: *blake3::hash(&page_two).as_bytes(),
5973 };
5974
5975 let first = backend
5978 .conflicting_pages_since_snapshot(&cx, snapshot, &[2], &[baseline])
5979 .expect("first (cache-populating) generation-change verification");
5980 assert!(
5981 first.is_empty(),
5982 "identical baseline must not conflict on the first check"
5983 );
5984
5985 write_main_db_pages(
5989 &vfs,
5990 &cx,
5991 &[
5992 sqlite_page_one(u16::try_from(PAGE_SIZE).expect("page size fits u16")),
5993 sample_page(0x33),
5994 ],
5995 );
5996
5997 let second = backend
5998 .conflicting_pages_since_snapshot(&cx, snapshot, &[2], &[baseline])
5999 .expect("second generation-change verification reuses the cached fd");
6000 assert_eq!(
6001 second,
6002 vec![2],
6003 "cached verification fd must read the live changed page, not stale cached bytes"
6004 );
6005 }
6006
6007 #[test]
6008 fn test_generation_change_rejects_changed_candidate_page() {
6009 let cx = test_cx();
6010 let vfs = MemoryVfs::new();
6011 let (mut backend, snapshot, page_two) = make_generation_transition_backend(&vfs, &cx);
6012 let changed_page_two = sample_page(0x33);
6013 write_main_db_pages(
6014 &vfs,
6015 &cx,
6016 &[
6017 sqlite_page_one(u16::try_from(PAGE_SIZE).expect("page size fits u16")),
6018 changed_page_two,
6019 ],
6020 );
6021 let baseline = TransactionConflictPageBaseline {
6022 page_number: 2,
6023 page_hash: *blake3::hash(&page_two).as_bytes(),
6024 };
6025
6026 let conflicts = backend
6027 .conflicting_pages_since_snapshot(&cx, snapshot, &[2], &[baseline])
6028 .expect("validate changed page across generation transition");
6029 assert_eq!(conflicts, vec![2]);
6030 }
6031
6032 #[test]
6033 fn test_generation_change_rejects_changed_candidate_from_replacement_wal() {
6034 let cx = test_cx();
6035 let vfs = MemoryVfs::new();
6036 let (mut backend, snapshot, page_two) = make_generation_transition_backend(&vfs, &cx);
6037 append_replacement_wal_page(&vfs, &cx, 2, &sample_page(0x44), 2);
6038 let baseline = TransactionConflictPageBaseline {
6039 page_number: 2,
6040 page_hash: *blake3::hash(&page_two).as_bytes(),
6041 };
6042
6043 let conflicts = backend
6044 .conflicting_pages_since_snapshot(&cx, snapshot, &[2], &[baseline])
6045 .expect("replacement WAL page must take precedence over identical main page");
6046 assert_eq!(conflicts, vec![2]);
6047 }
6048
6049 #[test]
6050 fn test_generation_change_rejects_missing_baseline() {
6051 let cx = test_cx();
6052 let vfs = MemoryVfs::new();
6053 let (mut backend, snapshot, _) = make_generation_transition_backend(&vfs, &cx);
6054
6055 let conflicts = backend
6056 .conflicting_pages_since_snapshot(&cx, snapshot, &[2], &[])
6057 .expect("missing baseline must fail closed");
6058 assert_eq!(conflicts, vec![2]);
6059 }
6060
6061 #[test]
6062 fn test_generation_change_rejects_conflicting_duplicate_baselines() {
6063 let cx = test_cx();
6064 let vfs = MemoryVfs::new();
6065 let (mut backend, snapshot, page_two) = make_generation_transition_backend(&vfs, &cx);
6066 let baselines = [
6067 TransactionConflictPageBaseline {
6068 page_number: 2,
6069 page_hash: *blake3::hash(&page_two).as_bytes(),
6070 },
6071 TransactionConflictPageBaseline {
6072 page_number: 2,
6073 page_hash: *blake3::hash(&sample_page(0x55)).as_bytes(),
6074 },
6075 ];
6076
6077 let conflicts = backend
6078 .conflicting_pages_since_snapshot(&cx, snapshot, &[2], &baselines)
6079 .expect("conflicting duplicate baselines must fail closed");
6080 assert_eq!(conflicts, vec![2]);
6081 }
6082
6083 #[test]
6084 fn test_generation_change_rejects_short_candidate_page() {
6085 let cx = test_cx();
6086 let vfs = MemoryVfs::new();
6087 let (mut backend, snapshot, page_two) = make_generation_transition_backend(&vfs, &cx);
6088 write_main_db_pages(
6089 &vfs,
6090 &cx,
6091 &[sqlite_page_one(
6092 u16::try_from(PAGE_SIZE).expect("page size fits u16"),
6093 )],
6094 );
6095 let baseline = TransactionConflictPageBaseline {
6096 page_number: 2,
6097 page_hash: *blake3::hash(&page_two).as_bytes(),
6098 };
6099
6100 let conflicts = backend
6101 .conflicting_pages_since_snapshot(&cx, snapshot, &[2], &[baseline])
6102 .expect("short page must fail closed");
6103 assert_eq!(conflicts, vec![2]);
6104 }
6105
6106 #[test]
6107 fn test_generation_change_rejects_database_page_size_change() {
6108 let cx = test_cx();
6109 let vfs = MemoryVfs::new();
6110 let (mut backend, snapshot, page_two) = make_generation_transition_backend(&vfs, &cx);
6111 write_main_db_pages(&vfs, &cx, &[sqlite_page_one(8192), page_two.clone()]);
6112 let baseline = TransactionConflictPageBaseline {
6113 page_number: 2,
6114 page_hash: *blake3::hash(&page_two).as_bytes(),
6115 };
6116
6117 let conflicts = backend
6118 .conflicting_pages_since_snapshot(&cx, snapshot, &[2], &[baseline])
6119 .expect("page-size change must fail closed");
6120 assert_eq!(conflicts, vec![2]);
6121 }
6122
6123 #[test]
6124 fn test_generation_change_decodes_64k_database_header_sentinel() {
6125 assert_eq!(
6126 sqlite_database_header_page_size(&sqlite_page_one(1)),
6127 Some(65_536)
6128 );
6129 }
6130
6131 #[test]
6132 fn test_adapter_batch_append_checksum_chain_matches_single_append() {
6133 let cx = test_cx();
6134 let vfs_single = MemoryVfs::new();
6135 let vfs_batch = MemoryVfs::new();
6136
6137 let mut adapter_single = make_adapter(&vfs_single, &cx);
6138 let mut adapter_batch = make_adapter(&vfs_batch, &cx);
6139
6140 let pages: Vec<Vec<u8>> = (0..4u8).map(sample_page).collect();
6141 let commit_sizes = [0_u32, 0, 0, 4];
6142
6143 for (index, page) in pages.iter().enumerate() {
6144 adapter_single
6145 .append_frame(
6146 &cx,
6147 u32::try_from(index + 1).expect("page number fits u32"),
6148 page,
6149 commit_sizes[index],
6150 )
6151 .expect("single append");
6152 }
6153
6154 let batch_frames: Vec<_> = pages
6155 .iter()
6156 .enumerate()
6157 .map(|(index, page)| WalFrameRef {
6158 page_number: u32::try_from(index + 1).expect("page number fits u32"),
6159 page_data: page,
6160 db_size_if_commit: commit_sizes[index],
6161 })
6162 .collect();
6163 adapter_batch
6164 .append_frames(&cx, &batch_frames)
6165 .expect("batch append");
6166
6167 assert_eq!(
6168 adapter_single.frame_count(),
6169 adapter_batch.frame_count(),
6170 "batch adapter append must preserve frame count"
6171 );
6172 assert_eq!(
6173 adapter_single.wal.running_checksum(),
6174 adapter_batch.wal.running_checksum(),
6175 "batch adapter append must preserve checksum chain"
6176 );
6177
6178 for frame_index in 0..pages.len() {
6179 let (single_header, single_data) = adapter_single
6180 .wal
6181 .read_frame(&cx, frame_index)
6182 .expect("read single frame");
6183 let (batch_header, batch_data) = adapter_batch
6184 .wal
6185 .read_frame(&cx, frame_index)
6186 .expect("read batch frame");
6187 assert_eq!(
6188 single_header, batch_header,
6189 "frame header {frame_index} must match"
6190 );
6191 assert_eq!(
6192 single_data, batch_data,
6193 "frame payload {frame_index} must match"
6194 );
6195 }
6196 }
6197
6198 #[test]
6199 fn test_adapter_prepared_batch_append_checksum_chain_matches_single_append() {
6200 let cx = test_cx();
6201 let vfs_single = MemoryVfs::new();
6202 let vfs_prepared = MemoryVfs::new();
6203
6204 let mut adapter_single = make_adapter(&vfs_single, &cx);
6205 let mut adapter_prepared = make_adapter(&vfs_prepared, &cx);
6206
6207 let pages: Vec<Vec<u8>> = (0..4u8).map(sample_page).collect();
6208 let commit_sizes = [0_u32, 0, 0, 4];
6209
6210 for (index, page) in pages.iter().enumerate() {
6211 adapter_single
6212 .append_frame(
6213 &cx,
6214 u32::try_from(index + 1).expect("page number fits u32"),
6215 page,
6216 commit_sizes[index],
6217 )
6218 .expect("single append");
6219 }
6220
6221 let batch_frames: Vec<_> = pages
6222 .iter()
6223 .enumerate()
6224 .map(|(index, page)| WalFrameRef {
6225 page_number: u32::try_from(index + 1).expect("page number fits u32"),
6226 page_data: page,
6227 db_size_if_commit: commit_sizes[index],
6228 })
6229 .collect();
6230 let mut prepared = adapter_prepared
6231 .prepare_append_frames(&batch_frames)
6232 .expect("prepare append")
6233 .expect("prepared batch");
6234 adapter_prepared
6235 .append_prepared_frames(&cx, &mut prepared)
6236 .expect("append prepared");
6237
6238 assert_eq!(
6239 adapter_single.frame_count(),
6240 adapter_prepared.frame_count(),
6241 "prepared adapter append must preserve frame count"
6242 );
6243 assert_eq!(
6244 adapter_single.wal.running_checksum(),
6245 adapter_prepared.wal.running_checksum(),
6246 "prepared adapter append must preserve checksum chain"
6247 );
6248
6249 for frame_index in 0..pages.len() {
6250 let (single_header, single_data) = adapter_single
6251 .wal
6252 .read_frame(&cx, frame_index)
6253 .expect("read single frame");
6254 let (prepared_header, prepared_data) = adapter_prepared
6255 .wal
6256 .read_frame(&cx, frame_index)
6257 .expect("read prepared frame");
6258 assert_eq!(
6259 single_header, prepared_header,
6260 "frame header {frame_index} must match"
6261 );
6262 assert_eq!(
6263 single_data, prepared_data,
6264 "frame payload {frame_index} must match"
6265 );
6266 }
6267 }
6268
6269 #[test]
6270 fn test_adapter_pre_finalize_reused_when_append_window_is_stable() {
6271 let cx = test_cx();
6272 let vfs_single = MemoryVfs::new();
6273 let vfs_prepared = MemoryVfs::new();
6274
6275 let mut adapter_single = make_adapter(&vfs_single, &cx);
6276 let mut adapter_prepared = make_adapter(&vfs_prepared, &cx);
6277
6278 let pages: Vec<Vec<u8>> = (0..3u8).map(sample_page).collect();
6279 let commit_sizes = [0_u32, 0, 3];
6280
6281 for (index, page) in pages.iter().enumerate() {
6282 adapter_single
6283 .append_frame(
6284 &cx,
6285 u32::try_from(index + 1).expect("page number fits u32"),
6286 page,
6287 commit_sizes[index],
6288 )
6289 .expect("single append");
6290 }
6291
6292 let batch_frames: Vec<_> = pages
6293 .iter()
6294 .enumerate()
6295 .map(|(index, page)| WalFrameRef {
6296 page_number: u32::try_from(index + 1).expect("page number fits u32"),
6297 page_data: page,
6298 db_size_if_commit: commit_sizes[index],
6299 })
6300 .collect();
6301 let mut prepared = adapter_prepared
6302 .prepare_append_frames(&batch_frames)
6303 .expect("prepare append")
6304 .expect("prepared batch");
6305 adapter_prepared
6306 .finalize_prepared_frames(&cx, &mut prepared)
6307 .expect("pre-finalize prepared batch");
6308 let finalized_for = prepared.finalized_for.expect("finalization state");
6309 let finalized_running_checksum = prepared
6310 .finalized_running_checksum
6311 .expect("finalized checksum");
6312
6313 adapter_prepared
6314 .append_prepared_frames(&cx, &mut prepared)
6315 .expect("append prepared");
6316
6317 assert_eq!(
6318 prepared.finalized_for,
6319 Some(finalized_for),
6320 "stable append window should reuse the pre-lock finalization state"
6321 );
6322 assert_eq!(
6323 prepared.finalized_running_checksum,
6324 Some(finalized_running_checksum),
6325 "stable append window should reuse the pre-lock finalized checksum"
6326 );
6327 assert_eq!(
6328 adapter_single.wal.running_checksum(),
6329 adapter_prepared.wal.running_checksum(),
6330 "stable reuse path must preserve checksum chain"
6331 );
6332 }
6333
6334 #[test]
6335 fn test_adapter_pre_finalize_reseeds_after_intervening_external_append() {
6336 let cx = test_cx();
6337 let baseline_vfs = MemoryVfs::new();
6338 let shared_vfs = MemoryVfs::new();
6339
6340 let mut baseline = make_adapter(&baseline_vfs, &cx);
6341 let mut prepared_writer = make_adapter(&shared_vfs, &cx);
6342 let intruder_file = open_wal_file(&shared_vfs, &cx);
6343 let intruder_wal = WalFile::open(&cx, intruder_file).expect("open shared WAL");
6344 let mut intruder = WalBackendAdapter::new(intruder_wal);
6345
6346 let pages: Vec<Vec<u8>> = (0..3u8).map(sample_page).collect();
6347 let commit_sizes = [0_u32, 0, 3];
6348 let intruder_page = sample_page(0xEE);
6349
6350 baseline
6351 .append_frame(&cx, 99, &intruder_page, 1)
6352 .expect("baseline intruder append");
6353 for (index, page) in pages.iter().enumerate() {
6354 baseline
6355 .append_frame(
6356 &cx,
6357 u32::try_from(index + 1).expect("page number fits u32"),
6358 page,
6359 commit_sizes[index],
6360 )
6361 .expect("baseline append");
6362 }
6363
6364 let batch_frames: Vec<_> = pages
6365 .iter()
6366 .enumerate()
6367 .map(|(index, page)| WalFrameRef {
6368 page_number: u32::try_from(index + 1).expect("page number fits u32"),
6369 page_data: page,
6370 db_size_if_commit: commit_sizes[index],
6371 })
6372 .collect();
6373 let mut prepared = prepared_writer
6374 .prepare_append_frames(&batch_frames)
6375 .expect("prepare append")
6376 .expect("prepared batch");
6377 prepared_writer
6378 .finalize_prepared_frames(&cx, &mut prepared)
6379 .expect("pre-finalize prepared batch");
6380 let stale_finalization_state = prepared.finalized_for;
6381
6382 intruder
6383 .append_frame(&cx, 99, &intruder_page, 1)
6384 .expect("intruder append");
6385 intruder.sync(&cx).expect("intruder sync");
6386
6387 prepared_writer
6388 .append_prepared_frames(&cx, &mut prepared)
6389 .expect("append prepared after external growth");
6390
6391 assert_ne!(
6392 prepared.finalized_for, stale_finalization_state,
6393 "intervening external growth should force prepared batch reseeding"
6394 );
6395 assert_eq!(
6396 baseline.wal.running_checksum(),
6397 prepared_writer.wal.running_checksum(),
6398 "reseeding path must preserve checksum chain"
6399 );
6400 assert_eq!(
6401 baseline.frame_count(),
6402 prepared_writer.frame_count(),
6403 "reseeding path must preserve frame count"
6404 );
6405 }
6406
6407 #[test]
6408 fn test_adapter_pins_read_snapshot_until_next_begin() {
6409 init_wal_publication_test_tracing();
6410 let cx = test_cx();
6411 let vfs = MemoryVfs::new();
6412
6413 let file_writer = open_wal_file(&vfs, &cx);
6414 let wal_writer =
6415 WalFile::create(&cx, file_writer, PAGE_SIZE, 0, test_salts()).expect("create WAL");
6416 let mut writer = WalBackendAdapter::new(wal_writer);
6417
6418 let file_reader = open_wal_file(&vfs, &cx);
6419 let wal_reader = WalFile::open(&cx, file_reader).expect("open WAL");
6420 let mut reader = WalBackendAdapter::new(wal_reader);
6421
6422 let v1 = sample_page(0x41);
6423 writer.append_frame(&cx, 3, &v1, 3).expect("append v1");
6424 writer.sync(&cx).expect("sync v1");
6425
6426 reader
6427 .begin_transaction(&cx)
6428 .expect("begin reader snapshot 1");
6429 let pinned_v1 = reader
6430 .pinned_read_snapshot()
6431 .expect("reader pins publication snapshot");
6432 assert_eq!(pinned_v1.last_commit_frame, Some(0));
6433 assert_eq!(pinned_v1.commit_count, 1);
6434 assert_eq!(pinned_v1.latest_frame_entries, 1);
6435 assert!(pinned_v1.lookup_contract_is_authoritative());
6436 assert_eq!(
6437 reader.read_page(&cx, 3).expect("reader sees v1"),
6438 Some(v1.clone())
6439 );
6440
6441 let v2 = sample_page(0x42);
6442 writer.append_frame(&cx, 3, &v2, 3).expect("append v2");
6443 writer.sync(&cx).expect("sync v2");
6444
6445 assert_eq!(
6447 reader
6448 .read_page(&cx, 3)
6449 .expect("reader remains on pinned snapshot"),
6450 Some(v1.clone())
6451 );
6452 assert_eq!(
6453 reader
6454 .pinned_read_snapshot()
6455 .expect("reader keeps the same pinned snapshot"),
6456 pinned_v1,
6457 "pinned publication metadata must stay stable until the next begin"
6458 );
6459
6460 reader
6462 .begin_transaction(&cx)
6463 .expect("begin reader snapshot 2");
6464 let pinned_v2 = reader
6465 .pinned_read_snapshot()
6466 .expect("reader repins publication snapshot");
6467 assert!(pinned_v2.publication_seq > pinned_v1.publication_seq);
6468 assert_eq!(pinned_v2.commit_count, 2);
6469 assert_eq!(pinned_v2.latest_frame_entries, 1);
6470 assert_eq!(reader.read_page(&cx, 3).expect("reader sees v2"), Some(v2));
6471 }
6472
6473 #[test]
6474 fn test_adapter_read_page_hides_uncommitted_frames() {
6475 let cx = test_cx();
6476 let vfs = MemoryVfs::new();
6477 let mut adapter = make_adapter(&vfs, &cx);
6478
6479 let committed = sample_page(0x31);
6480 let uncommitted = sample_page(0x32);
6481
6482 adapter
6483 .append_frame(&cx, 7, &committed, 7)
6484 .expect("append committed frame");
6485 adapter.sync(&cx).expect("publish committed frame");
6488 adapter
6489 .append_frame(&cx, 7, &uncommitted, 0)
6490 .expect("append uncommitted frame");
6491
6492 let result = adapter.read_page(&cx, 7).expect("read committed page");
6493 assert_eq!(
6494 result,
6495 Some(committed),
6496 "reader must ignore uncommitted (and unpublished) tail frames"
6497 );
6498 }
6499
6500 #[test]
6501 fn test_adapter_read_page_none_when_wal_has_no_commit_frame() {
6502 let cx = test_cx();
6503 let vfs = MemoryVfs::new();
6504 let mut adapter = make_adapter(&vfs, &cx);
6505
6506 adapter
6507 .append_frame(&cx, 3, &sample_page(0x44), 0)
6508 .expect("append uncommitted frame");
6509
6510 let result = adapter.read_page(&cx, 3).expect("read page");
6511 assert_eq!(result, None, "uncommitted WAL frames must stay invisible");
6512 }
6513
6514 #[test]
6515 fn test_adapter_read_page_empty_wal() {
6516 let cx = test_cx();
6517 let vfs = MemoryVfs::new();
6518 let mut adapter = make_adapter(&vfs, &cx);
6519
6520 let result = adapter.read_page(&cx, 1).expect("read from empty WAL");
6521 assert_eq!(result, None);
6522 }
6523
6524 #[test]
6525 fn test_adapter_sync() {
6526 let cx = test_cx();
6527 let vfs = MemoryVfs::new();
6528 let mut adapter = make_adapter(&vfs, &cx);
6529
6530 adapter
6531 .append_frame(&cx, 1, &sample_page(0), 1)
6532 .expect("append");
6533 adapter.sync(&cx).expect("sync should not fail");
6534 }
6535
6536 #[test]
6537 fn test_adapter_into_inner_fails_closed_until_sync() {
6538 let cx = test_cx();
6539 let staged_vfs = MemoryVfs::new();
6540 let mut staged = make_adapter(&staged_vfs, &cx);
6541
6542 staged
6543 .append_frame(&cx, 1, &sample_page(0), 1)
6544 .expect("append");
6545 assert!(
6546 matches!(staged.into_inner(), Err(FrankenError::Busy)),
6547 "an unsynced commit must prevent consuming the adapter"
6548 );
6549
6550 let synced_vfs = MemoryVfs::new();
6551 let mut synced = make_adapter(&synced_vfs, &cx);
6552 synced
6553 .append_frame(&cx, 1, &sample_page(0), 1)
6554 .expect("append");
6555 synced.sync(&cx).expect("sync staged commit");
6556
6557 assert_eq!(synced.inner().frame_count(), 1);
6558
6559 let wal = synced.into_inner().expect("sync drained the staged frames");
6560 assert_eq!(wal.frame_count(), 1);
6561 }
6562
6563 #[test]
6564 fn test_adapter_as_dyn_wal_backend() {
6565 let cx = test_cx();
6566 let vfs = MemoryVfs::new();
6567 let mut adapter = make_adapter(&vfs, &cx);
6568
6569 let backend: &mut dyn WalBackend = &mut adapter;
6571 backend
6572 .append_frame(&cx, 1, &sample_page(0x77), 1)
6573 .expect("append via dyn");
6574 assert_eq!(backend.frame_count(), 1);
6575
6576 backend.sync(&cx).expect("publish via dyn");
6578 let page = backend.read_page(&cx, 1).expect("read via dyn");
6579 assert_eq!(page, Some(sample_page(0x77)));
6580 }
6581
6582 #[test]
6583 fn test_publication_snapshots_are_visible_through_wal_backend_trait() {
6584 init_wal_publication_test_tracing();
6585 let cx = test_cx();
6586 let vfs = MemoryVfs::new();
6587
6588 let file_writer = open_wal_file(&vfs, &cx);
6589 let wal_writer =
6590 WalFile::create(&cx, file_writer, PAGE_SIZE, 0, test_salts()).expect("create WAL");
6591 let mut writer = WalBackendAdapter::new(wal_writer);
6592
6593 writer
6594 .append_frame(&cx, 4, &sample_page(0x84), 4)
6595 .expect("append committed frame");
6596 writer.sync(&cx).expect("sync committed frame");
6597
6598 let file_reader = open_wal_file(&vfs, &cx);
6599 let wal_reader = WalFile::open(&cx, file_reader).expect("open WAL");
6600 let mut reader = WalBackendAdapter::new(wal_reader);
6601 let backend: &mut dyn WalBackend = &mut reader;
6602
6603 let published_before = backend
6604 .published_snapshot()
6605 .expect("trait should expose the adapter publication summary");
6606 assert_eq!(published_before.last_commit_frame, None);
6607 assert_eq!(published_before.commit_count, 0);
6608
6609 let refreshed = backend
6610 .refresh_published_snapshot(&cx)
6611 .expect("refresh through trait should succeed")
6612 .expect("adapter should republish an existing committed prefix");
6613 assert_eq!(refreshed.last_commit_frame, Some(0));
6614 assert_eq!(refreshed.commit_count, 1);
6615 assert_eq!(refreshed.latest_frame_entries, 1);
6616
6617 backend
6618 .begin_transaction(&cx)
6619 .expect("begin_transaction through trait should pin snapshot");
6620 let pinned = backend
6621 .pinned_read_snapshot()
6622 .expect("trait should expose the pinned read snapshot");
6623 assert_eq!(pinned, refreshed);
6624 }
6625
6626 #[test]
6629 fn test_page_index_returns_correct_data() {
6630 let cx = test_cx();
6632 let vfs = MemoryVfs::new();
6633 let mut adapter = make_adapter(&vfs, &cx);
6634
6635 let page1 = sample_page(0x01);
6636 let page2 = sample_page(0x02);
6637 let page3 = sample_page(0x03);
6638
6639 adapter.append_frame(&cx, 1, &page1, 0).expect("append");
6640 adapter.append_frame(&cx, 2, &page2, 0).expect("append");
6641 adapter
6642 .append_frame(&cx, 3, &page3, 3)
6643 .expect("append commit");
6644 adapter.sync(&cx).expect("publish staged frames");
6645
6646 assert_eq!(adapter.read_page(&cx, 1).expect("read"), Some(page1));
6648 assert_eq!(adapter.read_page(&cx, 2).expect("read"), Some(page2));
6649 assert_eq!(adapter.read_page(&cx, 3).expect("read"), Some(page3));
6650
6651 assert_eq!(adapter.read_page(&cx, 99).expect("read"), None);
6653 }
6654
6655 #[test]
6656 fn test_page_index_returns_latest_version() {
6657 let cx = test_cx();
6659 let vfs = MemoryVfs::new();
6660 let mut adapter = make_adapter(&vfs, &cx);
6661
6662 let old_data = sample_page(0xAA);
6663 let new_data = sample_page(0xBB);
6664
6665 adapter
6666 .append_frame(&cx, 5, &old_data, 0)
6667 .expect("append old");
6668 adapter
6669 .append_frame(&cx, 5, &new_data, 1)
6670 .expect("append new (commit)");
6671 adapter.sync(&cx).expect("publish staged frames");
6672
6673 assert_eq!(
6674 adapter.read_page(&cx, 5).expect("read"),
6675 Some(new_data),
6676 "page index must return the latest frame for a page"
6677 );
6678 }
6679
6680 #[test]
6681 fn test_page_index_invalidated_on_wal_reset() {
6682 let cx = test_cx();
6685 let vfs = MemoryVfs::new();
6686 let mut adapter = make_adapter(&vfs, &cx);
6687
6688 let old_data = sample_page(0x11);
6689 adapter
6690 .append_frame(&cx, 1, &old_data, 1)
6691 .expect("append commit");
6692 adapter.sync(&cx).expect("publish staged frames");
6693
6694 assert_eq!(adapter.read_page(&cx, 1).expect("read old"), Some(old_data));
6696
6697 let new_salts = WalSalts {
6699 salt1: 0xAAAA_BBBB,
6700 salt2: 0xCCCC_DDDD,
6701 };
6702 adapter
6703 .inner_mut()
6704 .expect("no staged batch blocks inner access")
6705 .reset(&cx, 1, new_salts, false)
6706 .expect("WAL reset");
6707
6708 let new_data = sample_page(0x22);
6710 adapter
6711 .append_frame(&cx, 1, &new_data, 1)
6712 .expect("append new generation commit");
6713 adapter.sync(&cx).expect("publish new generation commit");
6714
6715 let result = adapter.read_page(&cx, 1).expect("read after reset");
6717 assert_eq!(
6718 result,
6719 Some(new_data),
6720 "after WAL reset, page index must return new-generation data, not stale cached data"
6721 );
6722
6723 let old_only = sample_page(0x33);
6725 assert_eq!(
6727 adapter.read_page(&cx, 99).expect("read non-existent"),
6728 None,
6729 "pages from old WAL generation must not appear after reset"
6730 );
6731 drop(old_only);
6733 }
6734
6735 #[test]
6736 fn test_page_index_invalidated_on_same_salt_generation_change() {
6737 init_wal_publication_test_tracing();
6738 let cx = test_cx();
6741 let vfs = MemoryVfs::new();
6742 let mut adapter = make_adapter(&vfs, &cx);
6743
6744 let reused_salts = adapter.inner().header().salts;
6745 let old_data = sample_page(0x11);
6746 adapter
6747 .append_frame(&cx, 1, &old_data, 1)
6748 .expect("append commit");
6749 adapter.sync(&cx).expect("publish staged frames");
6750 assert_eq!(adapter.read_page(&cx, 1).expect("read old"), Some(old_data));
6751
6752 adapter
6753 .inner_mut()
6754 .expect("no staged batch blocks inner access")
6755 .reset(&cx, 1, reused_salts, false)
6756 .expect("reset with same salts");
6757 let new_data = sample_page(0x22);
6758 adapter
6759 .append_frame(&cx, 2, &new_data, 2)
6760 .expect("append new generation commit");
6761 adapter.sync(&cx).expect("publish new generation commit");
6762 let refreshed = adapter
6763 .refresh_published_snapshot(&cx)
6764 .expect("refresh published snapshot after same-salt reset");
6765 assert_eq!(refreshed.generation.checkpoint_seq, 1);
6766 assert_eq!(refreshed.generation.salts, reused_salts);
6767 assert_eq!(refreshed.last_commit_frame, Some(0));
6768 assert_eq!(refreshed.commit_count, 1);
6769 assert_eq!(refreshed.latest_frame_entries, 1);
6770
6771 assert_eq!(
6772 adapter.read_page(&cx, 1).expect("old page should be gone"),
6773 None,
6774 "cached index entries from the previous generation must be invalidated"
6775 );
6776 assert_eq!(
6777 adapter.read_page(&cx, 2).expect("read new page"),
6778 Some(new_data),
6779 "adapter must resolve pages from the new generation even when salts are reused"
6780 );
6781 }
6782
6783 #[test]
6784 fn test_refresh_published_snapshot_materializes_existing_committed_prefix() {
6785 init_wal_publication_test_tracing();
6786 let cx = test_cx();
6787 let vfs = MemoryVfs::new();
6788
6789 let file_writer = open_wal_file(&vfs, &cx);
6790 let wal_writer =
6791 WalFile::create(&cx, file_writer, PAGE_SIZE, 0, test_salts()).expect("create WAL");
6792 let mut writer = WalBackendAdapter::new(wal_writer);
6793
6794 let p1 = sample_page(0x71);
6795 let p2 = sample_page(0x72);
6796 writer.append_frame(&cx, 1, &p1, 0).expect("append p1");
6797 writer
6798 .append_frame(&cx, 2, &p2, 2)
6799 .expect("append p2 commit");
6800 writer.sync(&cx).expect("sync writer");
6801
6802 let file_reader = open_wal_file(&vfs, &cx);
6803 let wal_reader = WalFile::open(&cx, file_reader).expect("open reader WAL");
6804 let mut reader = WalBackendAdapter::new(wal_reader);
6805
6806 let before = reader.published_snapshot();
6807 assert_eq!(before.last_commit_frame, None);
6808 assert_eq!(before.commit_count, 0);
6809 assert_eq!(before.latest_frame_entries, 0);
6810
6811 let refreshed = reader
6812 .refresh_published_snapshot(&cx)
6813 .expect("refresh published snapshot");
6814 assert_eq!(refreshed.last_commit_frame, Some(1));
6815 assert_eq!(refreshed.commit_count, 1);
6816 assert_eq!(refreshed.latest_frame_entries, 2);
6817 assert!(refreshed.lookup_contract_is_authoritative());
6818 assert_eq!(reader.read_page(&cx, 1).expect("read p1"), Some(p1));
6819 assert_eq!(reader.read_page(&cx, 2).expect("read p2"), Some(p2));
6820 }
6821
6822 #[test]
6823 fn test_page_index_incremental_extend_after_durable_sync() {
6824 let cx = test_cx();
6827 let vfs = MemoryVfs::new();
6828 let mut adapter = make_adapter(&vfs, &cx);
6829
6830 let page1 = sample_page(0x10);
6831 adapter
6832 .append_frame(&cx, 1, &page1, 1)
6833 .expect("append commit 1");
6834 adapter.sync(&cx).expect("durably publish commit 1");
6835
6836 assert_eq!(
6838 adapter.read_page(&cx, 1).expect("read"),
6839 Some(page1.clone())
6840 );
6841
6842 let page2 = sample_page(0x20);
6844 let page1_v2 = sample_page(0x30);
6845 adapter
6846 .append_frame(&cx, 2, &page2, 0)
6847 .expect("append page 2");
6848 adapter
6849 .append_frame(&cx, 1, &page1_v2, 3)
6850 .expect("append page 1 v2 (commit)");
6851 adapter.sync(&cx).expect("durably publish commit 2");
6852
6853 assert_eq!(
6855 adapter.read_page(&cx, 1).expect("read page 1 v2"),
6856 Some(page1_v2),
6857 "incremental index extend should pick up the updated page"
6858 );
6859 assert_eq!(adapter.read_page(&cx, 2).expect("read page 2"), Some(page2));
6860 }
6861
6862 fn commit_batch_pages() -> (Vec<u8>, Vec<u8>) {
6864 (sample_page(0x71), sample_page(0x72))
6865 }
6866
6867 fn assert_publication_unchanged(adapter: &WalBackendAdapter<impl VfsFile>, context: &str) {
6869 assert_eq!(
6870 adapter.published_snapshot.last_commit_frame, None,
6871 "{context}: publication must not advance before a successful sync"
6872 );
6873 assert_eq!(
6874 adapter.published_snapshot.commit_count, 0,
6875 "{context}: commit count must not advance before a successful sync"
6876 );
6877 assert!(
6878 adapter.published_snapshot.page_index.is_empty(),
6879 "{context}: no page may be visible before a successful sync"
6880 );
6881 }
6882
6883 #[test]
6884 fn test_append_frame_without_sync_leaves_publication_unchanged() {
6885 let cx = test_cx();
6886 let vfs = MemoryVfs::new();
6887 let mut adapter = make_adapter(&vfs, &cx);
6888
6889 let (p1, p2) = commit_batch_pages();
6890 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
6891 adapter.append_frame(&cx, 2, &p2, 2).expect("append commit");
6892
6893 assert_publication_unchanged(&adapter, "append_frame");
6894 assert_eq!(
6895 adapter.pending_publication_commit,
6896 Some(1),
6897 "append_frame must stage the commit horizon for a later sync"
6898 );
6899 }
6900
6901 #[test]
6902 fn test_append_frames_without_sync_leaves_publication_unchanged() {
6903 let cx = test_cx();
6904 let vfs = MemoryVfs::new();
6905 let mut adapter = make_adapter(&vfs, &cx);
6906
6907 let (p1, p2) = commit_batch_pages();
6908 let frames = [
6909 WalFrameRef {
6910 page_number: 1,
6911 page_data: &p1,
6912 db_size_if_commit: 0,
6913 },
6914 WalFrameRef {
6915 page_number: 2,
6916 page_data: &p2,
6917 db_size_if_commit: 2,
6918 },
6919 ];
6920 adapter
6921 .append_frames(&cx, &frames)
6922 .expect("append frames batch");
6923
6924 assert_publication_unchanged(&adapter, "append_frames");
6925 assert_eq!(
6926 adapter.pending_publication_commit,
6927 Some(1),
6928 "append_frames must stage the commit horizon for a later sync"
6929 );
6930 }
6931
6932 #[test]
6933 fn test_append_frames_tracked_without_sync_leaves_publication_unchanged() {
6934 let cx = test_cx();
6935 let vfs = MemoryVfs::new();
6936 let mut adapter = make_adapter(&vfs, &cx);
6937
6938 let (p1, p2) = commit_batch_pages();
6939 let frames = [
6940 WalFrameRef {
6941 page_number: 1,
6942 page_data: &p1,
6943 db_size_if_commit: 0,
6944 },
6945 WalFrameRef {
6946 page_number: 2,
6947 page_data: &p2,
6948 db_size_if_commit: 2,
6949 },
6950 ];
6951 adapter
6952 .append_frames_tracked(&cx, &frames, VfsWriteCompletion::new())
6953 .expect("append tracked frames batch");
6954
6955 assert_publication_unchanged(&adapter, "append_frames_tracked");
6956 assert_eq!(
6957 adapter.pending_publication_commit,
6958 Some(1),
6959 "append_frames_tracked must stage the commit horizon for a later sync"
6960 );
6961 }
6962
6963 #[test]
6964 fn test_append_prepared_frames_without_sync_leaves_publication_unchanged() {
6965 let cx = test_cx();
6966 let vfs = MemoryVfs::new();
6967 let mut adapter = make_adapter(&vfs, &cx);
6968
6969 let (p1, p2) = commit_batch_pages();
6970 let frames = [
6971 WalFrameRef {
6972 page_number: 1,
6973 page_data: &p1,
6974 db_size_if_commit: 0,
6975 },
6976 WalFrameRef {
6977 page_number: 2,
6978 page_data: &p2,
6979 db_size_if_commit: 2,
6980 },
6981 ];
6982 let mut prepared = adapter
6983 .prepare_append_frames(&frames)
6984 .expect("prepare append")
6985 .expect("prepared batch");
6986 adapter
6987 .append_prepared_frames(&cx, &mut prepared)
6988 .expect("append prepared");
6989
6990 assert_publication_unchanged(&adapter, "append_prepared_frames");
6991 assert_eq!(
6992 adapter.pending_publication_commit,
6993 Some(1),
6994 "append_prepared_frames must stage the commit horizon for a later sync"
6995 );
6996 }
6997
6998 #[test]
6999 fn test_append_prepared_frames_tracked_without_sync_leaves_publication_unchanged() {
7000 let cx = test_cx();
7001 let vfs = MemoryVfs::new();
7002 let mut adapter = make_adapter(&vfs, &cx);
7003
7004 let (p1, p2) = commit_batch_pages();
7005 let frames = [
7006 WalFrameRef {
7007 page_number: 1,
7008 page_data: &p1,
7009 db_size_if_commit: 0,
7010 },
7011 WalFrameRef {
7012 page_number: 2,
7013 page_data: &p2,
7014 db_size_if_commit: 2,
7015 },
7016 ];
7017 let mut prepared = adapter
7018 .prepare_append_frames(&frames)
7019 .expect("prepare append")
7020 .expect("prepared batch");
7021 adapter
7022 .append_prepared_frames_tracked(&cx, &mut prepared, VfsWriteCompletion::new())
7023 .expect("append prepared tracked");
7024
7025 assert_publication_unchanged(&adapter, "append_prepared_frames_tracked");
7026 assert_eq!(
7027 adapter.pending_publication_commit,
7028 Some(1),
7029 "append_prepared_frames_tracked must stage the commit horizon for a later sync"
7030 );
7031 }
7032
7033 #[test]
7034 fn test_successful_sync_publishes_staged_commit_horizon() {
7035 let cx = test_cx();
7036 let vfs = MemoryVfs::new();
7037 let mut adapter = make_adapter(&vfs, &cx);
7038
7039 let (p1, p2) = commit_batch_pages();
7040 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
7041 adapter.append_frame(&cx, 2, &p2, 2).expect("append commit");
7042 assert_publication_unchanged(&adapter, "before sync");
7043
7044 adapter.sync(&cx).expect("sync must succeed");
7045
7046 assert_eq!(
7047 adapter.published_snapshot.last_commit_frame,
7048 Some(1),
7049 "a successful sync must publish the staged commit horizon"
7050 );
7051 assert_eq!(
7052 adapter.published_snapshot.commit_count, 1,
7053 "a successful sync must publish the staged commit count"
7054 );
7055 assert_eq!(
7056 adapter.published_snapshot.page_index.len(),
7057 2,
7058 "a successful sync must publish every staged page"
7059 );
7060 assert_eq!(
7061 adapter.pending_publication_commit, None,
7062 "a published batch must no longer be staged"
7063 );
7064 assert!(
7065 adapter.pending_publication_frames.is_empty(),
7066 "a published batch must drain its staged frames"
7067 );
7068 }
7069
7070 #[test]
7071 fn test_failed_sync_advances_no_publication_and_retry_publishes() {
7072 let cx = test_cx();
7073 let vfs = CheckpointHandoffFaultVfs::new();
7074 let mut adapter = make_fault_adapter(&vfs, &cx);
7075
7076 let (p1, p2) = commit_batch_pages();
7077 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
7078 adapter.append_frame(&cx, 2, &p2, 2).expect("append commit");
7079
7080 vfs.fail_next_wal_sync();
7081 let failure = adapter
7082 .sync(&cx)
7083 .expect_err("injected WAL sync failure must surface");
7084 assert!(
7085 failure.to_string().contains("injected WAL sync failure"),
7086 "sync must report the injected durability failure, got: {failure}"
7087 );
7088
7089 assert_publication_unchanged(&adapter, "after failed sync");
7090 assert_eq!(
7091 adapter.pending_publication_commit,
7092 Some(1),
7093 "a failed sync must preserve the staged horizon for retry"
7094 );
7095 assert!(
7096 !adapter.pending_publication_frames.is_empty(),
7097 "a failed sync must preserve staged frames for retry"
7098 );
7099
7100 adapter.sync(&cx).expect("retry sync must succeed");
7102
7103 assert_eq!(
7104 adapter.published_snapshot.last_commit_frame,
7105 Some(1),
7106 "retrying sync must publish the preserved commit horizon"
7107 );
7108 assert_eq!(
7109 adapter.published_snapshot.commit_count, 1,
7110 "retrying sync must publish the preserved commit count"
7111 );
7112 assert_eq!(
7113 adapter.published_snapshot.page_index.len(),
7114 2,
7115 "retrying sync must publish every preserved page"
7116 );
7117 assert_eq!(
7118 adapter.pending_publication_commit, None,
7119 "a retried publication must clear the staged horizon"
7120 );
7121 }
7122
7123 #[test]
7124 fn test_failed_sync_then_append_cannot_drop_or_publish_pending() {
7125 let cx = test_cx();
7126 let vfs = CheckpointHandoffFaultVfs::new();
7127 let mut adapter = make_fault_adapter(&vfs, &cx);
7128
7129 let (p1, p2) = commit_batch_pages();
7130 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
7131 adapter.append_frame(&cx, 2, &p2, 2).expect("append commit");
7132
7133 vfs.fail_next_wal_sync();
7134 adapter
7135 .sync(&cx)
7136 .expect_err("injected WAL sync failure must surface");
7137
7138 let staged_after_failure = adapter.pending_publication_commit;
7139 let staged_frames_after_failure = adapter.pending_publication_frames.len();
7140 assert_eq!(
7141 staged_after_failure,
7142 Some(1),
7143 "failed sync must preserve the staged horizon"
7144 );
7145
7146 let p3 = sample_page(0x73);
7149 adapter
7150 .append_frame(&cx, 3, &p3, 3)
7151 .expect("append after failed sync");
7152
7153 assert_publication_unchanged(&adapter, "append after failed sync");
7154 assert!(
7155 adapter.pending_publication_frames.len() > staged_frames_after_failure,
7156 "append after a failed sync must extend, never discard, the staged batch"
7157 );
7158 assert_eq!(
7159 adapter.pending_publication_commit,
7160 Some(2),
7161 "append after a failed sync must carry the staged horizon forward"
7162 );
7163
7164 adapter.sync(&cx).expect("sync after failed attempt");
7166 assert_eq!(
7167 adapter.published_snapshot.last_commit_frame,
7168 Some(2),
7169 "recovered sync must publish the full preserved horizon"
7170 );
7171 assert_eq!(
7172 adapter.pending_publication_commit, None,
7173 "recovered sync must clear the staged horizon"
7174 );
7175 }
7176
7177 #[test]
7178 fn test_failed_sync_then_begin_transaction_then_append_fails_closed() {
7179 let cx = test_cx();
7180 let vfs = CheckpointHandoffFaultVfs::new();
7181 let mut adapter = make_fault_adapter(&vfs, &cx);
7182
7183 let (p1, p2) = commit_batch_pages();
7184 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
7185 adapter.append_frame(&cx, 2, &p2, 2).expect("append commit");
7186
7187 vfs.fail_next_wal_sync();
7188 adapter
7189 .sync(&cx)
7190 .expect_err("injected WAL sync failure must surface");
7191 assert_eq!(
7192 adapter.pending_publication_commit,
7193 Some(1),
7194 "failed sync must preserve the staged horizon"
7195 );
7196
7197 let begin_error = adapter
7201 .begin_transaction(&cx)
7202 .expect_err("begin_transaction must fail closed while frames are staged");
7203 assert!(
7204 matches!(begin_error, FrankenError::Busy),
7205 "staged-state rejection must be retryable Busy, not corruption: {begin_error:?}"
7206 );
7207 assert_publication_unchanged(&adapter, "begin_transaction refused after failed sync");
7208 assert_eq!(
7209 adapter.pending_publication_commit,
7210 Some(1),
7211 "a refused begin_transaction must not drop the staged horizon"
7212 );
7213 assert!(
7214 adapter.pinned_read_snapshot().is_none(),
7215 "a refused begin_transaction must not pin a read snapshot"
7216 );
7217
7218 adapter.refresh_before_append = true;
7221 let p3 = sample_page(0x74);
7222 let append_error = adapter
7223 .append_frame(&cx, 3, &p3, 3)
7224 .expect_err("append must fail closed while frames are staged");
7225 assert!(
7226 matches!(append_error, FrankenError::Busy),
7227 "append rejection must be retryable Busy: {append_error:?}"
7228 );
7229 assert_publication_unchanged(&adapter, "append refused after failed sync");
7230 assert_eq!(
7231 adapter.pending_publication_commit,
7232 Some(1),
7233 "a refused append must leave the staged horizon intact"
7234 );
7235 assert!(
7236 !adapter.pending_publication_frames.is_empty(),
7237 "a refused append must leave the staged frames intact"
7238 );
7239 adapter.refresh_before_append = false;
7240
7241 adapter.sync(&cx).expect("sync after failed attempt");
7243 assert_eq!(
7244 adapter.published_snapshot.last_commit_frame,
7245 Some(1),
7246 "recovered sync must publish the preserved horizon"
7247 );
7248 }
7249
7250 #[test]
7251 fn test_failed_sync_then_checkpoint_fails_closed_and_preserves_state() {
7252 let cx = test_cx();
7253 let vfs = CheckpointHandoffFaultVfs::new();
7254 let mut adapter = make_fault_adapter(&vfs, &cx);
7255
7256 let (p1, p2) = commit_batch_pages();
7257 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
7258 adapter.append_frame(&cx, 2, &p2, 2).expect("append commit");
7259
7260 vfs.fail_next_wal_sync();
7261 adapter
7262 .sync(&cx)
7263 .expect_err("injected WAL sync failure must surface");
7264
7265 let frames_before = adapter.frame_count();
7266 let staged_before = adapter.pending_publication_commit;
7267 let staged_frame_count_before = adapter.pending_publication_frames.len();
7268
7269 let mut writer = MockCheckpointPageWriter;
7273 let checkpoint_error = adapter
7274 .checkpoint(&cx, CheckpointMode::Passive, &mut writer, 0, None)
7275 .expect_err("checkpoint must fail closed while frames are staged");
7276 assert!(
7277 matches!(checkpoint_error, FrankenError::CheckpointFailed { .. }),
7278 "checkpoint rejection must be CheckpointFailed, not corruption: {checkpoint_error:?}"
7279 );
7280
7281 assert_eq!(
7282 adapter.frame_count(),
7283 frames_before,
7284 "a refused checkpoint must not mutate WAL bytes"
7285 );
7286 assert_publication_unchanged(&adapter, "checkpoint refused");
7287 assert_eq!(
7288 adapter.pending_publication_commit, staged_before,
7289 "a refused checkpoint must preserve the staged horizon"
7290 );
7291 assert_eq!(
7292 adapter.pending_publication_frames.len(),
7293 staged_frame_count_before,
7294 "a refused checkpoint must preserve the staged frames"
7295 );
7296
7297 adapter.sync(&cx).expect("retry sync must succeed");
7299 assert_eq!(
7300 adapter.published_snapshot.last_commit_frame,
7301 Some(1),
7302 "retry sync must publish the preserved horizon"
7303 );
7304 assert_eq!(
7305 adapter.pending_publication_commit, None,
7306 "a published batch must no longer be staged"
7307 );
7308 }
7309
7310 #[test]
7311 fn test_midtransaction_sync_preserves_uncommitted_frames_and_allows_continuation() {
7312 let cx = test_cx();
7313 let vfs = MemoryVfs::new();
7314 let mut adapter = make_adapter(&vfs, &cx);
7315
7316 let (p1, p2) = commit_batch_pages();
7317
7318 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
7321 assert_eq!(
7322 adapter.pending_publication_commit, None,
7323 "a non-commit append stages no commit horizon"
7324 );
7325 adapter
7326 .sync(&cx)
7327 .expect("mid-transaction sync must succeed");
7328
7329 assert_publication_unchanged(&adapter, "sync of uncommitted frames");
7330 assert!(
7331 !adapter.pending_publication_frames.is_empty(),
7332 "a mid-transaction sync must preserve durable-but-uncommitted frames"
7333 );
7334
7335 adapter
7337 .append_frame(&cx, 2, &p2, 2)
7338 .expect("commit append after mid-transaction sync must be allowed");
7339 assert_eq!(
7340 adapter.pending_publication_commit,
7341 Some(1),
7342 "the commit append must stage the horizon for the whole batch"
7343 );
7344 assert_publication_unchanged(&adapter, "commit staged but not yet synced");
7345
7346 adapter.sync(&cx).expect("commit sync must succeed");
7347
7348 assert_eq!(
7349 adapter.published_snapshot.last_commit_frame,
7350 Some(1),
7351 "the commit sync must publish the whole batch"
7352 );
7353 assert_eq!(
7354 adapter.published_snapshot.commit_count, 1,
7355 "the batch must publish exactly one commit"
7356 );
7357 assert_eq!(
7358 adapter.published_snapshot.page_index.len(),
7359 2,
7360 "both pages must be published exactly once"
7361 );
7362 assert_eq!(
7363 adapter.published_snapshot.page_index.get(&1),
7364 Some(&0),
7365 "page 1 must map to its frame from before the mid-transaction sync"
7366 );
7367 assert_eq!(
7368 adapter.published_snapshot.page_index.get(&2),
7369 Some(&1),
7370 "page 2 must map to the commit frame"
7371 );
7372 assert!(
7373 !adapter.has_pending_publication(),
7374 "a published batch must leave nothing staged"
7375 );
7376 }
7377
7378 #[test]
7379 fn test_inner_mut_fails_closed_while_batch_is_staged() {
7380 let cx = test_cx();
7381 let vfs = CheckpointHandoffFaultVfs::new();
7382 let mut adapter = make_fault_adapter(&vfs, &cx);
7383
7384 let (p1, p2) = commit_batch_pages();
7385 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
7386 adapter.append_frame(&cx, 2, &p2, 2).expect("append commit");
7387
7388 assert!(
7389 adapter.has_pending_publication(),
7390 "an appended-but-unsynced batch must report as pending"
7391 );
7392 assert!(
7395 matches!(adapter.inner_mut(), Err(FrankenError::Busy)),
7396 "inner_mut must fail closed with retryable Busy while frames are staged"
7397 );
7398 assert_eq!(
7399 adapter.pending_publication_commit,
7400 Some(1),
7401 "a refused inner_mut must preserve the staged horizon"
7402 );
7403
7404 adapter.sync(&cx).expect("sync staged batch");
7406 assert!(
7407 !adapter.has_pending_publication(),
7408 "a published batch must clear the pending flag"
7409 );
7410 adapter
7411 .inner_mut()
7412 .expect("inner_mut must succeed once the batch is drained");
7413 }
7414
7415 #[test]
7416 fn test_unpinned_refresh_does_not_expose_staged_horizon_before_sync() {
7417 let cx = test_cx();
7418 let vfs = MemoryVfs::new();
7419 let mut adapter = make_adapter(&vfs, &cx);
7420
7421 let (p1, p2) = commit_batch_pages();
7422 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
7423 adapter.append_frame(&cx, 2, &p2, 2).expect("append commit");
7424
7425 adapter
7428 .refresh_published_snapshot(&cx)
7429 .expect("refresh published snapshot");
7430 assert_publication_unchanged(&adapter, "refresh with staged frames");
7431 assert_eq!(
7432 adapter.pending_publication_commit,
7433 Some(1),
7434 "refresh must leave the staged horizon intact"
7435 );
7436
7437 adapter.sync(&cx).expect("sync staged batch");
7438 assert_eq!(
7439 adapter.published_snapshot.last_commit_frame,
7440 Some(1),
7441 "sync must publish once the staged batch is durable"
7442 );
7443 }
7444
7445 #[test]
7446 fn test_authorized_deferred_commit_publishes_without_claiming_fsync() {
7447 let cx = test_cx();
7448 let vfs = MemoryVfs::new();
7449 let mut adapter = make_adapter(&vfs, &cx);
7450
7451 let (p1, p2) = commit_batch_pages();
7452 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
7453 adapter.append_frame(&cx, 2, &p2, 2).expect("append commit");
7454 let fsynced_before = adapter.wal.last_fsynced_frame_count();
7455
7456 adapter
7457 .publish_authorized_deferred_commit(&cx)
7458 .expect("parallel-WAL authorization must publish the deferred commit");
7459
7460 assert_eq!(
7461 adapter.published_snapshot.last_commit_frame,
7462 Some(1),
7463 "the authorized commit marker must become visible"
7464 );
7465 assert_eq!(
7466 adapter.published_snapshot.commit_count, 1,
7467 "the authorized batch must publish exactly one commit"
7468 );
7469 assert!(
7470 !adapter.has_pending_publication(),
7471 "authorization must drain the staged publication horizon"
7472 );
7473 assert_eq!(
7474 adapter.wal.last_fsynced_frame_count(),
7475 fsynced_before,
7476 "deferred authorization must not claim or force an fsync"
7477 );
7478 adapter
7479 .begin_transaction(&cx)
7480 .expect("the next transaction must not see a stale Busy");
7481 }
7482
7483 #[test]
7484 fn test_commit_append_publishes_visibility_snapshot() {
7485 init_wal_publication_test_tracing();
7486 let cx = test_cx();
7487 let vfs = MemoryVfs::new();
7488 let mut adapter = make_adapter(&vfs, &cx);
7489
7490 let p1 = sample_page(0x41);
7491 let p2 = sample_page(0x42);
7492 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
7493 adapter.append_frame(&cx, 2, &p2, 2).expect("append commit");
7494 adapter.sync(&cx).expect("sync commit batch");
7497
7498 assert_eq!(
7499 adapter.published_snapshot.last_commit_frame,
7500 Some(1),
7501 "synced commit should publish the visible commit horizon"
7502 );
7503 assert_eq!(
7504 adapter.published_snapshot.commit_count, 1,
7505 "synced commit should track the visible WAL commit count"
7506 );
7507 assert_eq!(
7508 adapter.published_snapshot.page_index.len(),
7509 2,
7510 "published snapshot should track both committed pages"
7511 );
7512 assert_eq!(
7513 adapter.published_snapshot.page_index.get(&2),
7514 Some(&1),
7515 "published snapshot must map each page to its latest committed frame"
7516 );
7517 }
7518
7519 #[test]
7520 fn test_prepared_append_publishes_visibility_snapshot() {
7521 init_wal_publication_test_tracing();
7522 let cx = test_cx();
7523 let vfs = MemoryVfs::new();
7524 let mut adapter = make_adapter(&vfs, &cx);
7525
7526 let p1 = sample_page(0x51);
7527 let p2 = sample_page(0x52);
7528 let frames = [
7529 WalFrameRef {
7530 page_number: 1,
7531 page_data: &p1,
7532 db_size_if_commit: 0,
7533 },
7534 WalFrameRef {
7535 page_number: 2,
7536 page_data: &p2,
7537 db_size_if_commit: 2,
7538 },
7539 ];
7540 let mut prepared = adapter
7541 .prepare_append_frames(&frames)
7542 .expect("prepare append")
7543 .expect("prepared batch");
7544 adapter
7545 .append_prepared_frames(&cx, &mut prepared)
7546 .expect("append prepared");
7547 adapter.sync(&cx).expect("sync prepared commit batch");
7549
7550 assert_eq!(
7551 adapter.published_snapshot.last_commit_frame,
7552 Some(1),
7553 "synced prepared commit should publish the visible commit horizon"
7554 );
7555 assert_eq!(
7556 adapter.published_snapshot.commit_count, 1,
7557 "synced prepared commit should track the visible WAL commit count"
7558 );
7559 assert_eq!(
7560 adapter.published_snapshot.page_index.len(),
7561 2,
7562 "synced prepared commit should publish all committed pages"
7563 );
7564 assert_eq!(
7565 adapter.published_snapshot.page_index.get(&2),
7566 Some(&1),
7567 "prepared commit append must map each page to its latest committed frame"
7568 );
7569 }
7570
7571 #[test]
7572 fn test_commit_publication_refreshes_external_prefix_before_local_commit() {
7573 let cx = test_cx();
7574 let vfs = MemoryVfs::new();
7575
7576 let file_writer = open_wal_file(&vfs, &cx);
7577 let wal_writer =
7578 WalFile::create(&cx, file_writer, PAGE_SIZE, 0, test_salts()).expect("create WAL");
7579 let mut writer = WalBackendAdapter::new(wal_writer);
7580
7581 let file_follower = open_wal_file(&vfs, &cx);
7582 let wal_follower = WalFile::open(&cx, file_follower).expect("open WAL");
7583 let mut follower = WalBackendAdapter::new(wal_follower);
7584
7585 let p1 = sample_page(0x61);
7586 writer
7587 .append_frame(&cx, 1, &p1, 1)
7588 .expect("writer commit 1");
7589 writer.sync(&cx).expect("sync writer commit 1");
7590
7591 let p2 = sample_page(0x62);
7592 writer
7593 .append_frame(&cx, 2, &p2, 2)
7594 .expect("writer commit 2");
7595 writer.sync(&cx).expect("sync writer commit 2");
7596
7597 let p3 = sample_page(0x63);
7598 follower
7599 .append_frame(&cx, 3, &p3, 3)
7600 .expect("follower local commit");
7601
7602 assert_eq!(
7606 follower.published_snapshot.last_commit_frame,
7607 Some(1),
7608 "refresh-before-append must publish the external prefix only"
7609 );
7610 assert_eq!(
7611 follower.published_snapshot.commit_count, 2,
7612 "the staged local commit must not count until publication"
7613 );
7614 assert_eq!(
7615 follower.published_snapshot.page_index.get(&1),
7616 Some(&0),
7617 "refresh-before-append should preserve earlier committed pages"
7618 );
7619 assert_eq!(
7620 follower.published_snapshot.page_index.get(&2),
7621 Some(&1),
7622 "refresh-before-append should publish externally committed pages"
7623 );
7624 assert_eq!(
7625 follower.published_snapshot.page_index.get(&3),
7626 None,
7627 "the staged local page must stay out of the published map"
7628 );
7629
7630 follower.sync(&cx).expect("publish follower local commit");
7631 assert_eq!(
7632 follower.published_snapshot.last_commit_frame,
7633 Some(2),
7634 "publication must extend the map with the local commit"
7635 );
7636 assert_eq!(follower.published_snapshot.commit_count, 3);
7637 assert_eq!(
7638 follower.published_snapshot.page_index.get(&3),
7639 Some(&2),
7640 "published local commit extends the WAL visibility map"
7641 );
7642 assert_eq!(follower.read_page(&cx, 1).expect("read p1"), Some(p1));
7643 assert_eq!(follower.read_page(&cx, 2).expect("read p2"), Some(p2));
7644 assert_eq!(follower.read_page(&cx, 3).expect("read p3"), Some(p3));
7645 }
7646
7647 #[test]
7648 fn test_truncate_checkpoint_republishes_empty_generation_snapshot() {
7649 init_wal_publication_test_tracing();
7650 let cx = test_cx();
7651 let vfs = MemoryVfs::new();
7652 let mut adapter = make_adapter(&vfs, &cx);
7653 let mut writer = MockCheckpointPageWriter;
7654
7655 adapter
7656 .append_frame(&cx, 1, &sample_page(0x61), 1)
7657 .expect("append committed frame");
7658 adapter.sync(&cx).expect("sync committed frame");
7662 let before = adapter.published_snapshot();
7663 assert_eq!(before.last_commit_frame, Some(0));
7664 assert_eq!(before.commit_count, 1);
7665 assert_eq!(before.latest_frame_entries, 1);
7666
7667 let result = adapter
7668 .checkpoint(&cx, CheckpointMode::Truncate, &mut writer, 0, None)
7669 .expect("truncate checkpoint");
7670 assert!(result.completed);
7671 assert!(result.wal_was_reset);
7672
7673 let after = adapter.published_snapshot();
7674 assert_ne!(
7675 before.generation, after.generation,
7676 "truncate checkpoint should publish a new WAL generation"
7677 );
7678 assert_eq!(after.last_commit_frame, None);
7679 assert_eq!(after.commit_count, 0);
7680 assert_eq!(after.latest_frame_entries, 0);
7681 assert!(after.lookup_contract_is_authoritative());
7682 }
7683
7684 #[test]
7687 fn test_partial_index_falls_back_to_linear_scan() {
7688 init_wal_publication_test_tracing();
7689 let cx = test_cx();
7692 let vfs = MemoryVfs::new();
7693 let mut adapter = make_adapter(&vfs, &cx);
7694
7695 adapter.set_page_index_cap(2);
7698
7699 let p1 = sample_page(0x01);
7702 let p2 = sample_page(0x02);
7703 let p3 = sample_page(0x03);
7704 let p4 = sample_page(0x04);
7705 let p5 = sample_page(0x05);
7706
7707 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
7708 adapter.append_frame(&cx, 2, &p2, 0).expect("append p2");
7709 adapter.append_frame(&cx, 3, &p3, 0).expect("append p3");
7710 adapter.append_frame(&cx, 4, &p4, 0).expect("append p4");
7711 adapter
7712 .append_frame(&cx, 5, &p5, 5)
7713 .expect("append p5 (commit)");
7714 adapter.sync(&cx).expect("publish staged frames");
7715
7716 assert_eq!(
7718 adapter.read_page(&cx, 1).expect("read p1"),
7719 Some(p1),
7720 "indexed page should be found via HashMap"
7721 );
7722 assert_eq!(
7723 adapter.read_page(&cx, 2).expect("read p2"),
7724 Some(p2),
7725 "indexed page should be found via HashMap"
7726 );
7727
7728 assert_eq!(
7731 adapter.read_page(&cx, 3).expect("read p3"),
7732 Some(p3),
7733 "non-indexed page must be found via linear scan fallback"
7734 );
7735 assert_eq!(
7736 adapter.read_page(&cx, 4).expect("read p4"),
7737 Some(p4),
7738 "non-indexed page must be found via linear scan fallback"
7739 );
7740 assert_eq!(
7741 adapter.read_page(&cx, 5).expect("read p5"),
7742 Some(p5),
7743 "non-indexed page must be found via linear scan fallback"
7744 );
7745
7746 assert_eq!(
7748 adapter.read_page(&cx, 99).expect("read non-existent"),
7749 None,
7750 "non-existent page must return None even with partial index"
7751 );
7752
7753 assert!(
7755 adapter.published_snapshot.index_is_partial,
7756 "index_is_partial should be true when cap is exceeded"
7757 );
7758 }
7759
7760 #[test]
7761 fn test_partial_index_returns_latest_version_via_fallback() {
7762 let cx = test_cx();
7766 let vfs = MemoryVfs::new();
7767 let mut adapter = make_adapter(&vfs, &cx);
7768
7769 adapter.set_page_index_cap(1);
7771
7772 let old_p2 = sample_page(0xAA);
7773 let new_p2 = sample_page(0xBB);
7774
7775 adapter
7777 .append_frame(&cx, 1, &sample_page(0x01), 0)
7778 .expect("append p1");
7779 adapter
7781 .append_frame(&cx, 2, &old_p2, 0)
7782 .expect("append p2 old");
7783 adapter
7786 .append_frame(&cx, 2, &new_p2, 3)
7787 .expect("append p2 new (commit)");
7788 adapter.sync(&cx).expect("publish staged frames");
7789
7790 assert_eq!(
7792 adapter.read_page(&cx, 2).expect("read p2"),
7793 Some(new_p2),
7794 "backwards scan must return the most recent frame for the page"
7795 );
7796 }
7797
7798 #[test]
7799 fn test_lookup_contract_distinguishes_authoritative_and_fallback_paths() {
7800 init_wal_publication_test_tracing();
7801 let cx = test_cx();
7802 let vfs = MemoryVfs::new();
7803 let mut adapter = make_adapter(&vfs, &cx);
7804 adapter.set_page_index_cap(1);
7805
7806 let p1 = sample_page(0x01);
7807 let p2 = sample_page(0x02);
7808 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
7809 adapter
7810 .append_frame(&cx, 2, &p2, 2)
7811 .expect("append p2 commit");
7812 adapter.sync(&cx).expect("publish staged frames");
7813
7814 let last_commit = adapter
7815 .inner_mut()
7816 .expect("no staged batch blocks inner access")
7817 .last_commit_frame(&cx)
7818 .expect("last commit")
7819 .expect("commit exists");
7820 adapter
7821 .publish_visible_snapshot(&cx, Some(last_commit), "lookup_contract_test")
7822 .expect("build published snapshot");
7823 let snapshot = adapter.published_snapshot.clone();
7824
7825 assert_eq!(
7826 adapter
7827 .resolve_visible_frame(&cx, &snapshot, 1)
7828 .expect("resolve indexed page"),
7829 WalPageLookupResolution::AuthoritativeHit { frame_index: 0 }
7830 );
7831 assert_eq!(
7832 adapter
7833 .resolve_visible_frame(&cx, &snapshot, 2)
7834 .expect("resolve fallback page"),
7835 WalPageLookupResolution::PartialIndexFallbackHit { frame_index: 1 }
7836 );
7837 assert_eq!(
7838 adapter
7839 .resolve_visible_frame(&cx, &snapshot, 99)
7840 .expect("resolve missing page"),
7841 WalPageLookupResolution::PartialIndexFallbackMiss
7842 );
7843 }
7844
7845 #[test]
7846 fn test_lookup_contract_is_authoritative_by_default() {
7847 let cx = test_cx();
7848 let vfs = MemoryVfs::new();
7849 let mut adapter = make_adapter(&vfs, &cx);
7850
7851 let p1 = sample_page(0x11);
7852 let p2 = sample_page(0x22);
7853 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
7854 adapter
7855 .append_frame(&cx, 2, &p2, 2)
7856 .expect("append p2 commit");
7857 adapter.sync(&cx).expect("publish staged frames");
7858
7859 let last_commit = adapter
7860 .inner_mut()
7861 .expect("no staged batch blocks inner access")
7862 .last_commit_frame(&cx)
7863 .expect("last commit")
7864 .expect("commit exists");
7865 adapter
7866 .publish_visible_snapshot(&cx, Some(last_commit), "lookup_contract_default")
7867 .expect("build published snapshot");
7868 let snapshot = adapter.published_snapshot.clone();
7869
7870 assert!(
7871 !snapshot.index_is_partial,
7872 "default index should be authoritative"
7873 );
7874 assert_eq!(
7875 adapter
7876 .resolve_visible_frame(&cx, &snapshot, 1)
7877 .expect("resolve page 1"),
7878 WalPageLookupResolution::AuthoritativeHit { frame_index: 0 }
7879 );
7880 assert_eq!(
7881 adapter
7882 .resolve_visible_frame(&cx, &snapshot, 2)
7883 .expect("resolve page 2"),
7884 WalPageLookupResolution::AuthoritativeHit { frame_index: 1 }
7885 );
7886 assert_eq!(
7887 adapter
7888 .resolve_visible_frame(&cx, &snapshot, 99)
7889 .expect("resolve missing page"),
7890 WalPageLookupResolution::AuthoritativeMiss
7891 );
7892 }
7893
7894 #[test]
7895 fn test_committed_txns_since_page_uses_visible_frame_horizon() {
7896 let cx = test_cx();
7897 let vfs = MemoryVfs::new();
7898 let mut adapter = make_adapter(&vfs, &cx);
7899
7900 let p1 = sample_page(0x31);
7901 let p2 = sample_page(0x32);
7902 let p3 = sample_page(0x33);
7903
7904 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
7905 adapter.append_frame(&cx, 2, &p2, 2).expect("commit tx1");
7906 adapter.append_frame(&cx, 3, &p3, 0).expect("append p3");
7907 adapter.append_frame(&cx, 2, &p2, 3).expect("commit tx2");
7908 adapter.sync(&cx).expect("publish staged commits");
7911
7912 assert_eq!(
7913 adapter
7914 .committed_txns_since_page(&cx, 1)
7915 .expect("count txns since page 1"),
7916 1
7917 );
7918 assert_eq!(
7919 adapter
7920 .committed_txns_since_page(&cx, 2)
7921 .expect("count txns since page 2"),
7922 0
7923 );
7924 assert_eq!(
7925 adapter
7926 .committed_txns_since_page(&cx, 99)
7927 .expect("count txns since missing page"),
7928 2
7929 );
7930 assert_eq!(
7931 adapter
7932 .committed_txn_count(&cx)
7933 .expect("count visible transactions"),
7934 2
7935 );
7936 }
7937
7938 #[test]
7939 fn test_conflicting_pages_since_snapshot_detects_later_wal_writes() {
7940 let cx = test_cx();
7941 let vfs = MemoryVfs::new();
7942 let mut adapter = make_adapter(&vfs, &cx);
7943
7944 let p1 = sample_page(0x41);
7945 let p2_before = sample_page(0x42);
7946 let p2_after = sample_page(0x43);
7947 let p3 = sample_page(0x44);
7948
7949 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
7950 adapter
7951 .append_frame(&cx, 2, &p2_before, 2)
7952 .expect("commit tx1");
7953 adapter.sync(&cx).expect("publish staged frames");
7954 adapter
7955 .begin_transaction(&cx)
7956 .expect("pin transaction snapshot");
7957 let pinned = adapter
7958 .pinned_read_snapshot()
7959 .expect("transaction should expose pinned WAL snapshot");
7960 let conflict_snapshot = TransactionConflictSnapshot {
7961 generation: pinned.generation,
7962 last_commit_frame: pinned.last_commit_frame,
7963 commit_count: pinned.commit_count,
7964 snapshot_db_size: 0,
7965 };
7966
7967 adapter
7968 .append_frame(&cx, 3, &p3, 0)
7969 .expect("append unrelated later page");
7970 adapter
7971 .append_frame(&cx, 2, &p2_after, 3)
7972 .expect("commit later page 2 update");
7973 adapter.sync(&cx).expect("publish later commit");
7976
7977 let conflicts = adapter
7978 .conflicting_pages_since_snapshot(&cx, conflict_snapshot, &[2, 99], &[])
7979 .expect("conflict check should scan later committed frames");
7980 assert_eq!(conflicts, vec![2]);
7981
7982 let unrelated = adapter
7983 .conflicting_pages_since_snapshot(&cx, conflict_snapshot, &[99], &[])
7984 .expect("unrelated page should stay conflict-free");
7985 assert!(unrelated.is_empty());
7986 }
7987
7988 #[test]
7991 fn test_checkpoint_adapter_write_page() {
7992 let cx = test_cx();
7993 let mut writer = MockCheckpointPageWriter;
7994 let mut adapter = CheckpointTargetAdapterRef {
7995 writer: &mut writer,
7996 };
7997
7998 let page_no = PageNumber::new(1).expect("valid page number");
7999 adapter
8000 .write_page(&cx, page_no, &[0u8; 4096])
8001 .expect("write_page");
8002 }
8003
8004 #[test]
8005 fn test_checkpoint_adapter_truncate_db() {
8006 let cx = test_cx();
8007 let mut writer = MockCheckpointPageWriter;
8008 let mut adapter = CheckpointTargetAdapterRef {
8009 writer: &mut writer,
8010 };
8011
8012 adapter.truncate_db(&cx, 10).expect("truncate_db");
8013 }
8014
8015 #[test]
8016 fn test_checkpoint_adapter_sync_db() {
8017 let cx = test_cx();
8018 let mut writer = MockCheckpointPageWriter;
8019 let mut adapter = CheckpointTargetAdapterRef {
8020 writer: &mut writer,
8021 };
8022
8023 adapter.sync_db(&cx).expect("sync_db");
8024 }
8025
8026 #[test]
8027 fn test_checkpoint_adapter_as_dyn_target() {
8028 let cx = test_cx();
8029 let mut writer = MockCheckpointPageWriter;
8030 let mut adapter = CheckpointTargetAdapterRef {
8031 writer: &mut writer,
8032 };
8033
8034 let target: &mut dyn CheckpointTarget = &mut adapter;
8036 let page_no = PageNumber::new(3).expect("valid page number");
8037 target
8038 .write_page(&cx, page_no, &[0u8; 4096])
8039 .expect("write via dyn");
8040 target.truncate_db(&cx, 5).expect("truncate via dyn");
8041 target.sync_db(&cx).expect("sync via dyn");
8042 }
8043}