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
2099const LEGACY_IDENTITYLESS_CERTIFICATE_RECORD_VERSION: u16 = 3;
2103
2104fn durable_certificate_is_legacy_identityless(bytes: &[u8]) -> bool {
2115 if !bytes.starts_with(&PARALLEL_WAL_DURABLE_CERTIFICATE_MAGIC) {
2116 return false;
2117 }
2118 match bytes.get(8..10) {
2119 Some(version_bytes) => {
2120 u16::from_le_bytes([version_bytes[0], version_bytes[1]])
2121 == LEGACY_IDENTITYLESS_CERTIFICATE_RECORD_VERSION
2122 }
2123 None => false,
2124 }
2125}
2126
2127fn decode_durable_certificate_record(
2128 bytes: &[u8],
2129 location: &str,
2130) -> Result<ParallelWalDurableCertificateRecord> {
2131 ParallelWalDurableCertificateRecord::from_bytes(bytes).map_err(|error| {
2132 FrankenError::WalCorrupt {
2133 detail: format!("parallel WAL certificate {location} is invalid: {error}"),
2134 }
2135 })
2136}
2137
2138fn validate_incomplete_certificate_suffix(bytes: &[u8], anchored: bool) -> Result<()> {
2139 if bytes.is_empty() {
2140 return Ok(());
2141 }
2142 if bytes.len() > PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE {
2143 return Err(FrankenError::WalCorrupt {
2144 detail: format!(
2145 "parallel WAL certificate torn suffix exceeds {} bytes",
2146 PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE
2147 ),
2148 });
2149 }
2150
2151 if bytes.len() < PARALLEL_WAL_DURABLE_CERTIFICATE_MAGIC.len() {
2152 if anchored || PARALLEL_WAL_DURABLE_CERTIFICATE_MAGIC.starts_with(bytes) {
2157 return Ok(());
2158 }
2159 return Err(FrankenError::WalCorrupt {
2160 detail: "parallel WAL certificate sidecar starts with non-record garbage".to_owned(),
2161 });
2162 }
2163 if !bytes.starts_with(&PARALLEL_WAL_DURABLE_CERTIFICATE_MAGIC) {
2164 return Err(FrankenError::WalCorrupt {
2165 detail: "parallel WAL certificate suffix does not start at a record boundary"
2166 .to_owned(),
2167 });
2168 }
2169 if bytes.len() < 10 {
2170 return Ok(());
2171 }
2172 let version = u16::from_le_bytes([bytes[8], bytes[9]]);
2173 if version != fsqlite_wal::PARALLEL_WAL_DURABLE_CERTIFICATE_RECORD_VERSION {
2174 return Err(FrankenError::WalCorrupt {
2175 detail: format!(
2176 "parallel WAL certificate suffix has unsupported record version {version}"
2177 ),
2178 });
2179 }
2180 if bytes.len() < DURABLE_CERTIFICATE_RECORD_HEADER_SIZE {
2181 return Ok(());
2182 }
2183 let declared_len =
2184 durable_certificate_declared_len(bytes).ok_or_else(|| FrankenError::WalCorrupt {
2185 detail: "parallel WAL certificate suffix length exceeds usize".to_owned(),
2186 })?;
2187 if !(MIN_DURABLE_CERTIFICATE_RECORD_SIZE..=PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE)
2188 .contains(&declared_len)
2189 {
2190 return Err(FrankenError::WalCorrupt {
2191 detail: format!(
2192 "parallel WAL certificate suffix declares invalid record length {declared_len}"
2193 ),
2194 });
2195 }
2196 if bytes.len() < declared_len {
2197 return Ok(());
2198 }
2199
2200 decode_durable_certificate_record(&bytes[..declared_len], "suffix")?;
2205 Err(FrankenError::WalCorrupt {
2206 detail:
2207 "parallel WAL certificate sidecar contains a complete record outside the footer chain"
2208 .to_owned(),
2209 })
2210}
2211
2212fn combine_sidecar_io_results<const N: usize>(
2213 context: &str,
2214 results: [(&str, Result<()>); N],
2215) -> Result<()> {
2216 let failures = results
2217 .into_iter()
2218 .filter_map(|(stage, result)| result.err().map(|error| (stage, error)))
2219 .collect::<Vec<_>>();
2220 if failures.is_empty() {
2221 return Ok(());
2222 }
2223 if failures.len() == 1 {
2224 return failures
2225 .into_iter()
2226 .next()
2227 .map_or(Ok(()), |(_, error)| Err(error));
2228 }
2229 let details = failures
2230 .iter()
2231 .map(|(stage, error)| format!("{stage}={error}"))
2232 .collect::<Vec<_>>()
2233 .join("; ");
2234 Err(FrankenError::internal(format!("{context}: {details}")))
2235}
2236
2237pub struct PathRefreshingWalBackend<V: Vfs>
2247where
2248 V::File: Send + Sync + 'static,
2249{
2250 vfs: V,
2251 db_path: PathBuf,
2252 wal_path: PathBuf,
2253 page_size: u32,
2254 create_missing: bool,
2255 #[cfg(all(feature = "native", any(unix, windows)))]
2256 namespace_binding: Option<Arc<DatabaseNamespaceBinding>>,
2257 cached_verification_db: Option<V::File>,
2276 cached_certificate_read: std::sync::Mutex<Option<V::File>>,
2293 db_file_identity: Option<[u8; 16]>,
2307 inner: WalBackendAdapter<V::File>,
2308}
2309
2310impl<V> PathRefreshingWalBackend<V>
2311where
2312 V: Vfs + 'static,
2313 V::File: Send + Sync + 'static,
2314{
2315 #[must_use]
2316 pub fn new(
2317 vfs: V,
2318 db_path: impl AsRef<Path>,
2319 wal_path: impl AsRef<Path>,
2320 page_size: u32,
2321 wal: WalFile<V::File>,
2322 create_missing: bool,
2323 #[cfg(all(feature = "native", any(unix, windows)))] namespace_binding: Option<
2324 Arc<DatabaseNamespaceBinding>,
2325 >,
2326 ) -> Self {
2327 Self {
2328 vfs,
2329 db_path: db_path.as_ref().to_path_buf(),
2330 wal_path: wal_path.as_ref().to_path_buf(),
2331 page_size,
2332 create_missing,
2333 #[cfg(all(feature = "native", any(unix, windows)))]
2334 namespace_binding,
2335 cached_verification_db: None,
2336 cached_certificate_read: std::sync::Mutex::new(None),
2337 db_file_identity: None,
2338 inner: WalBackendAdapter::new(wal),
2339 }
2340 }
2341
2342 #[must_use]
2343 pub fn into_inner(self) -> WalBackendAdapter<V::File> {
2344 self.inner
2345 }
2346
2347 fn replace_inner(&mut self, cx: &Cx, wal: WalFile<V::File>) -> Result<()> {
2355 if self.inner.has_pending_publication() {
2356 let cleanup_cx = cx.create_child();
2357 let _cleanup_mask = cleanup_cx.masked();
2358 let _ = wal.close(&cleanup_cx);
2359 return Err(FrankenError::Busy);
2360 }
2361 let old = std::mem::replace(&mut self.inner, WalBackendAdapter::new(wal));
2362 if let Some(mut stale) = self
2366 .cached_certificate_read
2367 .get_mut()
2368 .unwrap_or_else(std::sync::PoisonError::into_inner)
2369 .take()
2370 {
2371 let _ = stale.close(cx);
2372 }
2373 let old_wal = old.into_inner()?;
2374 let _ = old_wal.close(cx);
2375 Ok(())
2376 }
2377
2378 async fn create_replacement_wal(&self, cx: &Cx) -> Result<WalFile<V::File>> {
2379 let flags = VfsOpenFlags::READWRITE | VfsOpenFlags::CREATE | VfsOpenFlags::WAL;
2380 let (file, _) = self.vfs.open(cx, Some(&self.wal_path), flags)?;
2381 let wal = WalFile::create(cx, file, self.page_size, 0, WalSalts::generate()).await?;
2384 if let Err(error) = self.vfs.sync_parent_directory(cx, &self.wal_path) {
2385 let cleanup_cx = cx.create_child();
2386 let _cleanup_mask = cleanup_cx.masked();
2387 let _ = wal.close(&cleanup_cx);
2388 return Err(error);
2389 }
2390 Ok(wal)
2391 }
2392
2393 async fn replace_with_created_wal(&mut self, cx: &Cx) -> Result<()> {
2394 let wal = self.create_replacement_wal(cx).await?;
2395 self.replace_inner(cx, wal)
2396 }
2397
2398 async fn open_replacement_wal(&self, cx: &Cx, path_file: V::File) -> Result<WalFile<V::File>> {
2399 let wal = WalFile::open(cx, path_file).await?;
2400 if u32::try_from(wal.page_size()).ok() != Some(self.page_size) {
2401 let actual_page_size = wal.page_size();
2402 let expected_page_size = self.page_size;
2403 let _ = wal.close(cx);
2404 return Err(FrankenError::WalCorrupt {
2405 detail: format!(
2406 "WAL page size {actual_page_size} does not match database page size {expected_page_size} during path refresh"
2407 ),
2408 });
2409 }
2410 Ok(wal)
2411 }
2412
2413 async fn path_header_matches_current_handle(
2414 &self,
2415 cx: &Cx,
2416 path_file: &V::File,
2417 ) -> Result<bool> {
2418 let mut header_buf = [0_u8; WAL_HEADER_SIZE];
2419 let bytes_read = path_file.read(cx, &mut header_buf, 0).await?;
2420 if bytes_read < WAL_HEADER_SIZE {
2421 return Ok(false);
2422 }
2423
2424 let path_header = WalHeader::from_bytes(&header_buf)?;
2425 if !validate_wal_header_checksum(&header_buf, path_header.big_endian_checksum())? {
2426 return Err(FrankenError::WalCorrupt {
2427 detail: "WAL header checksum mismatch during path refresh".to_owned(),
2428 });
2429 }
2430
2431 let current_header = self.inner.inner().header();
2432 Ok(path_header.magic == current_header.magic
2433 && path_header.format_version == current_header.format_version
2434 && path_header.page_size == current_header.page_size
2435 && path_header.checkpoint_seq == current_header.checkpoint_seq
2436 && path_header.salts == current_header.salts)
2437 }
2438
2439 async fn ensure_db_file_identity_captured(&mut self, cx: &Cx) {
2465 if self.db_file_identity.is_some() {
2466 return;
2467 }
2468 let page_size = match usize::try_from(self.page_size) {
2469 Ok(size) if size >= 92 => size,
2470 _ => return,
2471 };
2472 let db_file = match self.cached_verification_db.take() {
2473 Some(cached) => cached,
2474 None => {
2475 let main_db_flags = VfsOpenFlags::READWRITE | VfsOpenFlags::MAIN_DB;
2476 match self.vfs.open(cx, Some(&self.db_path), main_db_flags) {
2477 Ok((file, _)) => file,
2478 Err(_) => return,
2481 }
2482 }
2483 };
2484 let mut page_one = vec![0_u8; page_size];
2485 match db_file.read(cx, &mut page_one, 0).await {
2486 Ok(bytes_read) if bytes_read >= 92 => {
2487 let mut id = [0_u8; 16];
2488 id.copy_from_slice(&page_one[76..92]);
2489 self.db_file_identity = Some(id);
2490 }
2491 _ => {}
2493 }
2494 self.cached_verification_db = Some(db_file);
2496 }
2497
2498 fn db_file_identity_matches(&self, record: &ParallelWalDurableCertificateRecord) -> bool {
2505 match self.db_file_identity {
2506 Some(id) if id != [0u8; 16] => record.db_file_id != [0u8; 16] && record.db_file_id == id,
2507 _ => false,
2508 }
2509 }
2510
2511 fn db_file_identity_rejects_record(
2526 &self,
2527 record: &ParallelWalDurableCertificateRecord,
2528 ) -> bool {
2529 matches!(self.db_file_identity, Some(id) if id != [0u8; 16])
2533 && !self.db_file_identity_matches(record)
2534 }
2535
2536 fn db_file_id_for_written_certificate(&self) -> [u8; 16] {
2541 self.db_file_identity.unwrap_or_default()
2542 }
2543
2544 async fn conflicts_after_generation_change(
2545 &mut self,
2546 cx: &Cx,
2547 page_numbers: &[u32],
2548 page_baselines: &[TransactionConflictPageBaseline],
2549 ) -> Vec<u32> {
2550 let mut candidates = page_numbers
2551 .iter()
2552 .copied()
2553 .filter(|page| *page != 0)
2554 .collect::<Vec<_>>();
2555 candidates.sort_unstable();
2556 candidates.dedup();
2557 if candidates.is_empty() {
2558 return Vec::new();
2559 }
2560
2561 let mut baselines = HashMap::<u32, [u8; 32]>::new();
2562 let mut ambiguous_baselines = HashSet::<u32>::new();
2563 for baseline in page_baselines {
2564 if baseline.page_number == 0 {
2565 continue;
2566 }
2567 if let Some(previous) = baselines.insert(baseline.page_number, baseline.page_hash)
2568 && previous != baseline.page_hash
2569 {
2570 ambiguous_baselines.insert(baseline.page_number);
2571 }
2572 }
2573
2574 let mut db_file = match self.cached_verification_db.take() {
2581 Some(cached) => cached,
2582 None => {
2583 let main_db_flags = VfsOpenFlags::READWRITE | VfsOpenFlags::MAIN_DB;
2584 match self.vfs.open(cx, Some(&self.db_path), main_db_flags) {
2585 Ok((file, _)) => file,
2586 Err(_) => return candidates,
2587 }
2588 }
2589 };
2590 let page_size = match usize::try_from(self.page_size) {
2591 Ok(page_size) if page_size > 0 => page_size,
2592 _ => {
2593 let _ = db_file.close(cx);
2594 return candidates;
2595 }
2596 };
2597
2598 let mut page_one = vec![0_u8; page_size];
2601 let page_one_read = match db_file.read(cx, &mut page_one, 0).await {
2602 Ok(bytes_read) => bytes_read,
2603 Err(_) => {
2604 let _ = db_file.close(cx);
2605 return candidates;
2606 }
2607 };
2608 let header_page_size =
2609 (page_one_read == page_size).then(|| sqlite_database_header_page_size(&page_one));
2610 if header_page_size.flatten() != Some(self.page_size) {
2611 let _ = db_file.close(cx);
2612 return candidates;
2613 }
2614 let committed_page_count =
2625 u32::from_be_bytes([page_one[28], page_one[29], page_one[30], page_one[31]]);
2626
2627 let mut conflicts = Vec::new();
2628 for &page_number in &candidates {
2629 let Some(expected_hash) = baselines.get(&page_number).copied() else {
2630 if committed_page_count > 0 && page_number > committed_page_count {
2631 continue;
2632 }
2633 conflicts.push(page_number);
2634 continue;
2635 };
2636 if ambiguous_baselines.contains(&page_number) {
2637 conflicts.push(page_number);
2638 continue;
2639 }
2640
2641 let current_page = match self.inner.read_page(cx, page_number).await {
2642 Ok(Some(page)) if page.len() == page_size => page,
2643 Ok(Some(_)) | Err(_) => {
2644 conflicts.push(page_number);
2645 continue;
2646 }
2647 Ok(None) => {
2648 let mut page = vec![0_u8; page_size];
2649 let page_offset = u64::from(page_number.saturating_sub(1))
2650 .saturating_mul(u64::from(self.page_size));
2651 match db_file.read(cx, &mut page, page_offset).await {
2652 Ok(bytes_read) if bytes_read == page_size => page,
2653 Ok(_) | Err(_) => {
2654 conflicts.push(page_number);
2655 continue;
2656 }
2657 }
2658 }
2659 };
2660 let current_hash = *blake3::hash(¤t_page).as_bytes();
2661 if current_hash != expected_hash {
2662 conflicts.push(page_number);
2663 }
2664 }
2665
2666 self.cached_verification_db = Some(db_file);
2671 conflicts.sort_unstable();
2672 conflicts.dedup();
2673 conflicts
2674 }
2675
2676 async fn ensure_current_wal_path(&mut self, cx: &Cx) -> Result<()> {
2677 #[cfg(all(feature = "native", any(unix, windows)))]
2678 if let Some(binding) = &self.namespace_binding {
2679 binding.validate_path_identity()?;
2680 }
2681 self.ensure_db_file_identity_captured(cx).await;
2686 if !self.vfs.access(cx, &self.wal_path, AccessFlags::EXISTS)? {
2687 if self.create_missing {
2688 return self.replace_with_created_wal(cx).await;
2689 }
2690 return Ok(());
2691 }
2692
2693 let flags = VfsOpenFlags::READWRITE | VfsOpenFlags::WAL;
2694 let (mut path_file, _) = self.vfs.open(cx, Some(&self.wal_path), flags)?;
2695 let path_size = path_file.file_size(cx)?;
2696 if path_size < u64::try_from(WAL_HEADER_SIZE).unwrap_or(32) {
2697 let _ = path_file.close(cx);
2698 if self.create_missing {
2699 return self.replace_with_created_wal(cx).await;
2700 }
2701 return Ok(());
2702 }
2703
2704 let current_size = self.inner.inner().file().file_size(cx).unwrap_or(u64::MAX);
2705 let path_matches_current = if path_size == current_size {
2706 match self
2707 .path_header_matches_current_handle(cx, &path_file)
2708 .await
2709 {
2710 Ok(matches) => matches,
2711 Err(err) => {
2712 let _ = path_file.close(cx);
2713 return Err(err);
2714 }
2715 }
2716 } else {
2717 false
2718 };
2719 if !path_matches_current {
2720 let wal = self.open_replacement_wal(cx, path_file).await?;
2721 self.replace_inner(cx, wal)?;
2722 } else {
2723 let _ = path_file.close(cx);
2724 }
2725 Ok(())
2726 }
2727
2728 fn certificate_sidecar_path(&self) -> PathBuf {
2729 let mut path = self.wal_path.as_os_str().to_owned();
2730 path.push("-cert");
2731 PathBuf::from(path)
2732 }
2733
2734 fn certificate_checkpoint_handoff_path(&self) -> PathBuf {
2735 let mut path = self.wal_path.as_os_str().to_owned();
2736 path.push("-cert-head");
2737 PathBuf::from(path)
2738 }
2739
2740 async fn read_certificate_sidecar_exact(
2741 file: &V::File,
2742 cx: &Cx,
2743 offset: u64,
2744 len: usize,
2745 location: &str,
2746 ) -> Result<Vec<u8>> {
2747 let mut bytes = vec![0_u8; len];
2748 let bytes_read = file.read(cx, &mut bytes, offset).await?;
2749 if bytes_read != len {
2750 return Err(FrankenError::WalCorrupt {
2751 detail: format!(
2752 "parallel WAL certificate {location} at offset {offset} was short-read: got {bytes_read} of {len}"
2753 ),
2754 });
2755 }
2756 Ok(bytes)
2757 }
2758
2759 async fn read_certificate_record_ending_at(
2760 file: &V::File,
2761 cx: &Cx,
2762 record_end: u64,
2763 ) -> Result<(u64, ParallelWalDurableCertificateRecord)> {
2764 let footer_size =
2765 u64::try_from(ParallelWalDurableCertificateRecord::LENGTH_FOOTER_SIZE).unwrap_or(4);
2766 let footer_offset = record_end.checked_sub(footer_size).ok_or_else(|| {
2767 FrankenError::WalCorrupt {
2768 detail: format!(
2769 "parallel WAL certificate record ending at {record_end} has no length footer"
2770 ),
2771 }
2772 })?;
2773 let footer = Self::read_certificate_sidecar_exact(
2774 file,
2775 cx,
2776 footer_offset,
2777 ParallelWalDurableCertificateRecord::LENGTH_FOOTER_SIZE,
2778 "length footer",
2779 )
2780 .await?;
2781 let record_len = usize::try_from(u32::from_le_bytes([
2782 footer[0], footer[1], footer[2], footer[3],
2783 ]))
2784 .map_err(|_| FrankenError::WalCorrupt {
2785 detail: "parallel WAL certificate footer length exceeds usize".to_owned(),
2786 })?;
2787 if !(MIN_DURABLE_CERTIFICATE_RECORD_SIZE..=PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE)
2788 .contains(&record_len)
2789 {
2790 return Err(FrankenError::WalCorrupt {
2791 detail: format!(
2792 "parallel WAL certificate footer declares invalid record length {record_len}"
2793 ),
2794 });
2795 }
2796 let record_len_u64 = u64::try_from(record_len).map_err(|_| FrankenError::WalCorrupt {
2797 detail: "parallel WAL certificate record length exceeds u64".to_owned(),
2798 })?;
2799 let record_start =
2800 record_end
2801 .checked_sub(record_len_u64)
2802 .ok_or_else(|| FrankenError::WalCorrupt {
2803 detail: format!(
2804 "parallel WAL certificate record length {record_len} exceeds end offset {record_end}"
2805 ),
2806 })?;
2807 let bytes =
2808 Self::read_certificate_sidecar_exact(file, cx, record_start, record_len, "record")
2809 .await?;
2810 let record = decode_durable_certificate_record(&bytes, "record")?;
2811 Ok((record_start, record))
2812 }
2813
2814 async fn prepare_certificate_sidecar_for_append(file: &mut V::File, cx: &Cx) -> Result<u64> {
2821 let file_size = file.file_size(cx)?;
2822 if file_size == 0 {
2823 return Ok(0);
2824 }
2825
2826 let footer_size =
2827 u64::try_from(ParallelWalDurableCertificateRecord::LENGTH_FOOTER_SIZE).unwrap_or(4);
2828 if file_size >= footer_size {
2829 let footer = Self::read_certificate_sidecar_exact(
2830 file,
2831 cx,
2832 file_size - footer_size,
2833 ParallelWalDurableCertificateRecord::LENGTH_FOOTER_SIZE,
2834 "append-boundary length footer",
2835 )
2836 .await?;
2837 let record_len = usize::try_from(u32::from_le_bytes([
2838 footer[0], footer[1], footer[2], footer[3],
2839 ]))
2840 .unwrap_or(usize::MAX);
2841 let record_len_u64 = u64::try_from(record_len).unwrap_or(u64::MAX);
2842 if (MIN_DURABLE_CERTIFICATE_RECORD_SIZE
2843 ..=PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE)
2844 .contains(&record_len)
2845 && record_len_u64 <= file_size
2846 {
2847 let record_start = file_size - record_len_u64;
2848 let bytes = Self::read_certificate_sidecar_exact(
2849 file,
2850 cx,
2851 record_start,
2852 record_len,
2853 "append-boundary record",
2854 )
2855 .await?;
2856 if bytes.starts_with(&PARALLEL_WAL_DURABLE_CERTIFICATE_MAGIC)
2857 || durable_certificate_declares_len(&bytes, record_len)
2858 {
2859 decode_durable_certificate_record(&bytes, "append-boundary record")?;
2860 return Ok(file_size);
2861 }
2862 }
2863 }
2864
2865 let recovery_window_size =
2869 PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE.saturating_mul(2);
2870 let recovery_window_size_u64 = u64::try_from(recovery_window_size).unwrap_or(u64::MAX);
2871 let tail_offset = file_size.saturating_sub(recovery_window_size_u64);
2872 let tail_len =
2873 usize::try_from(file_size - tail_offset).map_err(|_| FrankenError::WalCorrupt {
2874 detail: "parallel WAL certificate append-repair window exceeds usize".to_owned(),
2875 })?;
2876 let tail = Self::read_certificate_sidecar_exact(
2877 file,
2878 cx,
2879 tail_offset,
2880 tail_len,
2881 "append-repair window",
2882 )
2883 .await?;
2884 let minimum_candidate_end = tail
2885 .len()
2886 .saturating_sub(PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE)
2887 .max(ParallelWalDurableCertificateRecord::LENGTH_FOOTER_SIZE);
2888 let mut anchor_end = None;
2889 for candidate_end in (minimum_candidate_end..tail.len()).rev() {
2890 let footer_start =
2891 candidate_end - ParallelWalDurableCertificateRecord::LENGTH_FOOTER_SIZE;
2892 let footer = &tail[footer_start..candidate_end];
2893 let record_len = usize::try_from(u32::from_le_bytes([
2894 footer[0], footer[1], footer[2], footer[3],
2895 ]))
2896 .unwrap_or(usize::MAX);
2897 if !(MIN_DURABLE_CERTIFICATE_RECORD_SIZE
2898 ..=PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE)
2899 .contains(&record_len)
2900 || record_len > candidate_end
2901 {
2902 continue;
2903 }
2904 let record_start = candidate_end - record_len;
2905 let record_bytes = &tail[record_start..candidate_end];
2906 if !record_bytes.starts_with(&PARALLEL_WAL_DURABLE_CERTIFICATE_MAGIC)
2907 || !durable_certificate_declares_len(record_bytes, record_len)
2908 {
2909 continue;
2910 }
2911 if ParallelWalDurableCertificateRecord::from_bytes(record_bytes).is_ok() {
2912 anchor_end = Some(candidate_end);
2913 break;
2914 }
2915 }
2916
2917 let safe_end = if let Some(anchor_end) = anchor_end {
2918 validate_incomplete_certificate_suffix(&tail[anchor_end..], true)?;
2919 tail_offset
2920 .checked_add(u64::try_from(anchor_end).unwrap_or(u64::MAX))
2921 .ok_or_else(|| FrankenError::WalCorrupt {
2922 detail: "parallel WAL certificate append-repair boundary overflow".to_owned(),
2923 })?
2924 } else {
2925 if file_size
2926 > u64::try_from(PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE)
2927 .unwrap_or(u64::MAX)
2928 {
2929 return Err(FrankenError::WalCorrupt {
2930 detail: format!(
2931 "parallel WAL certificate sidecar has no valid append boundary within its bounded {}-byte recovery suffix",
2932 PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE
2933 ),
2934 });
2935 }
2936 validate_incomplete_certificate_suffix(&tail, false)?;
2937 0
2938 };
2939
2940 if safe_end < file_size {
2941 file.truncate(cx, safe_end)?;
2942 }
2943 Ok(safe_end)
2944 }
2945
2946 async fn append_durable_certificate_record(
2947 &self,
2948 cx: &Cx,
2949 certificate: &ParallelWalCommitCertificate,
2950 wal_frame_start: u64,
2951 wal_frame_end: u64,
2952 sync: bool,
2953 ) -> Result<()> {
2954 self.append_durable_certificate_record_with_completion(
2955 cx,
2956 certificate,
2957 wal_frame_start,
2958 wal_frame_end,
2959 sync,
2960 None,
2961 )
2962 .await
2963 }
2964
2965 async fn append_durable_certificate_record_with_completion(
2966 &self,
2967 cx: &Cx,
2968 certificate: &ParallelWalCommitCertificate,
2969 wal_frame_start: u64,
2970 wal_frame_end: u64,
2971 sync: bool,
2972 completion: Option<&VfsWriteCompletion>,
2973 ) -> Result<()> {
2974 let mut preflight = WalWriteCompletionPreflight::new(completion);
2975 let expected_frame_start = u64::try_from(self.inner.frame_count())
2976 .unwrap_or(u64::MAX)
2977 .saturating_add(1);
2978 if wal_frame_start != expected_frame_start {
2979 return Err(FrankenError::internal(format!(
2980 "parallel WAL certificate starts at frame {wal_frame_start}, expected {expected_frame_start}"
2981 )));
2982 }
2983 let record = ParallelWalDurableCertificateRecord::new(
2984 self.inner.inner().generation_identity(),
2985 wal_frame_start,
2986 wal_frame_end,
2987 self.db_file_id_for_written_certificate(),
2991 certificate.clone(),
2992 )
2993 .map_err(|error| {
2994 FrankenError::internal(format!(
2995 "could not encode parallel WAL durability certificate: {error}"
2996 ))
2997 })?;
2998 let record_bytes = record.to_bytes();
2999 if record_bytes.len() > PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE {
3000 return Err(FrankenError::WalCorrupt {
3001 detail: format!(
3002 "parallel WAL certificate record is {} bytes; maximum is {}",
3003 record_bytes.len(),
3004 PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE
3005 ),
3006 });
3007 }
3008 let certificate_path = self.certificate_sidecar_path();
3009 let existed = self
3010 .vfs
3011 .access(cx, &certificate_path, AccessFlags::EXISTS)?;
3012 let flags = VfsOpenFlags::READWRITE | VfsOpenFlags::CREATE | VfsOpenFlags::WAL;
3013 let (mut file, _) = self.vfs.open(cx, Some(&certificate_path), flags)?;
3014 let append_offset = Self::prepare_certificate_sidecar_for_append(&mut file, cx).await?;
3015 preflight.hand_off();
3016 drop(preflight);
3017 let write_result = if let Some(completion) = completion {
3018 file.write_tracked(cx, &record_bytes, append_offset, completion.clone())
3019 .await
3020 } else {
3021 file.write(cx, &record_bytes, append_offset).await
3022 };
3023 if let Err(write_error) = write_result {
3024 let cleanup_cx = cx.create_child();
3029 let _cleanup_mask = cleanup_cx.masked();
3030 let cleanup_result = file.truncate(&cleanup_cx, append_offset);
3031 let close_result = file.close(&cleanup_cx);
3032 return combine_sidecar_io_results(
3033 "parallel WAL certificate append cleanup failed",
3034 [
3035 ("write", Err(write_error)),
3036 ("truncate", cleanup_result),
3037 ("close", close_result),
3038 ],
3039 );
3040 }
3041
3042 let finalization_cx = cx.create_child();
3046 let _finalization_mask = finalization_cx.masked();
3047 let sync_result = if sync {
3048 file.durable_sync(&finalization_cx, SyncKind::FullDurable)
3049 } else {
3050 Ok(())
3051 };
3052 let directory_sync_result = if sync && !existed && sync_result.is_ok() {
3053 self.vfs
3054 .sync_parent_directory(&finalization_cx, &certificate_path)
3055 } else {
3056 Ok(())
3057 };
3058 let close_result = file.close(&finalization_cx);
3059 combine_sidecar_io_results(
3060 "parallel WAL certificate append finalization failed",
3061 [
3062 ("file_sync", sync_result),
3063 ("directory_sync", directory_sync_result),
3064 ("close", close_result),
3065 ],
3066 )
3067 }
3068
3069 async fn reconcile_certificate_sidecar_record(
3070 &self,
3071 cx: &Cx,
3072 expected: &ParallelWalDurableCertificateRecord,
3073 remove_expected_orphan: bool,
3074 sync: bool,
3075 ) -> Result<bool> {
3076 let certificate_path = self.certificate_sidecar_path();
3077 if !self
3078 .vfs
3079 .access(cx, &certificate_path, AccessFlags::EXISTS)?
3080 {
3081 return Ok(false);
3082 }
3083
3084 let flags = VfsOpenFlags::READWRITE | VfsOpenFlags::WAL;
3085 let (mut file, _) = self.vfs.open(cx, Some(&certificate_path), flags)?;
3086 let reconciliation_result = async {
3087 let original_size = file.file_size(cx)?;
3088 let safe_end = Self::prepare_certificate_sidecar_for_append(&mut file, cx).await?;
3089 let latest = if safe_end == 0 {
3090 None
3091 } else {
3092 Some(Self::read_certificate_record_ending_at(&file, cx, safe_end).await?)
3093 };
3094 let latest_is_expected = latest
3095 .as_ref()
3096 .is_some_and(|(_, record)| record == expected);
3097 let sidecar_changed = if remove_expected_orphan
3098 && let Some((record_start, _)) = latest.as_ref()
3099 && latest_is_expected
3100 {
3101 file.truncate(cx, *record_start)?;
3102 true
3103 } else {
3104 safe_end != original_size
3105 };
3106 if sync && (latest_is_expected || sidecar_changed) {
3107 file.durable_sync(cx, SyncKind::FullDurable)?;
3108 }
3109 if sync && latest_is_expected && !remove_expected_orphan {
3110 self.vfs.sync_parent_directory(cx, &certificate_path)?;
3113 }
3114 Ok(latest_is_expected)
3115 }
3116 .await;
3117
3118 let cleanup_cx = cx.create_child();
3119 let _cleanup_mask = cleanup_cx.masked();
3120 let close_result = file.close(&cleanup_cx);
3121 match (reconciliation_result, close_result) {
3122 (Ok(latest_is_expected), Ok(())) => Ok(latest_is_expected),
3123 (Err(error), Ok(())) | (Ok(_), Err(error)) => Err(error),
3124 (Err(reconciliation_error), Err(close_error)) => Err(FrankenError::internal(format!(
3125 "parallel WAL certificate reconciliation failed and close also failed: reconciliation={reconciliation_error}; close={close_error}"
3126 ))),
3127 }
3128 }
3129
3130 async fn persist_checkpoint_certificate_handoff(
3131 &self,
3132 cx: &Cx,
3133 record: &ParallelWalDurableCertificateRecord,
3134 ) -> Result<()> {
3135 let handoff_path = self.certificate_checkpoint_handoff_path();
3136 let existed = self.vfs.access(cx, &handoff_path, AccessFlags::EXISTS)?;
3137 let flags = VfsOpenFlags::READWRITE | VfsOpenFlags::CREATE | VfsOpenFlags::WAL;
3138 let (mut file, _) = self.vfs.open(cx, Some(&handoff_path), flags)?;
3139 let record_bytes = record.to_bytes();
3140 if record_bytes.len() > PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE {
3141 let cleanup_cx = cx.create_child();
3142 let _cleanup_mask = cleanup_cx.masked();
3143 let close_result = file.close(&cleanup_cx);
3144 return combine_sidecar_io_results(
3145 "parallel WAL checkpoint certificate handoff is oversized",
3146 [
3147 (
3148 "record_size",
3149 Err(FrankenError::WalCorrupt {
3150 detail: format!(
3151 "parallel WAL checkpoint certificate handoff is {} bytes; maximum is {}",
3152 record_bytes.len(),
3153 PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE
3154 ),
3155 }),
3156 ),
3157 ("close", close_result),
3158 ],
3159 );
3160 }
3161 if existed {
3166 let file_size = file.file_size(cx)?;
3167 if file_size == u64::try_from(record_bytes.len()).unwrap_or(u64::MAX) {
3168 let mut current = vec![0_u8; record_bytes.len()];
3169 let unchanged = file
3170 .read(cx, &mut current, 0)
3171 .await
3172 .is_ok_and(|bytes_read| {
3173 bytes_read == record_bytes.len() && current == record_bytes
3174 });
3175 if unchanged {
3176 let cleanup_cx = cx.create_child();
3177 let _cleanup_mask = cleanup_cx.masked();
3178 return file.close(&cleanup_cx);
3179 }
3180 }
3181 }
3182 let mutation_cx = cx.create_child();
3187 let _mutation_mask = mutation_cx.masked();
3188 let truncate_result = file.truncate(&mutation_cx, 0);
3189 let write_result = if truncate_result.is_ok() {
3190 file.write(&mutation_cx, &record_bytes, 0).await
3191 } else {
3192 Ok(())
3193 };
3194 if let Err(write_error) = write_result {
3195 let cleanup_result = file.truncate(&mutation_cx, 0);
3196 let close_result = file.close(&mutation_cx);
3197 return combine_sidecar_io_results(
3198 "parallel WAL checkpoint certificate handoff cleanup failed",
3199 [
3200 ("truncate_before_write", truncate_result),
3201 ("write", Err(write_error)),
3202 ("truncate_after_write", cleanup_result),
3203 ("close", close_result),
3204 ],
3205 );
3206 }
3207 let sync_result = if truncate_result.is_ok() {
3208 file.durable_sync(&mutation_cx, SyncKind::FullDurable)
3209 } else {
3210 Ok(())
3211 };
3212 let directory_sync_result = if !existed && truncate_result.is_ok() && sync_result.is_ok() {
3213 self.vfs.sync_parent_directory(&mutation_cx, &handoff_path)
3214 } else {
3215 Ok(())
3216 };
3217 let close_result = file.close(&mutation_cx);
3218 combine_sidecar_io_results(
3219 "parallel WAL checkpoint certificate handoff finalization failed",
3220 [
3221 ("truncate", truncate_result),
3222 ("file_sync", sync_result),
3223 ("directory_sync", directory_sync_result),
3224 ("close", close_result),
3225 ],
3226 )
3227 }
3228
3229 async fn checkpoint_certificate_handoff(
3230 &self,
3231 cx: &Cx,
3232 ) -> Result<Option<ParallelWalCommitCertificate>> {
3233 let handoff_path = self.certificate_checkpoint_handoff_path();
3234 if !self.vfs.access(cx, &handoff_path, AccessFlags::EXISTS)? {
3235 return Ok(None);
3236 }
3237 let flags = VfsOpenFlags::READONLY | VfsOpenFlags::WAL;
3238 let (mut file, _) = self.vfs.open(cx, Some(&handoff_path), flags)?;
3239 let read_result = async {
3240 let file_size =
3241 usize::try_from(file.file_size(cx)?).map_err(|_| FrankenError::WalCorrupt {
3242 detail: "parallel WAL checkpoint certificate handoff exceeds usize".to_owned(),
3243 })?;
3244 if file_size == 0 || file_size > PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE {
3245 return Err(FrankenError::WalCorrupt {
3246 detail: format!(
3247 "parallel WAL checkpoint certificate handoff has invalid size {file_size}"
3248 ),
3249 });
3250 }
3251 let mut bytes = vec![0_u8; file_size];
3252 let bytes_read = file.read(cx, &mut bytes, 0).await?;
3253 if bytes_read != bytes.len() {
3254 return Err(FrankenError::WalCorrupt {
3255 detail: "parallel WAL checkpoint certificate handoff was short-read".to_owned(),
3256 });
3257 }
3258 if durable_certificate_is_legacy_identityless(&bytes) {
3263 tracing::debug!(
3264 target: "fsqlite::wal::durability_combiner",
3265 "ignored checkpoint handoff certificate from a legacy identity-less build"
3266 );
3267 return Ok(None);
3268 }
3269 let record =
3270 ParallelWalDurableCertificateRecord::from_bytes(&bytes).map_err(|error| {
3271 FrankenError::WalCorrupt {
3272 detail: format!(
3273 "parallel WAL checkpoint certificate handoff is invalid: {error}"
3274 ),
3275 }
3276 })?;
3277 if self.db_file_identity_rejects_record(&record) {
3286 tracing::debug!(
3287 target: "fsqlite::wal::durability_combiner",
3288 "ignored checkpoint handoff certificate bound to a foreign database identity"
3289 );
3290 return Ok(None);
3291 }
3292 Ok(Some(record.certificate))
3293 }
3294 .await;
3295 let cleanup_cx = cx.create_child();
3296 let _cleanup_mask = cleanup_cx.masked();
3297 let close_result = file.close(&cleanup_cx);
3298 match (read_result, close_result) {
3299 (Ok(certificate), Ok(())) => Ok(certificate),
3300 (Err(error), Ok(())) | (Ok(_), Err(error)) => Err(error),
3301 (Err(read_error), Err(close_error)) => Err(FrankenError::internal(format!(
3302 "parallel WAL checkpoint handoff read failed and close also failed: read={read_error}; close={close_error}"
3303 ))),
3304 }
3305 }
3306
3307 async fn wal_frame_payload_digest(
3308 &self,
3309 cx: &Cx,
3310 wal_frame_start: u64,
3311 wal_frame_end: u64,
3312 ) -> Result<[u8; 32]> {
3313 if wal_frame_start == 0 || wal_frame_end < wal_frame_start {
3314 return Err(FrankenError::WalCorrupt {
3315 detail: format!(
3316 "invalid parallel WAL digest interval {wal_frame_start}..={wal_frame_end}"
3317 ),
3318 });
3319 }
3320
3321 let mut digest = ParallelWalFramePayloadDigestBuilder::new();
3322 for frame_number in wal_frame_start..=wal_frame_end {
3323 let frame_index = usize::try_from(frame_number.saturating_sub(1)).map_err(|_| {
3324 FrankenError::WalCorrupt {
3325 detail: format!(
3326 "parallel WAL digest frame number {frame_number} exceeds usize"
3327 ),
3328 }
3329 })?;
3330 let (header, page_data) = self.inner.inner().read_frame(cx, frame_index).await?;
3331 let page_number =
3332 PageNumber::new(header.page_number).ok_or_else(|| FrankenError::WalCorrupt {
3333 detail: format!(
3334 "parallel WAL digest frame {frame_number} has invalid page number {}",
3335 header.page_number
3336 ),
3337 })?;
3338 digest.update(page_number, header.db_size, &page_data);
3339 }
3340 Ok(digest.finalize())
3341 }
3342
3343 async fn latest_authorized_durable_certificate_record(
3344 &self,
3345 cx: &Cx,
3346 ) -> Result<Option<ParallelWalDurableCertificateRecord>> {
3347 let certificate_path = self.certificate_sidecar_path();
3348 let cached_reader = self
3358 .cached_certificate_read
3359 .lock()
3360 .unwrap_or_else(std::sync::PoisonError::into_inner)
3361 .take();
3362 let mut file = match cached_reader {
3363 Some(cached) => cached,
3364 None => {
3365 if !self
3366 .vfs
3367 .access(cx, &certificate_path, AccessFlags::EXISTS)?
3368 {
3369 return Ok(None);
3370 }
3371 let flags = VfsOpenFlags::READONLY | VfsOpenFlags::WAL;
3372 let (file, _) = self.vfs.open(cx, Some(&certificate_path), flags)?;
3373 file
3374 }
3375 };
3376 let read_result = async {
3377 let file_size = file.file_size(cx)?;
3378 if file_size == 0 {
3379 return Ok(None);
3380 }
3381
3382 let footer_size =
3385 u64::try_from(ParallelWalDurableCertificateRecord::LENGTH_FOOTER_SIZE)
3386 .unwrap_or(4);
3387 let mut newest = None;
3388 if file_size >= footer_size {
3389 let footer_offset = file_size - footer_size;
3390 let footer = Self::read_certificate_sidecar_exact(
3391 &file,
3392 cx,
3393 footer_offset,
3394 ParallelWalDurableCertificateRecord::LENGTH_FOOTER_SIZE,
3395 "newest length footer",
3396 )
3397 .await?;
3398 let record_len = usize::try_from(u32::from_le_bytes([
3399 footer[0], footer[1], footer[2], footer[3],
3400 ]))
3401 .map_err(|_| FrankenError::WalCorrupt {
3402 detail: "parallel WAL certificate newest footer length exceeds usize"
3403 .to_owned(),
3404 })?;
3405 let record_len_u64 = u64::try_from(record_len).unwrap_or(u64::MAX);
3406 if (MIN_DURABLE_CERTIFICATE_RECORD_SIZE
3407 ..=PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE)
3408 .contains(&record_len)
3409 && record_len_u64 <= file_size
3410 {
3411 let record_start = file_size - record_len_u64;
3412 let bytes = Self::read_certificate_sidecar_exact(
3413 &file,
3414 cx,
3415 record_start,
3416 record_len,
3417 "newest record",
3418 )
3419 .await?;
3420 if durable_certificate_is_legacy_identityless(&bytes) {
3427 tracing::debug!(
3428 target: "fsqlite::wal::durability_combiner",
3429 "ignored durable certificate sidecar from a legacy identity-less build"
3430 );
3431 return Ok(None);
3432 }
3433 if bytes.starts_with(&PARALLEL_WAL_DURABLE_CERTIFICATE_MAGIC)
3438 || durable_certificate_declares_len(&bytes, record_len)
3439 {
3440 let record =
3441 decode_durable_certificate_record(&bytes, "newest record")?;
3442 newest = Some((record_start, record));
3443 }
3444 }
3445 }
3446
3447 if newest.is_none() {
3448 let recovery_window_size =
3455 PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE.saturating_mul(2);
3456 let recovery_window_size_u64 =
3457 u64::try_from(recovery_window_size).unwrap_or(u64::MAX);
3458 let tail_offset = file_size.saturating_sub(recovery_window_size_u64);
3459 let tail_len =
3460 usize::try_from(file_size - tail_offset).map_err(|_| {
3461 FrankenError::WalCorrupt {
3462 detail: "parallel WAL certificate recovery window exceeds usize"
3463 .to_owned(),
3464 }
3465 })?;
3466 let tail = Self::read_certificate_sidecar_exact(
3467 &file,
3468 cx,
3469 tail_offset,
3470 tail_len,
3471 "recovery window",
3472 )
3473 .await?;
3474 let minimum_candidate_end = tail
3475 .len()
3476 .saturating_sub(PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE)
3477 .max(ParallelWalDurableCertificateRecord::LENGTH_FOOTER_SIZE);
3478 let mut anchor = None;
3479 for candidate_end in (minimum_candidate_end..tail.len()).rev() {
3480 let footer_start = candidate_end
3481 - ParallelWalDurableCertificateRecord::LENGTH_FOOTER_SIZE;
3482 let footer = &tail[footer_start..candidate_end];
3483 let record_len = usize::try_from(u32::from_le_bytes([
3484 footer[0], footer[1], footer[2], footer[3],
3485 ]))
3486 .unwrap_or(usize::MAX);
3487 if !(MIN_DURABLE_CERTIFICATE_RECORD_SIZE
3488 ..=PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE)
3489 .contains(&record_len)
3490 || record_len > candidate_end
3491 {
3492 continue;
3493 }
3494 let record_start = candidate_end - record_len;
3495 let record_bytes = &tail[record_start..candidate_end];
3496 if !record_bytes.starts_with(&PARALLEL_WAL_DURABLE_CERTIFICATE_MAGIC)
3497 || !durable_certificate_declares_len(record_bytes, record_len)
3498 {
3499 continue;
3500 }
3501 if let Ok(record) =
3502 ParallelWalDurableCertificateRecord::from_bytes(record_bytes)
3503 {
3504 anchor = Some((record_start, candidate_end, record));
3505 break;
3506 }
3507 }
3508
3509 if let Some((record_start, record_end, record)) = anchor {
3510 validate_incomplete_certificate_suffix(&tail[record_end..], true)?;
3511 let absolute_start = tail_offset
3512 .checked_add(u64::try_from(record_start).unwrap_or(u64::MAX))
3513 .ok_or_else(|| FrankenError::WalCorrupt {
3514 detail:
3515 "parallel WAL certificate recovery anchor offset overflow"
3516 .to_owned(),
3517 })?;
3518 newest = Some((absolute_start, record));
3519 } else {
3520 if file_size
3521 > u64::try_from(PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE)
3522 .unwrap_or(u64::MAX)
3523 {
3524 return Err(FrankenError::WalCorrupt {
3525 detail: format!(
3526 "parallel WAL certificate sidecar has no valid record within its bounded {}-byte recovery suffix",
3527 PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE
3528 ),
3529 });
3530 }
3531 validate_incomplete_certificate_suffix(&tail, false)?;
3532 return Ok(None);
3533 }
3534 }
3535
3536 let valid_frame_count = u64::try_from(self.inner.frame_count()).unwrap_or(u64::MAX);
3537 let wal_generation = self.inner.inner().generation_identity();
3538 let (mut record_start, mut record) = newest.ok_or_else(|| {
3539 FrankenError::WalCorrupt {
3540 detail: "parallel WAL certificate recovery produced no record".to_owned(),
3541 }
3542 })?;
3543 let mut unauthorized_records = 0_usize;
3544 loop {
3545 if record.wal_generation != wal_generation {
3550 return Ok(None);
3551 }
3552 if self.db_file_identity_rejects_record(&record) {
3559 tracing::debug!(
3560 target: "fsqlite::wal::durability_combiner",
3561 "ignored durable certificate sidecar bound to a foreign database identity"
3562 );
3563 return Ok(None);
3564 }
3565 let frame_index =
3566 usize::try_from(record.wal_frame_end.saturating_sub(1)).map_err(|_| {
3567 FrankenError::WalCorrupt {
3568 detail: "parallel WAL certificate commit-marker index exceeds usize"
3569 .to_owned(),
3570 }
3571 })?;
3572 let commit_marker_frame = if frame_index < self.inner.frame_count()
3573 && self
3574 .inner
3575 .inner()
3576 .read_frame_header(cx, frame_index)
3577 .await?
3578 .is_commit()
3579 {
3580 record.wal_frame_end
3581 } else {
3582 0
3583 };
3584 let actual_wal_frame_payload_digest =
3585 if commit_marker_frame == record.wal_frame_end {
3586 Some(
3587 self.wal_frame_payload_digest(
3588 cx,
3589 record.wal_frame_start,
3590 record.wal_frame_end,
3591 )
3592 .await?,
3593 )
3594 } else {
3595 None
3596 };
3597 if actual_wal_frame_payload_digest.is_some_and(|actual_digest| {
3598 record.authorizes_wal_boundary(
3599 wal_generation,
3600 valid_frame_count,
3601 commit_marker_frame,
3602 actual_digest,
3603 )
3604 }) {
3605 return Ok(Some(record));
3606 }
3607
3608 if record.wal_frame_end > valid_frame_count {
3621 tracing::debug!(
3622 target: "fsqlite::wal::durability_combiner",
3623 future_certificate_epoch = record.certificate.certificate_epoch,
3624 future_commit_seq_hi = record.certificate.commit_seq_hi.get(),
3625 future_wal_frame_end = record.wal_frame_end,
3626 valid_frame_count,
3627 "skipped parallel WAL certificate newer than reader frame snapshot"
3628 );
3629 } else {
3630 unauthorized_records = unauthorized_records.saturating_add(1);
3631 if unauthorized_records > MAX_ORPHAN_CERTIFICATE_LOOKBACK {
3632 return Err(FrankenError::WalCorrupt {
3633 detail: format!(
3634 "parallel WAL certificate sidecar exceeded bounded orphan lookback {MAX_ORPHAN_CERTIFICATE_LOOKBACK}"
3635 ),
3636 });
3637 }
3638 tracing::debug!(
3639 target: "fsqlite::wal::durability_combiner",
3640 orphan_certificate_epoch = record.certificate.certificate_epoch,
3641 orphan_commit_seq_hi = record.certificate.commit_seq_hi.get(),
3642 orphan_wal_frame_end = record.wal_frame_end,
3643 lookback = unauthorized_records,
3644 "ignored unauthorized parallel WAL certificate tail"
3645 );
3646 }
3647 if record_start == 0 {
3648 return Ok(None);
3649 }
3650 (record_start, record) =
3651 Self::read_certificate_record_ending_at(&file, cx, record_start).await?;
3652 }
3653 }
3654 .await;
3655 match read_result {
3656 Ok(certificate) => {
3661 *self
3662 .cached_certificate_read
3663 .lock()
3664 .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(file);
3665 Ok(certificate)
3666 }
3667 Err(error) => {
3670 let cleanup_cx = cx.create_child();
3671 let _cleanup_mask = cleanup_cx.masked();
3672 let _ = file.close(&cleanup_cx);
3673 Err(error)
3674 }
3675 }
3676 }
3677}
3678
3679impl<V> WalBackend for PathRefreshingWalBackend<V>
3680where
3681 V: Vfs + 'static,
3682 V::File: Send + Sync + 'static,
3683{
3684 fn begin_transaction<'a>(&'a mut self, cx: &'a Cx) -> WalFuture<'a, ()> {
3685 Box::pin(async move {
3686 self.ensure_current_wal_path(cx).await?;
3687 self.inner.begin_transaction(cx).await
3688 })
3689 }
3690
3691 fn published_snapshot(&self) -> Option<WalPublicationSnapshot> {
3692 Some(self.inner.published_snapshot())
3693 }
3694
3695 fn pinned_read_snapshot(&self) -> Option<WalPublicationSnapshot> {
3696 self.inner.pinned_read_snapshot()
3697 }
3698
3699 fn pinned_logical_read_snapshot<'a>(
3700 &'a self,
3701 cx: &'a Cx,
3702 ) -> WalFuture<'a, Option<WalLogicalReadSnapshot>> {
3703 Box::pin(async move {
3704 let Some(pinned) = self.inner.pinned_read_snapshot() else {
3705 return Ok(None);
3706 };
3707 let Some(last_commit_frame) = pinned.last_commit_frame else {
3708 return Ok(None);
3709 };
3710 let Some(record) = self
3711 .latest_authorized_durable_certificate_record(cx)
3712 .await?
3713 else {
3714 return Ok(None);
3715 };
3716 if record.wal_generation != pinned.generation {
3717 return Err(FrankenError::WalCorrupt {
3718 detail: "current logical WAL certificate generation differs from pinned reader"
3719 .to_owned(),
3720 });
3721 }
3722 let certificate_commit_frame =
3723 usize::try_from(record.wal_frame_end.checked_sub(1).ok_or_else(|| {
3724 FrankenError::WalCorrupt {
3725 detail: "current logical WAL certificate ends at frame zero".to_owned(),
3726 }
3727 })?)
3728 .map_err(|_| FrankenError::WalCorrupt {
3729 detail: "current logical WAL certificate frame exceeds usize".to_owned(),
3730 })?;
3731 if certificate_commit_frame > last_commit_frame {
3732 return Err(FrankenError::WalCorrupt {
3733 detail: "current logical WAL certificate extends past pinned reader horizon"
3734 .to_owned(),
3735 });
3736 }
3737
3738 let first_tail_frame =
3739 usize::try_from(record.wal_frame_end).map_err(|_| FrankenError::WalCorrupt {
3740 detail: "logical WAL tail frame exceeds usize".to_owned(),
3741 })?;
3742 let mut tail_commit_count = 0_u64;
3743 if first_tail_frame <= last_commit_frame {
3744 for frame_index in first_tail_frame..=last_commit_frame {
3745 if self
3746 .inner
3747 .inner()
3748 .read_frame_header(cx, frame_index)
3749 .await?
3750 .is_commit()
3751 {
3752 tail_commit_count = tail_commit_count.checked_add(1).ok_or_else(|| {
3753 FrankenError::WalCorrupt {
3754 detail: "logical WAL tail commit count overflow".to_owned(),
3755 }
3756 })?;
3757 }
3758 }
3759 }
3760 let visible_commit_seq = CommitSeq::new(
3761 record
3762 .certificate
3763 .commit_seq_hi
3764 .get()
3765 .checked_add(tail_commit_count)
3766 .ok_or_else(|| FrankenError::WalCorrupt {
3767 detail: "logical WAL visible commit sequence overflow".to_owned(),
3768 })?,
3769 );
3770 Ok(Some(WalLogicalReadSnapshot {
3771 generation: pinned.generation,
3772 last_commit_frame: pinned.last_commit_frame,
3773 visible_commit_seq,
3774 }))
3775 })
3776 }
3777
3778 fn refresh_published_snapshot<'a>(
3779 &'a mut self,
3780 cx: &'a Cx,
3781 ) -> WalFuture<'a, Option<WalPublicationSnapshot>> {
3782 Box::pin(async move {
3783 self.ensure_current_wal_path(cx).await?;
3784 self.inner.refresh_published_snapshot(cx).await.map(Some)
3785 })
3786 }
3787
3788 fn publish_authorized_deferred_commit<'a>(&'a mut self, cx: &'a Cx) -> WalFuture<'a, ()> {
3789 Box::pin(async move { self.inner.publish_authorized_deferred_commit(cx) })
3790 }
3791
3792 fn append_frame<'a>(
3793 &'a mut self,
3794 cx: &'a Cx,
3795 page_number: u32,
3796 page_data: &'a [u8],
3797 db_size_if_commit: u32,
3798 ) -> WalFuture<'a, ()> {
3799 Box::pin(async move {
3800 self.ensure_current_wal_path(cx).await?;
3801 self.inner
3802 .append_frame(cx, page_number, page_data, db_size_if_commit)
3803 .await
3804 })
3805 }
3806
3807 fn append_frames<'a>(
3808 &'a mut self,
3809 cx: &'a Cx,
3810 frames: &'a [WalFrameRef<'a>],
3811 ) -> WalFuture<'a, ()> {
3812 Box::pin(async move {
3813 self.ensure_current_wal_path(cx).await?;
3814 self.inner.append_frames(cx, frames).await
3815 })
3816 }
3817
3818 fn append_frames_tracked<'a>(
3819 &'a mut self,
3820 cx: &'a Cx,
3821 frames: &'a [WalFrameRef<'a>],
3822 completion: VfsWriteCompletion,
3823 ) -> WalFuture<'a, ()> {
3824 Box::pin(async move {
3825 let mut preflight = WalWriteCompletionPreflight::new(Some(&completion));
3826 self.ensure_current_wal_path(cx).await?;
3827 preflight.hand_off();
3828 drop(preflight);
3829 self.inner
3830 .append_frames_tracked(cx, frames, completion)
3831 .await
3832 })
3833 }
3834
3835 fn prepare_append_frames(
3836 &self,
3837 frames: &[WalFrameRef<'_>],
3838 ) -> Result<Option<PreparedWalFrameBatch>> {
3839 self.inner.prepare_append_frames(frames)
3840 }
3841
3842 fn finalize_prepared_frames(
3843 &self,
3844 cx: &Cx,
3845 prepared: &mut PreparedWalFrameBatch,
3846 ) -> Result<()> {
3847 self.inner.finalize_prepared_frames(cx, prepared)
3848 }
3849
3850 fn append_prepared_frames<'a>(
3851 &'a mut self,
3852 cx: &'a Cx,
3853 prepared: &'a mut PreparedWalFrameBatch,
3854 ) -> WalFuture<'a, ()> {
3855 Box::pin(async move {
3856 self.ensure_current_wal_path(cx).await?;
3857 self.inner.append_prepared_frames(cx, prepared).await
3858 })
3859 }
3860
3861 fn append_prepared_frames_tracked<'a>(
3862 &'a mut self,
3863 cx: &'a Cx,
3864 prepared: &'a mut PreparedWalFrameBatch,
3865 completion: VfsWriteCompletion,
3866 ) -> WalFuture<'a, ()> {
3867 Box::pin(async move {
3868 let mut preflight = WalWriteCompletionPreflight::new(Some(&completion));
3869 self.ensure_current_wal_path(cx).await?;
3870 preflight.hand_off();
3871 drop(preflight);
3872 self.inner
3873 .append_prepared_frames_tracked(cx, prepared, completion)
3874 .await
3875 })
3876 }
3877
3878 fn persist_parallel_wal_commit_certificate<'a>(
3879 &'a mut self,
3880 cx: &'a Cx,
3881 certificate: &'a ParallelWalCommitCertificate,
3882 wal_frame_start: u64,
3883 wal_frame_end: u64,
3884 sync: bool,
3885 ) -> WalFuture<'a, ()> {
3886 Box::pin(async move {
3887 self.ensure_current_wal_path(cx).await?;
3888 self.append_durable_certificate_record(
3889 cx,
3890 certificate,
3891 wal_frame_start,
3892 wal_frame_end,
3893 sync,
3894 )
3895 .await
3896 })
3897 }
3898
3899 fn persist_parallel_wal_commit_certificate_tracked<'a>(
3900 &'a mut self,
3901 cx: &'a Cx,
3902 certificate: &'a ParallelWalCommitCertificate,
3903 wal_frame_start: u64,
3904 wal_frame_end: u64,
3905 sync: bool,
3906 completion: VfsWriteCompletion,
3907 ) -> WalFuture<'a, ()> {
3908 Box::pin(async move {
3909 let mut preflight = WalWriteCompletionPreflight::new(Some(&completion));
3910 self.ensure_current_wal_path(cx).await?;
3911 preflight.hand_off();
3912 drop(preflight);
3913 self.append_durable_certificate_record_with_completion(
3914 cx,
3915 certificate,
3916 wal_frame_start,
3917 wal_frame_end,
3918 sync,
3919 Some(&completion),
3920 )
3921 .await
3922 })
3923 }
3924
3925 fn reconcile_parallel_wal_commit<'a>(
3926 &'a mut self,
3927 cx: &'a Cx,
3928 certificate: &'a ParallelWalCommitCertificate,
3929 wal_frame_start: u64,
3930 wal_frame_end: u64,
3931 sync: bool,
3932 ) -> WalFuture<'a, ParallelWalCommitReconciliation> {
3933 Box::pin(async move {
3934 self.ensure_current_wal_path(cx).await?;
3935 self.inner.wal.refresh(cx).await?;
3936 let wal_generation = self.inner.wal.generation_identity();
3937 let expected_record = ParallelWalDurableCertificateRecord::new(
3938 wal_generation,
3939 wal_frame_start,
3940 wal_frame_end,
3941 self.db_file_id_for_written_certificate(),
3944 certificate.clone(),
3945 )
3946 .map_err(|error| {
3947 FrankenError::internal(format!(
3948 "could not reconstruct in-doubt parallel WAL certificate: {error}"
3949 ))
3950 })?;
3951
3952 let valid_frame_count = u64::try_from(self.inner.wal.frame_count()).unwrap_or(u64::MAX);
3953 let target_commit_present = if valid_frame_count < wal_frame_end {
3954 false
3955 } else {
3956 let target_index =
3957 usize::try_from(wal_frame_end.saturating_sub(1)).map_err(|_| {
3958 FrankenError::WalCorrupt {
3959 detail: "in-doubt WAL commit-marker index exceeds usize".to_owned(),
3960 }
3961 })?;
3962 self.inner
3963 .wal
3964 .read_frame_header(cx, target_index)
3965 .await?
3966 .is_commit()
3967 };
3968
3969 if target_commit_present {
3970 if valid_frame_count != wal_frame_end {
3971 return Err(FrankenError::WalCorrupt {
3972 detail: format!(
3973 "in-doubt parallel WAL interval ends at frame {wal_frame_end}, but the retained writer gate observed committed frame count {valid_frame_count}"
3974 ),
3975 });
3976 }
3977 let actual_wal_frame_payload_digest = self
3978 .wal_frame_payload_digest(cx, wal_frame_start, wal_frame_end)
3979 .await?;
3980 if !expected_record.authorizes_wal_boundary(
3981 wal_generation,
3982 valid_frame_count,
3983 wal_frame_end,
3984 actual_wal_frame_payload_digest,
3985 ) {
3986 return Err(FrankenError::WalCorrupt {
3987 detail: format!(
3988 "in-doubt parallel WAL interval {wal_frame_start}..={wal_frame_end} does not match its content-bound certificate"
3989 ),
3990 });
3991 }
3992 let sidecar_is_exact = self
3993 .reconcile_certificate_sidecar_record(cx, &expected_record, false, sync)
3994 .await?;
3995 if !sidecar_is_exact {
3996 return Err(FrankenError::WalCorrupt {
3997 detail: format!(
3998 "parallel WAL commit marker at frame {wal_frame_end} has no exact durable certificate"
3999 ),
4000 });
4001 }
4002 if sync {
4003 self.inner.wal.sync(cx, SyncFlags::NORMAL)?;
4004 self.vfs.sync_parent_directory(cx, &self.wal_path)?;
4005 }
4006 return Ok(ParallelWalCommitReconciliation::Authorized);
4007 }
4008
4009 let committed_prefix_before =
4010 wal_frame_start
4011 .checked_sub(1)
4012 .ok_or_else(|| FrankenError::WalCorrupt {
4013 detail: "parallel WAL recovery interval starts at frame zero".to_owned(),
4014 })?;
4015 if valid_frame_count != committed_prefix_before {
4016 return Err(FrankenError::WalCorrupt {
4017 detail: format!(
4018 "in-doubt WAL interval {wal_frame_start}..={wal_frame_end} has unexpected committed prefix {valid_frame_count}"
4019 ),
4020 });
4021 }
4022 self.reconcile_certificate_sidecar_record(cx, &expected_record, true, sync)
4027 .await?;
4028 self.inner.wal.repair_uncommitted_tail(cx)?;
4029 if sync {
4030 self.inner.wal.sync(cx, SyncFlags::NORMAL)?;
4031 self.vfs.sync_parent_directory(cx, &self.wal_path)?;
4032 }
4033 Ok(ParallelWalCommitReconciliation::NotCommitted)
4034 })
4035 }
4036
4037 fn latest_authorized_parallel_wal_commit_certificate<'a>(
4038 &'a mut self,
4039 cx: &'a Cx,
4040 ) -> WalFuture<'a, Option<ParallelWalCommitCertificate>> {
4041 Box::pin(async move {
4042 self.ensure_current_wal_path(cx).await?;
4043 if let Some(record) = self
4044 .latest_authorized_durable_certificate_record(cx)
4045 .await?
4046 {
4047 return Ok(Some(record.certificate));
4048 }
4049 self.checkpoint_certificate_handoff(cx).await
4050 })
4051 }
4052
4053 fn read_page<'a>(&'a mut self, cx: &'a Cx, page_number: u32) -> WalFuture<'a, Option<Vec<u8>>> {
4054 Box::pin(async move {
4055 self.ensure_current_wal_path(cx).await?;
4056 self.inner.read_page(cx, page_number).await
4057 })
4058 }
4059
4060 fn read_page_at_appended_tail<'a>(
4064 &'a mut self,
4065 cx: &'a Cx,
4066 page_number: u32,
4067 ) -> WalFuture<'a, Option<Vec<u8>>> {
4068 Box::pin(async move {
4069 self.ensure_current_wal_path(cx).await?;
4070 self.inner.read_page_at_appended_tail(cx, page_number).await
4071 })
4072 }
4073
4074 fn read_page_pinned<'a>(
4075 &'a self,
4076 cx: &'a Cx,
4077 page_number: u32,
4078 ) -> WalFuture<'a, Option<Vec<u8>>> {
4079 Box::pin(async move { self.inner.read_page_pinned(cx, page_number).await })
4080 }
4081
4082 fn supports_pinned_reads(&self) -> bool {
4083 self.inner.supports_pinned_reads()
4084 }
4085
4086 fn committed_txns_since_page<'a>(
4087 &'a mut self,
4088 cx: &'a Cx,
4089 page_number: u32,
4090 ) -> WalFuture<'a, u64> {
4091 Box::pin(async move {
4092 self.ensure_current_wal_path(cx).await?;
4093 self.inner.committed_txns_since_page(cx, page_number).await
4094 })
4095 }
4096
4097 fn conflicting_pages_since_snapshot<'a>(
4098 &'a mut self,
4099 cx: &'a Cx,
4100 snapshot: TransactionConflictSnapshot,
4101 page_numbers: &'a [u32],
4102 page_baselines: &'a [TransactionConflictPageBaseline],
4103 ) -> WalFuture<'a, Vec<u32>> {
4104 Box::pin(async move {
4105 self.ensure_current_wal_path(cx).await?;
4106 let latest = self.inner.refresh_published_snapshot(cx).await?;
4107 if latest.generation != snapshot.generation {
4108 return Ok(self
4109 .conflicts_after_generation_change(cx, page_numbers, page_baselines)
4110 .await);
4111 }
4112 self.inner
4113 .conflicting_pages_since_snapshot(cx, snapshot, page_numbers, page_baselines)
4114 .await
4115 })
4116 }
4117
4118 fn committed_txn_count<'a>(&'a mut self, cx: &'a Cx) -> WalFuture<'a, u64> {
4119 Box::pin(async move {
4120 self.ensure_current_wal_path(cx).await?;
4121 self.inner.committed_txn_count(cx).await
4122 })
4123 }
4124
4125 fn sync(&mut self, cx: &Cx) -> Result<()> {
4126 #[cfg(all(feature = "native", any(unix, windows)))]
4127 if let Some(binding) = &self.namespace_binding {
4128 binding.validate_path_identity()?;
4129 }
4130 self.inner.sync(cx)
4131 }
4132
4133 fn frame_count(&self) -> usize {
4134 self.inner.frame_count()
4135 }
4136
4137 fn checkpoint<'a>(
4138 &'a mut self,
4139 cx: &'a Cx,
4140 mode: CheckpointMode,
4141 writer: &'a mut dyn CheckpointPageWriter,
4142 backfilled_frames: u32,
4143 oldest_reader_frame: Option<u32>,
4144 ) -> WalFuture<'a, CheckpointResult> {
4145 Box::pin(async move {
4146 self.ensure_current_wal_path(cx).await?;
4147 let checkpoint_handoff = self
4148 .latest_authorized_durable_certificate_record(cx)
4149 .await?;
4150 if let Some(record) = checkpoint_handoff.as_ref() {
4151 self.persist_checkpoint_certificate_handoff(cx, record)
4158 .await?;
4159 }
4160 let result = self
4161 .inner
4162 .checkpoint(cx, mode, writer, backfilled_frames, oldest_reader_frame)
4163 .await?;
4164 if let Some(mut stale) = self
4168 .cached_certificate_read
4169 .get_mut()
4170 .unwrap_or_else(std::sync::PoisonError::into_inner)
4171 .take()
4172 {
4173 let cleanup_cx = cx.create_child();
4174 let _cleanup_mask = cleanup_cx.masked();
4175 let _ = stale.close(&cleanup_cx);
4176 }
4177 Ok(result)
4178 })
4179 }
4180}
4181
4182struct CheckpointTargetAdapterRef<'a> {
4187 writer: &'a mut dyn CheckpointPageWriter,
4188}
4189
4190impl CheckpointTarget for CheckpointTargetAdapterRef<'_> {
4191 fn write_page<'a>(
4192 &'a mut self,
4193 cx: &'a Cx,
4194 page_no: PageNumber,
4195 data: &'a [u8],
4196 ) -> CheckpointTargetFuture<'a, ()> {
4197 Box::pin(async move { self.writer.write_page(cx, page_no, data).await })
4198 }
4199
4200 fn truncate_db<'a>(&'a mut self, cx: &'a Cx, n_pages: u32) -> CheckpointTargetFuture<'a, ()> {
4201 Box::pin(async move { self.writer.truncate(cx, n_pages).await })
4202 }
4203
4204 fn sync_db<'a>(&'a mut self, cx: &'a Cx) -> CheckpointTargetFuture<'a, ()> {
4205 Box::pin(async move { self.writer.sync(cx).await })
4206 }
4207}
4208
4209#[cfg(test)]
4214mod tests {
4215 use std::sync::Mutex;
4216
4217 use fsqlite_pager::MockCheckpointPageWriter;
4218 use fsqlite_pager::traits::WalFrameRef;
4219 use fsqlite_types::flags::VfsOpenFlags;
4220 use fsqlite_vfs::MemoryVfs;
4221 use fsqlite_vfs::traits::{Vfs, VfsFile};
4222 use fsqlite_wal::checksum::WalSalts;
4223
4224 use super::*;
4225
4226 const PAGE_SIZE: u32 = 4096;
4227 const CERTIFICATE_PATH: &str = "test.db-wal-cert";
4228 const CHECKPOINT_HANDOFF_PATH: &str = "test.db-wal-cert-head";
4229
4230 #[derive(Clone, Copy, Debug)]
4231 enum CheckpointHandoffWriteFault {
4232 Error,
4233 Pending,
4234 }
4235
4236 #[derive(Clone, Debug, Eq, PartialEq)]
4237 enum CertificateSyncObservation {
4238 Ordinary(PathBuf),
4239 Durable(PathBuf, SyncKind),
4240 }
4241
4242 #[derive(Debug, Default)]
4243 struct CheckpointHandoffFaultState {
4244 next_write: Option<CheckpointHandoffWriteFault>,
4245 fail_next_sync: bool,
4246 fail_next_wal_sync: bool,
4248 sync_observations: Vec<CertificateSyncObservation>,
4249 }
4250
4251 #[derive(Clone, Debug)]
4252 struct CheckpointHandoffFaultVfs {
4253 inner: MemoryVfs,
4254 faults: Arc<Mutex<CheckpointHandoffFaultState>>,
4255 }
4256
4257 impl CheckpointHandoffFaultVfs {
4258 fn new() -> Self {
4259 Self {
4260 inner: MemoryVfs::new(),
4261 faults: Arc::new(Mutex::new(CheckpointHandoffFaultState::default())),
4262 }
4263 }
4264
4265 fn fail_next_handoff_write(&self) {
4266 self.faults
4267 .lock()
4268 .unwrap_or_else(std::sync::PoisonError::into_inner)
4269 .next_write = Some(CheckpointHandoffWriteFault::Error);
4270 }
4271
4272 fn pend_next_handoff_write(&self) {
4273 self.faults
4274 .lock()
4275 .unwrap_or_else(std::sync::PoisonError::into_inner)
4276 .next_write = Some(CheckpointHandoffWriteFault::Pending);
4277 }
4278
4279 fn fail_next_handoff_sync(&self) {
4280 self.faults
4281 .lock()
4282 .unwrap_or_else(std::sync::PoisonError::into_inner)
4283 .fail_next_sync = true;
4284 }
4285
4286 fn fail_next_wal_sync(&self) {
4288 self.faults
4289 .lock()
4290 .unwrap_or_else(std::sync::PoisonError::into_inner)
4291 .fail_next_wal_sync = true;
4292 }
4293
4294 fn take_sync_observations(&self) -> Vec<CertificateSyncObservation> {
4295 std::mem::take(
4296 &mut self
4297 .faults
4298 .lock()
4299 .unwrap_or_else(std::sync::PoisonError::into_inner)
4300 .sync_observations,
4301 )
4302 }
4303 }
4304
4305 #[derive(Debug)]
4306 struct CheckpointHandoffFaultFile {
4307 inner: <MemoryVfs as Vfs>::File,
4308 faults: Arc<Mutex<CheckpointHandoffFaultState>>,
4309 path: Option<PathBuf>,
4310 is_checkpoint_handoff: bool,
4311 }
4312
4313 impl Vfs for CheckpointHandoffFaultVfs {
4314 type File = CheckpointHandoffFaultFile;
4315
4316 fn name(&self) -> &'static str {
4317 "checkpoint-handoff-fault"
4318 }
4319
4320 fn open(
4321 &self,
4322 cx: &Cx,
4323 path: Option<&Path>,
4324 flags: VfsOpenFlags,
4325 ) -> Result<(Self::File, VfsOpenFlags)> {
4326 let is_checkpoint_handoff =
4327 path.is_some_and(|candidate| candidate == Path::new(CHECKPOINT_HANDOFF_PATH));
4328 let (inner, actual_flags) = self.inner.open(cx, path, flags)?;
4329 Ok((
4330 CheckpointHandoffFaultFile {
4331 inner,
4332 faults: Arc::clone(&self.faults),
4333 path: path.map(Path::to_path_buf),
4334 is_checkpoint_handoff,
4335 },
4336 actual_flags,
4337 ))
4338 }
4339
4340 fn delete(&self, cx: &Cx, path: &Path, sync_dir: bool) -> Result<()> {
4341 self.inner.delete(cx, path, sync_dir)
4342 }
4343
4344 fn sync_parent_directory(&self, cx: &Cx, path: &Path) -> Result<()> {
4345 self.inner.sync_parent_directory(cx, path)
4346 }
4347
4348 fn access(&self, cx: &Cx, path: &Path, flags: AccessFlags) -> Result<bool> {
4349 self.inner.access(cx, path, flags)
4350 }
4351
4352 fn path_entry_exists(&self, cx: &Cx, path: &Path) -> Result<bool> {
4353 self.inner.path_entry_exists(cx, path)
4354 }
4355
4356 fn full_pathname(&self, cx: &Cx, path: &Path) -> Result<PathBuf> {
4357 self.inner.full_pathname(cx, path)
4358 }
4359
4360 fn randomness(&self, cx: &Cx, buf: &mut [u8]) {
4361 self.inner.randomness(cx, buf);
4362 }
4363
4364 fn current_time(&self, cx: &Cx) -> f64 {
4365 self.inner.current_time(cx)
4366 }
4367
4368 fn is_memory(&self) -> bool {
4369 true
4370 }
4371 }
4372
4373 impl VfsFile for CheckpointHandoffFaultFile {
4374 fn close(&mut self, cx: &Cx) -> Result<()> {
4375 self.inner.close(cx)
4376 }
4377
4378 fn file_identity(&self) -> Result<Option<fsqlite_vfs::FileIdentity>> {
4379 self.inner.file_identity()
4380 }
4381
4382 fn read<'a>(
4383 &'a self,
4384 cx: &'a Cx,
4385 buf: &'a mut [u8],
4386 offset: u64,
4387 ) -> impl std::future::Future<Output = Result<usize>> + Send + 'a {
4388 self.inner.read(cx, buf, offset)
4389 }
4390
4391 async fn write<'a>(&'a self, cx: &'a Cx, buf: &'a [u8], offset: u64) -> Result<()> {
4392 let fault = if self.is_checkpoint_handoff {
4393 self.faults
4394 .lock()
4395 .unwrap_or_else(std::sync::PoisonError::into_inner)
4396 .next_write
4397 .take()
4398 } else {
4399 None
4400 };
4401 match fault {
4402 Some(CheckpointHandoffWriteFault::Error) => Err(FrankenError::Io(
4403 std::io::Error::other("injected checkpoint handoff write failure"),
4404 )),
4405 Some(CheckpointHandoffWriteFault::Pending) => {
4406 std::future::pending::<Result<()>>().await
4407 }
4408 None => self.inner.write(cx, buf, offset).await,
4409 }
4410 }
4411
4412 fn truncate(&mut self, cx: &Cx, size: u64) -> Result<()> {
4413 self.inner.truncate(cx, size)
4414 }
4415
4416 fn sync(&mut self, cx: &Cx, flags: SyncFlags) -> Result<()> {
4417 let mut faults = self
4418 .faults
4419 .lock()
4420 .unwrap_or_else(std::sync::PoisonError::into_inner);
4421 if let Some(path) = self.path.as_ref().filter(|path| {
4422 path.as_path() == Path::new(CERTIFICATE_PATH)
4423 || path.as_path() == Path::new(CHECKPOINT_HANDOFF_PATH)
4424 }) {
4425 faults
4426 .sync_observations
4427 .push(CertificateSyncObservation::Ordinary(path.clone()));
4428 }
4429 let fail = self.is_checkpoint_handoff && std::mem::take(&mut faults.fail_next_sync);
4430 let fail_wal =
4431 !self.is_checkpoint_handoff && std::mem::take(&mut faults.fail_next_wal_sync);
4432 drop(faults);
4433 if fail {
4434 Err(FrankenError::Io(std::io::Error::other(
4435 "injected checkpoint handoff sync failure",
4436 )))
4437 } else if fail_wal {
4438 Err(FrankenError::Io(std::io::Error::other(
4439 "injected WAL sync failure",
4440 )))
4441 } else {
4442 self.inner.sync(cx, flags)
4443 }
4444 }
4445
4446 fn durable_sync(&mut self, cx: &Cx, kind: SyncKind) -> Result<()> {
4447 let mut faults = self
4448 .faults
4449 .lock()
4450 .unwrap_or_else(std::sync::PoisonError::into_inner);
4451 if let Some(path) = self.path.as_ref().filter(|path| {
4452 path.as_path() == Path::new(CERTIFICATE_PATH)
4453 || path.as_path() == Path::new(CHECKPOINT_HANDOFF_PATH)
4454 }) {
4455 faults
4456 .sync_observations
4457 .push(CertificateSyncObservation::Durable(path.clone(), kind));
4458 }
4459 let fail = self.is_checkpoint_handoff && std::mem::take(&mut faults.fail_next_sync);
4460 drop(faults);
4461 if fail {
4462 Err(FrankenError::Io(std::io::Error::other(
4463 "injected checkpoint handoff durable-sync failure",
4464 )))
4465 } else {
4466 self.inner.durable_sync(cx, kind)
4467 }
4468 }
4469
4470 fn file_size(&self, cx: &Cx) -> Result<u64> {
4471 self.inner.file_size(cx)
4472 }
4473
4474 fn lock(&mut self, cx: &Cx, level: fsqlite_types::LockLevel) -> Result<()> {
4475 self.inner.lock(cx, level)
4476 }
4477
4478 fn unlock(&mut self, cx: &Cx, level: fsqlite_types::LockLevel) -> Result<()> {
4479 self.inner.unlock(cx, level)
4480 }
4481
4482 fn lock_external_shared_snapshot(&mut self, cx: &Cx) -> Result<()> {
4483 self.inner.lock_external_shared_snapshot(cx)
4484 }
4485
4486 fn restore_external_shared_snapshot_attempt(&mut self, cx: &Cx) -> Result<()> {
4487 self.inner.restore_external_shared_snapshot_attempt(cx)
4488 }
4489
4490 fn lock_external_maintenance(&mut self, cx: &Cx, wal_mode: bool) -> Result<()> {
4491 self.inner.lock_external_maintenance(cx, wal_mode)
4492 }
4493
4494 fn restore_external_maintenance_attempt(&mut self, cx: &Cx) -> Result<()> {
4495 self.inner.restore_external_maintenance_attempt(cx)
4496 }
4497
4498 fn check_reserved_lock(&self, cx: &Cx) -> Result<bool> {
4499 self.inner.check_reserved_lock(cx)
4500 }
4501
4502 fn sector_size(&self) -> u32 {
4503 self.inner.sector_size()
4504 }
4505
4506 fn device_characteristics(&self) -> u32 {
4507 self.inner.device_characteristics()
4508 }
4509
4510 fn shm_map(
4511 &mut self,
4512 cx: &Cx,
4513 region: u32,
4514 size: u32,
4515 extend: bool,
4516 ) -> Result<fsqlite_vfs::ShmRegion> {
4517 self.inner.shm_map(cx, region, size, extend)
4518 }
4519
4520 fn shm_lock(&mut self, cx: &Cx, offset: u32, n: u32, flags: u32) -> Result<()> {
4521 self.inner.shm_lock(cx, offset, n, flags)
4522 }
4523
4524 fn shm_barrier(&self) {
4525 self.inner.shm_barrier();
4526 }
4527
4528 fn shm_unmap(&mut self, cx: &Cx, delete: bool) -> Result<()> {
4529 self.inner.shm_unmap(cx, delete)
4530 }
4531
4532 fn set_busy_timeout_ms(&mut self, ms: u64) {
4533 self.inner.set_busy_timeout_ms(ms);
4534 }
4535 }
4536
4537 fn init_wal_publication_test_tracing() {}
4553
4554 #[test]
4564 fn wal_publication_tracing_helper_installs_no_global_subscriber() {
4565 let before = tracing::dispatcher::has_been_set();
4566 init_wal_publication_test_tracing();
4567
4568 assert_eq!(
4569 before,
4570 tracing::dispatcher::has_been_set(),
4571 "init_wal_publication_test_tracing must not install or alter a global subscriber"
4572 );
4573 }
4574
4575 fn test_cx() -> Cx {
4576 Cx::default()
4577 }
4578
4579 fn test_salts() -> WalSalts {
4580 WalSalts {
4581 salt1: 0xDEAD_BEEF,
4582 salt2: 0xCAFE_BABE,
4583 }
4584 }
4585
4586 fn sample_page(seed: u8) -> Vec<u8> {
4587 let page_size = usize::try_from(PAGE_SIZE).expect("page size fits usize");
4588 let mut page = vec![0u8; page_size];
4589 for (i, byte) in page.iter_mut().enumerate() {
4590 let reduced = u8::try_from(i % 251).expect("modulo fits u8");
4591 *byte = reduced ^ seed;
4592 }
4593 page
4594 }
4595
4596 fn test_frame_payload_digest(
4597 page_number: u32,
4598 page_data: &[u8],
4599 db_size_if_commit: u32,
4600 ) -> [u8; 32] {
4601 let mut digest = ParallelWalFramePayloadDigestBuilder::new();
4602 digest.update(
4603 PageNumber::new(page_number).expect("test page number must be valid"),
4604 db_size_if_commit,
4605 page_data,
4606 );
4607 digest.finalize()
4608 }
4609
4610 fn sample_certificate(
4611 certificate_epoch: u64,
4612 commit_seq: u64,
4613 lane_record_counts: Vec<u32>,
4614 ) -> ParallelWalCommitCertificate {
4615 let lane_count = u16::try_from(lane_record_counts.len()).expect("test lane count fits u16");
4616 let mut certificate = ParallelWalCommitCertificate {
4617 format_version: fsqlite_wal::PARALLEL_WAL_COMMIT_CERTIFICATE_VERSION,
4618 residue: fsqlite_wal::ParallelWalOrderedResidue::CommitCertificateThenPublish,
4619 certificate_epoch,
4620 commit_seq_lo: fsqlite_types::CommitSeq::new(commit_seq),
4621 commit_seq_hi: fsqlite_types::CommitSeq::new(commit_seq),
4622 durable_segment_epoch: certificate_epoch,
4623 lane_count,
4624 lane_record_counts,
4625 db_size_pages: 1,
4626 page_set_size: 1,
4627 wal_frame_payload_digest: [0xA5; 32],
4628 certificate_crc32c: 0,
4629 fallback_active: false,
4630 };
4631 certificate.certificate_crc32c = certificate.computed_crc32c();
4632 certificate
4633 }
4634
4635 fn make_path_refreshing_backend(
4636 vfs: &MemoryVfs,
4637 cx: &Cx,
4638 ) -> PathRefreshingWalBackend<MemoryVfs> {
4639 let wal = WalFile::create(cx, open_wal_file(vfs, cx), PAGE_SIZE, 0, test_salts())
4640 .expect("create WAL");
4641 PathRefreshingWalBackend::new(
4642 vfs.clone(),
4643 std::path::Path::new("test.db"),
4644 std::path::Path::new("test.db-wal"),
4645 PAGE_SIZE,
4646 wal,
4647 true,
4648 #[cfg(all(feature = "native", any(unix, windows)))]
4649 None,
4650 )
4651 }
4652
4653 fn make_authorized_certificate_backend(
4654 vfs: &MemoryVfs,
4655 cx: &Cx,
4656 ) -> (
4657 PathRefreshingWalBackend<MemoryVfs>,
4658 ParallelWalCommitCertificate,
4659 ) {
4660 let mut backend = make_path_refreshing_backend(vfs, cx);
4661 let committed_page = sample_page(0x44);
4662 let mut certificate = sample_certificate(1, 1, vec![1]);
4663 certificate.wal_frame_payload_digest = test_frame_payload_digest(1, &committed_page, 1);
4664 certificate.certificate_crc32c = certificate.computed_crc32c();
4665 backend
4666 .persist_parallel_wal_commit_certificate(cx, &certificate, 1, 1, true)
4667 .expect("persist authorized certificate");
4668 backend
4669 .append_frame(cx, 1, &committed_page, 1)
4670 .expect("append matching commit marker");
4671 backend.sync(cx).expect("sync matching commit marker");
4672 (backend, certificate)
4673 }
4674
4675 struct AuthoritativeWalSnapshot {
4676 generation: WalGenerationIdentity,
4677 frame_count: usize,
4678 wal_bytes: Vec<u8>,
4679 certificate: ParallelWalCommitCertificate,
4680 committed_page: Vec<u8>,
4681 }
4682
4683 fn make_checkpoint_handoff_fault_backend(
4684 vfs: &CheckpointHandoffFaultVfs,
4685 cx: &Cx,
4686 ) -> (
4687 PathRefreshingWalBackend<CheckpointHandoffFaultVfs>,
4688 ParallelWalCommitCertificate,
4689 Vec<u8>,
4690 ) {
4691 let flags = VfsOpenFlags::READWRITE | VfsOpenFlags::CREATE | VfsOpenFlags::WAL;
4692 let (file, _) = vfs
4693 .open(cx, Some(Path::new("test.db-wal")), flags)
4694 .expect("open fault-injected WAL file");
4695 let wal = WalFile::create(cx, file, PAGE_SIZE, 0, test_salts())
4696 .expect("create fault-injected WAL");
4697 let mut backend = PathRefreshingWalBackend::new(
4698 vfs.clone(),
4699 Path::new("test.db"),
4700 Path::new("test.db-wal"),
4701 PAGE_SIZE,
4702 wal,
4703 true,
4704 #[cfg(all(feature = "native", any(unix, windows)))]
4705 None,
4706 );
4707 let committed_page = sample_page(0x47);
4708 let mut certificate = sample_certificate(1, 1, vec![1]);
4709 certificate.wal_frame_payload_digest = test_frame_payload_digest(1, &committed_page, 1);
4710 certificate.certificate_crc32c = certificate.computed_crc32c();
4711 backend
4712 .persist_parallel_wal_commit_certificate(cx, &certificate, 1, 1, true)
4713 .expect("persist authorized certificate");
4714 backend
4715 .append_frame(cx, 1, &committed_page, 1)
4716 .expect("append matching commit marker");
4717 backend.sync(cx).expect("sync matching commit marker");
4718 (backend, certificate, committed_page)
4719 }
4720
4721 fn read_fault_injected_wal(vfs: &CheckpointHandoffFaultVfs, cx: &Cx) -> Vec<u8> {
4722 let flags = VfsOpenFlags::READONLY | VfsOpenFlags::WAL;
4723 let (mut file, _) = vfs
4724 .open(cx, Some(Path::new("test.db-wal")), flags)
4725 .expect("open WAL snapshot");
4726 let len = usize::try_from(file.file_size(cx).expect("read WAL size"))
4727 .expect("WAL size fits usize");
4728 let mut bytes = vec![0_u8; len];
4729 assert_eq!(
4730 file.read(cx, &mut bytes, 0).expect("read WAL snapshot"),
4731 len
4732 );
4733 file.close(cx).expect("close WAL snapshot");
4734 bytes
4735 }
4736
4737 fn capture_authoritative_wal(
4738 backend: &PathRefreshingWalBackend<CheckpointHandoffFaultVfs>,
4739 vfs: &CheckpointHandoffFaultVfs,
4740 cx: &Cx,
4741 certificate: ParallelWalCommitCertificate,
4742 committed_page: Vec<u8>,
4743 ) -> AuthoritativeWalSnapshot {
4744 AuthoritativeWalSnapshot {
4745 generation: backend.inner.inner().generation_identity(),
4746 frame_count: backend.inner.frame_count(),
4747 wal_bytes: read_fault_injected_wal(vfs, cx),
4748 certificate,
4749 committed_page,
4750 }
4751 }
4752
4753 fn assert_authoritative_wal_unchanged(
4754 backend: &mut PathRefreshingWalBackend<CheckpointHandoffFaultVfs>,
4755 vfs: &CheckpointHandoffFaultVfs,
4756 cx: &Cx,
4757 before: &AuthoritativeWalSnapshot,
4758 ) {
4759 assert_eq!(
4760 backend.inner.inner().generation_identity(),
4761 before.generation,
4762 "checkpoint handoff failure must not reset the WAL generation"
4763 );
4764 assert_eq!(
4765 backend.inner.frame_count(),
4766 before.frame_count,
4767 "checkpoint handoff failure must not change the visible frame count"
4768 );
4769 assert_eq!(
4770 read_fault_injected_wal(vfs, cx),
4771 before.wal_bytes,
4772 "checkpoint handoff failure must leave the authoritative WAL byte-for-byte unchanged"
4773 );
4774 assert!(
4775 backend
4776 .inner
4777 .inner()
4778 .read_frame_header(cx, 0)
4779 .expect("read original commit frame")
4780 .is_commit(),
4781 "the original generation's commit marker must remain authoritative"
4782 );
4783 assert_eq!(
4784 backend
4785 .latest_authorized_parallel_wal_commit_certificate(cx)
4786 .expect("recover certificate from unchanged WAL generation"),
4787 Some(before.certificate.clone())
4788 );
4789 assert_eq!(
4790 backend
4791 .read_page(cx, 1)
4792 .expect("read committed page from unchanged WAL generation"),
4793 Some(before.committed_page.clone())
4794 );
4795 }
4796
4797 fn read_certificate_sidecar(vfs: &MemoryVfs, cx: &Cx) -> Vec<u8> {
4798 let path = std::path::Path::new("test.db-wal-cert");
4799 let (mut file, _) = vfs
4800 .open(cx, Some(path), VfsOpenFlags::READONLY | VfsOpenFlags::WAL)
4801 .expect("open certificate sidecar");
4802 let len = usize::try_from(file.file_size(cx).expect("read certificate sidecar size"))
4803 .expect("certificate sidecar size fits usize");
4804 let mut bytes = vec![0_u8; len];
4805 assert_eq!(
4806 file.read(cx, &mut bytes, 0)
4807 .expect("read certificate sidecar"),
4808 len
4809 );
4810 file.close(cx).expect("close certificate sidecar");
4811 bytes
4812 }
4813
4814 fn replace_certificate_sidecar(vfs: &MemoryVfs, cx: &Cx, bytes: &[u8]) {
4815 let path = std::path::Path::new("test.db-wal-cert");
4816 let flags = VfsOpenFlags::READWRITE | VfsOpenFlags::CREATE | VfsOpenFlags::WAL;
4817 let (mut file, _) = vfs
4818 .open(cx, Some(path), flags)
4819 .expect("open mutable certificate sidecar");
4820 file.truncate(cx, 0)
4821 .expect("truncate mutable certificate sidecar");
4822 file.write(cx, bytes, 0)
4823 .expect("replace certificate sidecar bytes");
4824 file.close(cx).expect("close mutable certificate sidecar");
4825 }
4826
4827 fn assert_wal_corrupt<T: std::fmt::Debug>(result: Result<T>, scenario: &str) {
4828 assert!(
4829 matches!(&result, Err(FrankenError::WalCorrupt { .. })),
4830 "{scenario} must fail closed with WalCorrupt, got {result:?}"
4831 );
4832 }
4833
4834 fn sqlite_page_one(encoded_page_size: u16) -> Vec<u8> {
4835 let mut page = sample_page(0x11);
4836 page[..16].copy_from_slice(b"SQLite format 3\0");
4837 page[16..18].copy_from_slice(&encoded_page_size.to_be_bytes());
4838 page[76..92].fill(0);
4842 page
4843 }
4844
4845 fn write_main_db_pages(vfs: &MemoryVfs, cx: &Cx, pages: &[Vec<u8>]) {
4846 let flags = VfsOpenFlags::READWRITE | VfsOpenFlags::CREATE | VfsOpenFlags::MAIN_DB;
4847 let (mut file, _) = vfs
4848 .open(cx, Some(std::path::Path::new("test.db")), flags)
4849 .expect("open main database");
4850 file.truncate(cx, 0).expect("truncate main database");
4851 for (index, page) in pages.iter().enumerate() {
4852 let offset = u64::try_from(index)
4853 .expect("page index fits u64")
4854 .saturating_mul(u64::from(PAGE_SIZE));
4855 file.write(cx, page, offset).expect("write database page");
4856 }
4857 file.close(cx).expect("close main database");
4858 }
4859
4860 fn replacement_salts() -> WalSalts {
4861 WalSalts {
4862 salt1: 0x1234_5678,
4863 salt2: 0x9ABC_DEF0,
4864 }
4865 }
4866
4867 fn replace_path_visible_wal(vfs: &MemoryVfs, cx: &Cx) {
4868 let wal_path = std::path::Path::new("test.db-wal");
4869 vfs.delete(cx, wal_path, false)
4870 .expect("remove old path-visible WAL");
4871 let file = open_wal_file(vfs, cx);
4872 WalFile::create(cx, file, PAGE_SIZE, 1, replacement_salts())
4873 .expect("create replacement WAL")
4874 .close(cx)
4875 .expect("close replacement WAL");
4876 }
4877
4878 fn append_replacement_wal_page(
4879 vfs: &MemoryVfs,
4880 cx: &Cx,
4881 page_number: u32,
4882 page: &[u8],
4883 db_size_if_commit: u32,
4884 ) {
4885 let file = open_wal_file(vfs, cx);
4886 let wal = WalFile::open(cx, file).expect("open replacement WAL");
4887 let mut adapter = WalBackendAdapter::new(wal);
4888 adapter
4889 .append_frame(cx, page_number, page, db_size_if_commit)
4890 .expect("append replacement WAL page");
4891 adapter.sync(cx).expect("sync replacement WAL page");
4892 adapter
4893 .into_inner()
4894 .expect("sync drained the staged frames")
4895 .close(cx)
4896 .expect("close replacement WAL");
4897 }
4898
4899 fn make_generation_transition_backend(
4900 vfs: &MemoryVfs,
4901 cx: &Cx,
4902 ) -> (
4903 PathRefreshingWalBackend<MemoryVfs>,
4904 TransactionConflictSnapshot,
4905 Vec<u8>,
4906 ) {
4907 let page_one = sqlite_page_one(u16::try_from(PAGE_SIZE).expect("page size fits u16"));
4908 let page_two = sample_page(0x22);
4909 write_main_db_pages(vfs, cx, &[page_one.clone(), page_two.clone()]);
4910
4911 let file = open_wal_file(vfs, cx);
4912 let wal =
4913 WalFile::create(cx, file, PAGE_SIZE, 0, test_salts()).expect("create original WAL");
4914 let mut backend = PathRefreshingWalBackend::new(
4915 vfs.clone(),
4916 std::path::Path::new("test.db"),
4917 std::path::Path::new("test.db-wal"),
4918 PAGE_SIZE,
4919 wal,
4920 true,
4921 #[cfg(all(feature = "native", any(unix, windows)))]
4922 None,
4923 );
4924 backend
4925 .append_frame(cx, 1, &page_one, 0)
4926 .expect("append original page 1");
4927 backend
4928 .append_frame(cx, 2, &page_two, 2)
4929 .expect("append original commit");
4930 backend.sync(cx).expect("publish original commit");
4934 backend
4935 .begin_transaction(cx)
4936 .expect("pin original WAL generation");
4937 let pinned = backend.pinned_read_snapshot().expect("pinned WAL snapshot");
4938 let snapshot = TransactionConflictSnapshot {
4939 generation: pinned.generation,
4940 last_commit_frame: pinned.last_commit_frame,
4941 commit_count: pinned.commit_count,
4942 snapshot_db_size: 0,
4943 };
4944 replace_path_visible_wal(vfs, cx);
4945 (backend, snapshot, page_two)
4946 }
4947
4948 #[test]
4949 fn durable_certificate_sidecar_precedes_and_reconstructs_wal_commit() {
4950 let cx = test_cx();
4951 let vfs = MemoryVfs::new();
4952 let committed_page = sample_page(0x44);
4953 let file = open_wal_file(&vfs, &cx);
4954 let wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create WAL");
4955 let mut backend = PathRefreshingWalBackend::new(
4956 vfs.clone(),
4957 std::path::Path::new("test.db"),
4958 std::path::Path::new("test.db-wal"),
4959 PAGE_SIZE,
4960 wal,
4961 true,
4962 #[cfg(all(feature = "native", any(unix, windows)))]
4963 None,
4964 );
4965 let mut certificate = ParallelWalCommitCertificate {
4966 format_version: fsqlite_wal::PARALLEL_WAL_COMMIT_CERTIFICATE_VERSION,
4967 residue: fsqlite_wal::ParallelWalOrderedResidue::CommitCertificateThenPublish,
4968 certificate_epoch: 1,
4969 commit_seq_lo: fsqlite_types::CommitSeq::new(1),
4970 commit_seq_hi: fsqlite_types::CommitSeq::new(1),
4971 durable_segment_epoch: 1,
4972 lane_count: 1,
4973 lane_record_counts: vec![1],
4974 db_size_pages: 1,
4975 page_set_size: 1,
4976 wal_frame_payload_digest: test_frame_payload_digest(1, &committed_page, 1),
4977 certificate_crc32c: 0,
4978 fallback_active: false,
4979 };
4980 certificate.certificate_crc32c = certificate.computed_crc32c();
4981
4982 backend
4983 .persist_parallel_wal_commit_certificate(&cx, &certificate, 1, 1, true)
4984 .expect("persist certificate before WAL commit marker");
4985 assert_eq!(
4986 backend.inner.frame_count(),
4987 0,
4988 "certificate persistence must not itself expose a WAL commit marker"
4989 );
4990
4991 let certificate_path = std::path::Path::new("test.db-wal-cert");
4992 let (mut certificate_file, _) = vfs
4993 .open(
4994 &cx,
4995 Some(certificate_path),
4996 VfsOpenFlags::READONLY | VfsOpenFlags::WAL,
4997 )
4998 .expect("open certificate sidecar");
4999 let certificate_len = usize::try_from(
5000 certificate_file
5001 .file_size(&cx)
5002 .expect("certificate sidecar size"),
5003 )
5004 .expect("certificate sidecar size fits usize");
5005 let mut record_bytes = vec![0_u8; certificate_len];
5006 assert_eq!(
5007 certificate_file
5008 .read(&cx, &mut record_bytes, 0)
5009 .expect("read certificate sidecar"),
5010 certificate_len
5011 );
5012 certificate_file
5013 .close(&cx)
5014 .expect("close certificate sidecar");
5015 let reconstructed = ParallelWalDurableCertificateRecord::from_bytes(&record_bytes)
5016 .expect("reconstruct durable certificate record");
5017 assert_eq!(reconstructed.certificate, certificate);
5018 assert_eq!(reconstructed.wal_frame_start, 1);
5019 assert_eq!(reconstructed.wal_frame_end, 1);
5020 assert_eq!(
5021 reconstructed.wal_generation,
5022 backend.inner.inner().generation_identity()
5023 );
5024 assert!(
5025 !reconstructed.authorizes_wal_boundary(
5026 backend.inner.inner().generation_identity(),
5027 0,
5028 0,
5029 test_frame_payload_digest(1, &committed_page, 1),
5030 ),
5031 "orphan certificate must not authorize visibility before the matching commit marker"
5032 );
5033
5034 backend
5035 .append_frame(&cx, 1, &committed_page, 1)
5036 .expect("append matching WAL commit marker");
5037 backend.sync(&cx).expect("sync WAL commit marker");
5038 assert!(
5039 backend
5040 .inner
5041 .inner()
5042 .read_frame_header(&cx, 0)
5043 .expect("read matching WAL commit frame")
5044 .is_commit()
5045 );
5046 assert!(reconstructed.authorizes_wal_boundary(
5047 backend.inner.inner().generation_identity(),
5048 1,
5049 1,
5050 test_frame_payload_digest(1, &committed_page, 1),
5051 ));
5052
5053 let (mut certificate_file, _) = vfs
5054 .open(
5055 &cx,
5056 Some(certificate_path),
5057 VfsOpenFlags::READWRITE | VfsOpenFlags::WAL,
5058 )
5059 .expect("reopen certificate sidecar");
5060 let torn_offset = certificate_file
5061 .file_size(&cx)
5062 .expect("certificate sidecar size before torn tail");
5063 certificate_file
5064 .write(&cx, &[0xA5], torn_offset)
5065 .expect("append torn footer byte");
5066 certificate_file
5067 .close(&cx)
5068 .expect("close sidecar with torn tail");
5069 let recovered = backend
5070 .latest_authorized_parallel_wal_commit_certificate(&cx)
5071 .wait()
5072 .expect("torn certificate tail should recover the prior valid record")
5073 .expect("prior authorized certificate should remain discoverable");
5074 assert_eq!(recovered, certificate);
5075 }
5076
5077 #[test]
5078 fn content_mismatched_wal_interval_cannot_be_authorized_or_repaired() {
5079 let cx = test_cx();
5080 let vfs = MemoryVfs::new();
5081 let certified_page = sample_page(0x61);
5082 let actual_page = sample_page(0x62);
5083 let mut backend = make_path_refreshing_backend(&vfs, &cx);
5084 let mut certificate = sample_certificate(1, 1, vec![1]);
5085 certificate.wal_frame_payload_digest = test_frame_payload_digest(1, &certified_page, 1);
5086 certificate.certificate_crc32c = certificate.computed_crc32c();
5087
5088 backend
5089 .persist_parallel_wal_commit_certificate(&cx, &certificate, 1, 1, true)
5090 .expect("persist content-bound certificate");
5091 backend
5092 .append_frame(&cx, 1, &actual_page, 1)
5093 .expect("append differently valued commit frame");
5094 backend.sync(&cx).expect("sync mismatched commit frame");
5095
5096 let sidecar_before = read_certificate_sidecar(&vfs, &cx);
5097 assert!(
5098 backend
5099 .latest_authorized_parallel_wal_commit_certificate(&cx)
5100 .wait()
5101 .expect("content mismatch is a non-authorizing record")
5102 .is_none(),
5103 "matching generation and commit marker must not authorize different frame bytes"
5104 );
5105
5106 assert_wal_corrupt(
5107 backend
5108 .reconcile_parallel_wal_commit(&cx, &certificate, 1, 1, true)
5109 .wait(),
5110 "in-doubt content-bound reconciliation mismatch",
5111 );
5112 assert_eq!(
5113 read_certificate_sidecar(&vfs, &cx),
5114 sidecar_before,
5115 "digest mismatch must be diagnosed before sidecar repair"
5116 );
5117 assert_eq!(
5118 backend.inner.frame_count(),
5119 1,
5120 "digest mismatch must preserve the live WAL for diagnosis and retry"
5121 );
5122 }
5123
5124 #[test]
5125 fn absent_commit_marker_repairs_certificate_and_partial_wal_tail() {
5126 let cx = test_cx();
5127 let vfs = MemoryVfs::new();
5128 let mut backend = make_path_refreshing_backend(&vfs, &cx);
5129 let certificate = sample_certificate(1, 1, vec![1]);
5130 backend
5131 .persist_parallel_wal_commit_certificate(&cx, &certificate, 1, 1, true)
5132 .expect("persist orphan certificate");
5133
5134 let (mut tail_writer, _) = vfs
5135 .open(
5136 &cx,
5137 Some(std::path::Path::new("test.db-wal")),
5138 VfsOpenFlags::READWRITE | VfsOpenFlags::WAL,
5139 )
5140 .expect("open WAL for partial-tail injection");
5141 let committed_size = tail_writer.file_size(&cx).expect("read committed WAL size");
5142 tail_writer
5143 .write(&cx, &[0xA5; 7], committed_size)
5144 .expect("inject a partial physical frame");
5145 assert!(
5146 tail_writer.file_size(&cx).expect("read extended WAL size") > committed_size,
5147 "fault fixture must extend the physical WAL"
5148 );
5149 tail_writer.close(&cx).expect("close partial-tail injector");
5150
5151 assert_eq!(
5152 backend
5153 .reconcile_parallel_wal_commit(&cx, &certificate, 1, 1, true)
5154 .wait()
5155 .expect("missing commit marker must be exactly repairable"),
5156 ParallelWalCommitReconciliation::NotCommitted
5157 );
5158 assert!(
5159 read_certificate_sidecar(&vfs, &cx).is_empty(),
5160 "matching orphan certificate must be removed after NotCommitted proof"
5161 );
5162 let (mut repaired_wal, _) = vfs
5163 .open(
5164 &cx,
5165 Some(std::path::Path::new("test.db-wal")),
5166 VfsOpenFlags::READONLY | VfsOpenFlags::WAL,
5167 )
5168 .expect("open repaired WAL");
5169 assert_eq!(
5170 repaired_wal.file_size(&cx).expect("read repaired WAL size"),
5171 committed_size,
5172 "NotCommitted reconciliation must truncate the physical partial tail"
5173 );
5174 repaired_wal.close(&cx).expect("close repaired WAL");
5175 }
5176
5177 #[test]
5178 fn durable_certificate_recovery_accepts_every_truncated_record_prefix() {
5179 let cx = test_cx();
5180 let vfs = MemoryVfs::new();
5181 let (mut backend, authorized) = make_authorized_certificate_backend(&vfs, &cx);
5182 let authorized_bytes = read_certificate_sidecar(&vfs, &cx);
5183 let orphan = sample_certificate(2, 2, vec![1]);
5184 let orphan_bytes = ParallelWalDurableCertificateRecord::new(
5185 backend.inner.inner().generation_identity(),
5186 2,
5187 2,
5188 [0u8; 16],
5189 orphan,
5190 )
5191 .expect("construct orphan record")
5192 .to_bytes();
5193
5194 for prefix_len in 1..orphan_bytes.len() {
5195 let mut sidecar = authorized_bytes.clone();
5196 sidecar.extend_from_slice(&orphan_bytes[..prefix_len]);
5197 replace_certificate_sidecar(&vfs, &cx, &sidecar);
5198 let recovered_result = backend
5199 .latest_authorized_parallel_wal_commit_certificate(&cx)
5200 .wait();
5201 assert!(
5202 recovered_result.is_ok(),
5203 "truncated certificate prefix of {prefix_len} bytes must recover: {recovered_result:?}"
5204 );
5205 let recovered = recovered_result
5206 .expect("truncated certificate recovery was asserted successful")
5207 .expect("authorized record must remain discoverable");
5208 assert_eq!(recovered, authorized, "failed at prefix {prefix_len}");
5209 }
5210 }
5211
5212 #[test]
5213 fn durable_certificate_append_repairs_the_accepted_torn_suffix() {
5214 let cx = test_cx();
5215 let vfs = MemoryVfs::new();
5216 let (mut backend, authorized) = make_authorized_certificate_backend(&vfs, &cx);
5217 let authorized_bytes = read_certificate_sidecar(&vfs, &cx);
5218 let orphan = sample_certificate(2, 2, vec![1]);
5219 let orphan_bytes = ParallelWalDurableCertificateRecord::new(
5220 backend.inner.inner().generation_identity(),
5221 2,
5222 2,
5223 [0u8; 16],
5224 orphan.clone(),
5225 )
5226 .expect("construct orphan record")
5227 .to_bytes();
5228 for prefix_len in 1..orphan_bytes.len() {
5229 let mut torn_sidecar = authorized_bytes.clone();
5230 torn_sidecar.extend_from_slice(&orphan_bytes[..prefix_len]);
5231 replace_certificate_sidecar(&vfs, &cx, &torn_sidecar);
5232
5233 assert_eq!(
5234 backend
5235 .latest_authorized_parallel_wal_commit_certificate(&cx)
5236 .wait()
5237 .expect("one torn suffix should recover")
5238 .expect("authorized predecessor remains visible"),
5239 authorized,
5240 "read recovery failed for prefix {prefix_len}"
5241 );
5242
5243 backend
5244 .persist_parallel_wal_commit_certificate(&cx, &orphan, 2, 2, true)
5245 .expect("next append repairs the torn suffix first");
5246 let repaired_sidecar = read_certificate_sidecar(&vfs, &cx);
5247 assert_eq!(
5248 repaired_sidecar.len(),
5249 authorized_bytes.len() + orphan_bytes.len(),
5250 "replacement record did not start at the prior complete boundary for prefix {prefix_len}"
5251 );
5252 assert_eq!(
5253 backend
5254 .latest_authorized_parallel_wal_commit_certificate(&cx)
5255 .wait()
5256 .expect("orphan lookback crosses the repaired boundary")
5257 .expect("authorized predecessor remains discoverable"),
5258 authorized,
5259 "orphan lookback failed after repairing prefix {prefix_len}"
5260 );
5261 }
5262
5263 let mut corrupt_record = orphan_bytes;
5264 let envelope_crc_offset =
5265 corrupt_record.len() - ParallelWalDurableCertificateRecord::LENGTH_FOOTER_SIZE - 4;
5266 corrupt_record[envelope_crc_offset] ^= 0x80;
5267 let mut corrupt_sidecar = authorized_bytes;
5268 corrupt_sidecar.extend_from_slice(&corrupt_record);
5269 replace_certificate_sidecar(&vfs, &cx, &corrupt_sidecar);
5270 assert_wal_corrupt(
5271 backend
5272 .persist_parallel_wal_commit_certificate(&cx, &orphan, 2, 2, true)
5273 .wait(),
5274 "append-time complete record corruption",
5275 );
5276 }
5277
5278 #[test]
5279 fn durable_certificate_recovery_rejects_complete_corruption_and_garbage() {
5280 let cx = test_cx();
5281 let vfs = MemoryVfs::new();
5282 let (mut backend, _) = make_authorized_certificate_backend(&vfs, &cx);
5283 let authorized_bytes = read_certificate_sidecar(&vfs, &cx);
5284 let orphan = sample_certificate(2, 2, vec![1]);
5285 let orphan_bytes = ParallelWalDurableCertificateRecord::new(
5286 backend.inner.inner().generation_identity(),
5287 2,
5288 2,
5289 [0u8; 16],
5290 orphan,
5291 )
5292 .expect("construct orphan record")
5293 .to_bytes();
5294
5295 let mut bad_crc = orphan_bytes.clone();
5296 let envelope_crc_offset =
5297 bad_crc.len() - ParallelWalDurableCertificateRecord::LENGTH_FOOTER_SIZE - 4;
5298 bad_crc[envelope_crc_offset] ^= 0x80;
5299 let mut sidecar = authorized_bytes.clone();
5300 sidecar.extend_from_slice(&bad_crc);
5301 replace_certificate_sidecar(&vfs, &cx, &sidecar);
5302 assert_wal_corrupt(
5303 backend
5304 .latest_authorized_parallel_wal_commit_certificate(&cx)
5305 .wait(),
5306 "complete record with bad CRC",
5307 );
5308
5309 let mut bad_version = orphan_bytes.clone();
5310 bad_version[8] ^= 0x01;
5311 let mut sidecar = authorized_bytes.clone();
5312 sidecar.extend_from_slice(&bad_version);
5313 replace_certificate_sidecar(&vfs, &cx, &sidecar);
5314 assert_wal_corrupt(
5315 backend
5316 .latest_authorized_parallel_wal_commit_certificate(&cx)
5317 .wait(),
5318 "complete record with bad version",
5319 );
5320
5321 let mut bad_magic = orphan_bytes.clone();
5322 bad_magic[0] ^= 0x01;
5323 let mut sidecar = authorized_bytes.clone();
5324 sidecar.extend_from_slice(&bad_magic);
5325 replace_certificate_sidecar(&vfs, &cx, &sidecar);
5326 assert_wal_corrupt(
5327 backend
5328 .latest_authorized_parallel_wal_commit_certificate(&cx)
5329 .wait(),
5330 "complete record with bad magic",
5331 );
5332
5333 let mut bad_footer = orphan_bytes;
5334 let footer_offset =
5335 bad_footer.len() - ParallelWalDurableCertificateRecord::LENGTH_FOOTER_SIZE;
5336 bad_footer[footer_offset] ^= 0x80;
5337 let mut sidecar = authorized_bytes;
5338 sidecar.extend_from_slice(&bad_footer);
5339 replace_certificate_sidecar(&vfs, &cx, &sidecar);
5340 assert_wal_corrupt(
5341 backend
5342 .latest_authorized_parallel_wal_commit_certificate(&cx)
5343 .wait(),
5344 "complete record with bad footer",
5345 );
5346
5347 let garbage_vfs = MemoryVfs::new();
5348 let mut garbage_backend = make_path_refreshing_backend(&garbage_vfs, &cx);
5349 replace_certificate_sidecar(&garbage_vfs, &cx, &[0xA5; 128]);
5350 assert_wal_corrupt(
5351 garbage_backend
5352 .latest_authorized_parallel_wal_commit_certificate(&cx)
5353 .wait(),
5354 "nonempty garbage sidecar",
5355 );
5356
5357 let mut fake_magic = vec![0_u8; MIN_DURABLE_CERTIFICATE_RECORD_SIZE];
5358 fake_magic[..PARALLEL_WAL_DURABLE_CERTIFICATE_MAGIC.len()]
5359 .copy_from_slice(&PARALLEL_WAL_DURABLE_CERTIFICATE_MAGIC);
5360 fake_magic[8..10].copy_from_slice(
5361 &fsqlite_wal::PARALLEL_WAL_DURABLE_CERTIFICATE_RECORD_VERSION.to_le_bytes(),
5362 );
5363 let fake_record_len = u32::try_from(fake_magic.len()).expect("fake record length fits u32");
5364 fake_magic[10..14].copy_from_slice(&fake_record_len.to_le_bytes());
5365 let fake_footer_offset =
5366 fake_magic.len() - ParallelWalDurableCertificateRecord::LENGTH_FOOTER_SIZE;
5367 fake_magic[fake_footer_offset..].copy_from_slice(&fake_record_len.to_le_bytes());
5368 replace_certificate_sidecar(&garbage_vfs, &cx, &fake_magic);
5369 assert_wal_corrupt(
5370 garbage_backend
5371 .latest_authorized_parallel_wal_commit_certificate(&cx)
5372 .wait(),
5373 "fake magic and length without a valid envelope",
5374 );
5375 }
5376
5377 #[test]
5378 fn durable_certificate_maximum_size_is_shared_by_writer_and_reader() {
5379 let cx = test_cx();
5380 let vfs = MemoryVfs::new();
5381 let mut backend = make_path_refreshing_backend(&vfs, &cx);
5382 let certificate = sample_certificate(1, 1, vec![1; usize::from(u16::MAX)]);
5383 let record = ParallelWalDurableCertificateRecord::new(
5384 backend.inner.inner().generation_identity(),
5385 1,
5386 1,
5387 [0u8; 16],
5388 certificate.clone(),
5389 )
5390 .expect("construct maximum-size record");
5391 assert_eq!(
5392 record.to_bytes().len(),
5393 PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE
5394 );
5395 backend
5396 .persist_parallel_wal_commit_certificate(&cx, &certificate, 1, 1, true)
5397 .expect("writer accepts maximum-size record");
5398 assert!(
5399 backend
5400 .latest_authorized_parallel_wal_commit_certificate(&cx)
5401 .wait()
5402 .expect("reader accepts maximum-size record")
5403 .is_none(),
5404 "record remains unauthorized until its WAL commit marker exists"
5405 );
5406 }
5407
5408 #[test]
5409 fn durable_certificate_orphan_lookback_allows_exact_boundary_plus_torn_tail() {
5410 let cx = test_cx();
5411 let vfs = MemoryVfs::new();
5412 let (mut backend, authorized) = make_authorized_certificate_backend(&vfs, &cx);
5413 let mut sidecar = read_certificate_sidecar(&vfs, &cx);
5414 for orphan_index in 0..MAX_ORPHAN_CERTIFICATE_LOOKBACK {
5421 let epoch = u64::try_from(orphan_index).expect("orphan index fits u64") + 2;
5422 let orphan = sample_certificate(epoch, epoch, vec![1]);
5423 sidecar.extend_from_slice(
5424 &ParallelWalDurableCertificateRecord::new(
5425 backend.inner.inner().generation_identity(),
5426 1,
5427 1,
5428 [0u8; 16],
5429 orphan,
5430 )
5431 .expect("construct bounded orphan")
5432 .to_bytes(),
5433 );
5434 }
5435 sidecar.push(0xA5);
5436 replace_certificate_sidecar(&vfs, &cx, &sidecar);
5437 assert_eq!(
5438 backend
5439 .latest_authorized_parallel_wal_commit_certificate(&cx)
5440 .wait()
5441 .expect("64 orphans plus one torn suffix remain within bound")
5442 .expect("authorized predecessor is found"),
5443 authorized
5444 );
5445
5446 sidecar.pop();
5447 let overflow_epoch =
5448 u64::try_from(MAX_ORPHAN_CERTIFICATE_LOOKBACK).expect("lookback fits u64") + 2;
5449 let overflow = sample_certificate(overflow_epoch, overflow_epoch, vec![1]);
5450 sidecar.extend_from_slice(
5451 &ParallelWalDurableCertificateRecord::new(
5452 backend.inner.inner().generation_identity(),
5453 1,
5454 1,
5455 [0u8; 16],
5456 overflow,
5457 )
5458 .expect("construct overflow orphan")
5459 .to_bytes(),
5460 );
5461 replace_certificate_sidecar(&vfs, &cx, &sidecar);
5462 assert_wal_corrupt(
5463 backend
5464 .latest_authorized_parallel_wal_commit_certificate(&cx)
5465 .wait(),
5466 "65 unauthorized records",
5467 );
5468
5469 let mut future_sidecar = read_certificate_sidecar(&vfs, &cx);
5473 future_sidecar.truncate(
5474 future_sidecar.len()
5475 - (MAX_ORPHAN_CERTIFICATE_LOOKBACK + 1)
5476 * ParallelWalDurableCertificateRecord::new(
5477 backend.inner.inner().generation_identity(),
5478 1,
5479 1,
5480 [0u8; 16],
5481 sample_certificate(2, 2, vec![1]),
5482 )
5483 .expect("sizing record")
5484 .to_bytes()
5485 .len(),
5486 );
5487 for future_index in 0..=MAX_ORPHAN_CERTIFICATE_LOOKBACK {
5488 let epoch = u64::try_from(future_index).expect("future index fits u64") + 2;
5489 let future = sample_certificate(epoch, epoch, vec![1]);
5490 future_sidecar.extend_from_slice(
5491 &ParallelWalDurableCertificateRecord::new(
5492 backend.inner.inner().generation_identity(),
5493 2,
5494 2,
5495 [0u8; 16],
5496 future,
5497 )
5498 .expect("construct future record")
5499 .to_bytes(),
5500 );
5501 }
5502 future_sidecar.push(0xA5);
5503 replace_certificate_sidecar(&vfs, &cx, &future_sidecar);
5504 assert_eq!(
5505 backend
5506 .latest_authorized_parallel_wal_commit_certificate(&cx)
5507 .wait()
5508 .expect("future-boundary records are budget-exempt")
5509 .expect("authorized predecessor is found beneath futures"),
5510 authorized
5511 );
5512 }
5513
5514 const GH364_IDENTITY_A: [u8; 16] = [0xA1; 16];
5518 const GH364_IDENTITY_B: [u8; 16] = [0xB2; 16];
5520
5521 fn write_identity_stamped_main_db(vfs: &MemoryVfs, cx: &Cx, identity: [u8; 16]) {
5524 let mut page_one = sqlite_page_one(u16::try_from(PAGE_SIZE).expect("page size fits u16"));
5525 page_one[76..92].copy_from_slice(&identity);
5526 write_main_db_pages(vfs, cx, &[page_one]);
5527 }
5528
5529 fn write_checkpoint_handoff(vfs: &MemoryVfs, cx: &Cx, bytes: &[u8]) {
5532 let path = std::path::Path::new("test.db-wal-cert-head");
5533 let flags = VfsOpenFlags::READWRITE | VfsOpenFlags::CREATE | VfsOpenFlags::WAL;
5534 let (mut file, _) = vfs.open(cx, Some(path), flags).expect("open handoff sidecar");
5535 file.truncate(cx, 0).expect("truncate handoff sidecar");
5536 file.write(cx, bytes, 0).expect("write handoff sidecar");
5537 file.close(cx).expect("close handoff sidecar");
5538 }
5539
5540 fn handoff_record_bytes(
5542 backend: &PathRefreshingWalBackend<MemoryVfs>,
5543 db_file_id: [u8; 16],
5544 ) -> Vec<u8> {
5545 let certificate = sample_certificate(1, 1, vec![1]);
5546 ParallelWalDurableCertificateRecord::new(
5547 backend.inner.inner().generation_identity(),
5548 1,
5549 1,
5550 db_file_id,
5551 certificate,
5552 )
5553 .expect("construct handoff record")
5554 .to_bytes()
5555 }
5556
5557 #[test]
5558 fn gh364_checkpoint_handoff_with_foreign_identity_is_absent() {
5559 let cx = test_cx();
5564 let vfs = MemoryVfs::new();
5565 write_identity_stamped_main_db(&vfs, &cx, GH364_IDENTITY_A);
5566 let mut backend = make_path_refreshing_backend(&vfs, &cx);
5567 let stale = handoff_record_bytes(&backend, GH364_IDENTITY_B);
5568 write_checkpoint_handoff(&vfs, &cx, &stale);
5569
5570 let recovered = backend
5571 .latest_authorized_parallel_wal_commit_certificate(&cx)
5572 .wait()
5573 .expect("handoff read must not fail closed on a foreign identity");
5574 assert!(
5575 recovered.is_none(),
5576 "a handoff bound to a foreign db-file identity must be absent, got {recovered:?}"
5577 );
5578 }
5579
5580 #[test]
5581 fn gh364_checkpoint_handoff_with_matching_identity_is_applied() {
5582 let cx = test_cx();
5586 let vfs = MemoryVfs::new();
5587 write_identity_stamped_main_db(&vfs, &cx, GH364_IDENTITY_A);
5588 let mut backend = make_path_refreshing_backend(&vfs, &cx);
5589 let own = handoff_record_bytes(&backend, GH364_IDENTITY_A);
5590 write_checkpoint_handoff(&vfs, &cx, &own);
5591
5592 let recovered = backend
5593 .latest_authorized_parallel_wal_commit_certificate(&cx)
5594 .wait()
5595 .expect("same-identity handoff must recover");
5596 assert_eq!(
5597 recovered,
5598 Some(sample_certificate(1, 1, vec![1])),
5599 "a handoff bound to this database's own identity must be applied"
5600 );
5601 }
5602
5603 #[test]
5604 fn gh364_checkpoint_handoff_with_legacy_v3_version_is_absent_not_fatal() {
5605 let cx = test_cx();
5609 let vfs = MemoryVfs::new();
5610 write_identity_stamped_main_db(&vfs, &cx, GH364_IDENTITY_A);
5611 let mut backend = make_path_refreshing_backend(&vfs, &cx);
5612 let mut record = handoff_record_bytes(&backend, GH364_IDENTITY_A);
5613 record[8..10].copy_from_slice(&3u16.to_le_bytes());
5615 write_checkpoint_handoff(&vfs, &cx, &record);
5616
5617 let recovered = backend
5618 .latest_authorized_parallel_wal_commit_certificate(&cx)
5619 .wait()
5620 .expect("a legacy v3 handoff must be absent, never a fatal error");
5621 assert!(
5622 recovered.is_none(),
5623 "a legacy v3 (identity-less) handoff must be absent, got {recovered:?}"
5624 );
5625 }
5626
5627 #[test]
5628 fn certificate_and_handoff_fences_request_full_durability() {
5629 let cx = test_cx();
5630 let vfs = CheckpointHandoffFaultVfs::new();
5631 let (mut backend, certificate, _) = make_checkpoint_handoff_fault_backend(&vfs, &cx);
5632
5633 assert_eq!(
5634 vfs.take_sync_observations(),
5635 vec![CertificateSyncObservation::Durable(
5636 PathBuf::from(CERTIFICATE_PATH),
5637 SyncKind::FullDurable,
5638 )],
5639 "certificate append must use the strongest durability intent"
5640 );
5641
5642 assert_eq!(
5643 backend
5644 .reconcile_parallel_wal_commit(&cx, &certificate, 1, 1, true)
5645 .wait()
5646 .expect("reconcile committed certificate"),
5647 ParallelWalCommitReconciliation::Authorized
5648 );
5649 assert_eq!(
5650 vfs.take_sync_observations(),
5651 vec![CertificateSyncObservation::Durable(
5652 PathBuf::from(CERTIFICATE_PATH),
5653 SyncKind::FullDurable,
5654 )],
5655 "certificate reconciliation must preserve full durability intent"
5656 );
5657
5658 let record = backend
5659 .latest_authorized_durable_certificate_record(&cx)
5660 .wait()
5661 .expect("read authorized certificate record")
5662 .expect("authorized certificate record must exist");
5663 backend
5664 .persist_checkpoint_certificate_handoff(&cx, &record)
5665 .wait()
5666 .expect("persist checkpoint certificate handoff");
5667 assert_eq!(
5668 vfs.take_sync_observations(),
5669 vec![CertificateSyncObservation::Durable(
5670 PathBuf::from(CHECKPOINT_HANDOFF_PATH),
5671 SyncKind::FullDurable,
5672 )],
5673 "checkpoint handoff must use the strongest durability intent"
5674 );
5675 }
5676
5677 #[test]
5678 fn checkpoint_handoff_write_failure_preserves_authoritative_wal_generation() {
5679 let cx = test_cx();
5680 let vfs = CheckpointHandoffFaultVfs::new();
5681 let (mut backend, certificate, committed_page) =
5682 make_checkpoint_handoff_fault_backend(&vfs, &cx);
5683 let before = capture_authoritative_wal(&backend, &vfs, &cx, certificate, committed_page);
5684 vfs.fail_next_handoff_write();
5685
5686 let mut checkpoint_writer = MockCheckpointPageWriter;
5687 let error = backend
5688 .checkpoint(
5689 &cx,
5690 CheckpointMode::Truncate,
5691 &mut checkpoint_writer,
5692 0,
5693 None,
5694 )
5695 .expect_err("checkpoint must fail before reset when the handoff write fails");
5696 assert!(
5697 error
5698 .to_string()
5699 .contains("injected checkpoint handoff write failure"),
5700 "unexpected handoff write error: {error}"
5701 );
5702 assert_authoritative_wal_unchanged(&mut backend, &vfs, &cx, &before);
5703 }
5704
5705 #[test]
5706 fn checkpoint_handoff_durable_sync_failure_preserves_authoritative_wal_generation() {
5707 let cx = test_cx();
5708 let vfs = CheckpointHandoffFaultVfs::new();
5709 let (mut backend, certificate, committed_page) =
5710 make_checkpoint_handoff_fault_backend(&vfs, &cx);
5711 let before = capture_authoritative_wal(&backend, &vfs, &cx, certificate, committed_page);
5712 vfs.fail_next_handoff_sync();
5713
5714 let mut checkpoint_writer = MockCheckpointPageWriter;
5715 let error = backend
5716 .checkpoint(
5717 &cx,
5718 CheckpointMode::Truncate,
5719 &mut checkpoint_writer,
5720 0,
5721 None,
5722 )
5723 .expect_err("checkpoint must fail before reset when the handoff sync fails");
5724 assert!(
5725 error
5726 .to_string()
5727 .contains("injected checkpoint handoff durable-sync failure"),
5728 "unexpected handoff durable-sync error: {error}"
5729 );
5730 assert_authoritative_wal_unchanged(&mut backend, &vfs, &cx, &before);
5731 }
5732
5733 #[test]
5734 fn dropping_pending_checkpoint_handoff_write_preserves_authoritative_wal_generation() {
5735 let cx = test_cx();
5736 let vfs = CheckpointHandoffFaultVfs::new();
5737 let (mut backend, certificate, committed_page) =
5738 make_checkpoint_handoff_fault_backend(&vfs, &cx);
5739 let before = capture_authoritative_wal(&backend, &vfs, &cx, certificate, committed_page);
5740 vfs.pend_next_handoff_write();
5741
5742 let mut checkpoint_writer = MockCheckpointPageWriter;
5743 let reached_pending_handoff = {
5744 let mut checkpoint = backend.checkpoint(
5745 &cx,
5746 CheckpointMode::Truncate,
5747 &mut checkpoint_writer,
5748 0,
5749 None,
5750 );
5751 let mut task_cx = std::task::Context::from_waker(std::task::Waker::noop());
5752 matches!(
5753 std::future::Future::poll(checkpoint.as_mut(), &mut task_cx),
5754 std::task::Poll::Pending
5755 )
5756 };
5757 assert!(
5758 reached_pending_handoff,
5759 "checkpoint should remain pending inside the injected handoff write"
5760 );
5761 assert_authoritative_wal_unchanged(&mut backend, &vfs, &cx, &before);
5762 }
5763
5764 #[test]
5765 fn two_backend_instances_continue_authorized_certificate_clocks() {
5766 let cx = test_cx();
5767 let vfs = MemoryVfs::new();
5768 let wal = WalFile::create(&cx, open_wal_file(&vfs, &cx), PAGE_SIZE, 0, test_salts())
5769 .expect("create shared WAL");
5770 let mut first_backend = PathRefreshingWalBackend::new(
5771 vfs.clone(),
5772 std::path::Path::new("test.db"),
5773 std::path::Path::new("test.db-wal"),
5774 PAGE_SIZE,
5775 wal,
5776 true,
5777 #[cfg(all(feature = "native", any(unix, windows)))]
5778 None,
5779 );
5780 let request =
5781 |batch_id, wal_frame_payload_digest| fsqlite_wal::ParallelWalDurabilityRequest {
5782 trace_id: batch_id,
5783 scenario_id: "two-instance-continuity".to_owned(),
5784 certificate_epoch: 0,
5785 durable_segment_epoch: 0,
5786 batch_size: 1,
5787 batch_ids: vec![batch_id],
5788 lane_record_counts: vec![1],
5789 db_size_pages: 1,
5790 page_set_size: 1,
5791 control_mode: fsqlite_wal::ParallelWalOperatingMode::Auto,
5792 fallback_reason: None,
5793 checkpoint_active: false,
5794 wal_frame_payload_digest,
5795 };
5796
5797 let first_combiner = fsqlite_wal::ParallelWalDurabilityCombiner::default();
5798 let first_page = sample_page(0x51);
5799 let first_receipt = first_combiner
5800 .certify_and_publish(
5801 request(1, test_frame_payload_digest(1, &first_page, 1)),
5802 |certificate| {
5803 first_backend
5804 .persist_parallel_wal_commit_certificate(&cx, certificate, 1, 1, true)
5805 .wait()
5806 .and_then(|()| first_backend.append_frame(&cx, 1, &first_page, 1).wait())
5807 .and_then(|()| first_backend.sync(&cx))
5808 .map_err(|error| error.to_string())
5809 },
5810 )
5811 .expect("first backend publishes certificate");
5812
5813 let second_wal =
5814 WalFile::open(&cx, open_wal_file(&vfs, &cx)).expect("second backend opens shared WAL");
5815 let mut second_backend = PathRefreshingWalBackend::new(
5816 vfs.clone(),
5817 std::path::Path::new("test.db"),
5818 std::path::Path::new("test.db-wal"),
5819 PAGE_SIZE,
5820 second_wal,
5821 true,
5822 #[cfg(all(feature = "native", any(unix, windows)))]
5823 None,
5824 );
5825
5826 let orphan_combiner = fsqlite_wal::ParallelWalDurabilityCombiner::default();
5830 orphan_combiner
5831 .reconcile_authorized_seed(&first_receipt.certificate)
5832 .expect("seed orphan-producing process");
5833 let orphan_receipt = orphan_combiner
5834 .certify_and_publish(
5835 request(99, test_frame_payload_digest(1, &sample_page(0x52), 1)),
5836 |_| Ok(()),
5837 )
5838 .expect("construct deterministic orphan certificate");
5839 second_backend
5840 .persist_parallel_wal_commit_certificate(&cx, &orphan_receipt.certificate, 2, 2, true)
5841 .expect("persist well-formed orphan certificate tail");
5842 let authorized_seed = second_backend
5843 .latest_authorized_parallel_wal_commit_certificate(&cx)
5844 .expect("second backend performs bounded orphan lookback")
5845 .expect("preceding first certificate remains authorized");
5846 assert_eq!(authorized_seed, first_receipt.certificate);
5847
5848 let second_combiner = fsqlite_wal::ParallelWalDurabilityCombiner::default();
5849 second_combiner
5850 .reconcile_authorized_seed(&authorized_seed)
5851 .expect("seed second process-local combiner");
5852 let second_page = sample_page(0x52);
5853 let second_receipt = second_combiner
5854 .certify_and_publish(
5855 request(2, test_frame_payload_digest(1, &second_page, 1)),
5856 |certificate| {
5857 second_backend
5858 .persist_parallel_wal_commit_certificate(&cx, certificate, 2, 2, true)
5859 .wait()
5860 .and_then(|()| second_backend.append_frame(&cx, 1, &second_page, 1).wait())
5861 .and_then(|()| second_backend.sync(&cx))
5862 .map_err(|error| error.to_string())
5863 },
5864 )
5865 .expect("second backend publishes certificate");
5866
5867 assert_eq!(
5868 second_receipt.certificate.commit_seq_lo.get(),
5869 first_receipt.certificate.commit_seq_hi.get() + 1
5870 );
5871 assert_eq!(
5872 second_receipt.certificate.certificate_epoch,
5873 first_receipt.certificate.certificate_epoch + 1
5874 );
5875 assert_eq!(
5876 second_receipt.certificate, orphan_receipt.certificate,
5877 "continuation may reuse an orphan identity but must not overlap any authorized certificate"
5878 );
5879 let latest = second_backend
5880 .latest_authorized_parallel_wal_commit_certificate(&cx)
5881 .expect("read second bounded authorized tail")
5882 .expect("second certificate is authorized");
5883 assert_eq!(latest, second_receipt.certificate);
5884
5885 let generation_before_checkpoint = second_backend.inner.inner().generation_identity();
5886 let mut checkpoint_writer = MockCheckpointPageWriter;
5887 let checkpoint = second_backend
5888 .checkpoint(
5889 &cx,
5890 CheckpointMode::Truncate,
5891 &mut checkpoint_writer,
5892 0,
5893 None,
5894 )
5895 .expect("truncate checkpoint records certificate clock handoff");
5896 assert!(checkpoint.wal_was_reset);
5897 assert_ne!(
5898 second_backend.inner.inner().generation_identity(),
5899 generation_before_checkpoint
5900 );
5901 let checkpoint_seed = second_backend
5902 .latest_authorized_parallel_wal_commit_certificate(&cx)
5903 .expect("read checkpoint certificate clock handoff")
5904 .expect("reset generation retains the last consumed certificate clock");
5905 assert_eq!(checkpoint_seed, second_receipt.certificate);
5906 second_backend
5907 .begin_transaction(&cx)
5908 .expect("pin reset-generation reader snapshot");
5909 let reset_pinned = second_backend
5910 .pinned_read_snapshot()
5911 .expect("reset-generation reader snapshot");
5912 assert_eq!(
5913 reset_pinned.generation,
5914 second_backend.inner.inner().generation_identity(),
5915 "reader snapshot must bind the reset WAL generation"
5916 );
5917 assert_eq!(
5918 reset_pinned.last_commit_frame, None,
5919 "truncate checkpoint leaves no current-generation commit marker"
5920 );
5921 assert_eq!(
5922 second_backend
5923 .pinned_logical_read_snapshot(&cx)
5924 .expect("inspect reset-generation reader horizon"),
5925 None,
5926 "an earlier-generation checkpoint handoff is a clock seed, never reader visibility"
5927 );
5928
5929 let post_checkpoint_combiner = fsqlite_wal::ParallelWalDurabilityCombiner::default();
5930 post_checkpoint_combiner
5931 .reconcile_authorized_seed(&checkpoint_seed)
5932 .expect("seed fresh post-checkpoint combiner");
5933 let post_checkpoint_page = sample_page(0x53);
5934 let post_checkpoint_receipt = post_checkpoint_combiner
5935 .certify_and_publish(
5936 request(3, test_frame_payload_digest(1, &post_checkpoint_page, 1)),
5937 |certificate| {
5938 second_backend
5939 .persist_parallel_wal_commit_certificate(&cx, certificate, 1, 1, true)
5940 .wait()
5941 .and_then(|()| {
5942 second_backend
5943 .append_frame(&cx, 1, &post_checkpoint_page, 1)
5944 .wait()
5945 })
5946 .and_then(|()| second_backend.sync(&cx))
5947 .map_err(|error| error.to_string())
5948 },
5949 )
5950 .expect("publish first certificate in reset WAL generation");
5951 assert_eq!(
5952 post_checkpoint_receipt.certificate.commit_seq_lo.get(),
5953 second_receipt.certificate.commit_seq_hi.get() + 1
5954 );
5955 assert_eq!(
5956 post_checkpoint_receipt.certificate.certificate_epoch,
5957 second_receipt.certificate.certificate_epoch + 1
5958 );
5959 assert_eq!(
5960 second_backend
5961 .latest_authorized_parallel_wal_commit_certificate(&cx)
5962 .expect("read post-checkpoint current-generation certificate")
5963 .expect("post-checkpoint certificate is authorized"),
5964 post_checkpoint_receipt.certificate
5965 );
5966 second_backend
5967 .begin_transaction(&cx)
5968 .expect("pin post-checkpoint reader snapshot");
5969 let pinned = second_backend
5970 .pinned_read_snapshot()
5971 .expect("post-checkpoint reader snapshot");
5972 let logical = second_backend
5973 .pinned_logical_read_snapshot(&cx)
5974 .expect("inspect post-checkpoint reader horizon")
5975 .expect("current-generation certificate exposes a reader horizon");
5976 assert_eq!(logical.generation, pinned.generation);
5977 assert_eq!(logical.last_commit_frame, pinned.last_commit_frame);
5978 assert_eq!(
5979 logical.visible_commit_seq,
5980 post_checkpoint_receipt.certificate.commit_seq_hi
5981 );
5982 }
5983
5984 #[test]
5985 fn pinned_logical_reader_horizon_counts_physical_tail_after_current_certificate() {
5986 let cx = test_cx();
5987 let vfs = MemoryVfs::new();
5988 let (mut backend, certificate) = make_authorized_certificate_backend(&vfs, &cx);
5989
5990 backend
5991 .begin_transaction(&cx)
5992 .expect("pin certificate reader snapshot");
5993 let initial_pinned = backend
5994 .pinned_read_snapshot()
5995 .expect("initial reader snapshot");
5996 let initial_logical = backend
5997 .pinned_logical_read_snapshot(&cx)
5998 .expect("inspect certificate reader horizon")
5999 .expect("current certificate exposes reader horizon");
6000 assert_eq!(initial_logical.generation, initial_pinned.generation);
6001 assert_eq!(
6002 initial_logical.last_commit_frame,
6003 initial_pinned.last_commit_frame
6004 );
6005 assert_eq!(
6006 initial_logical.visible_commit_seq, certificate.commit_seq_hi,
6007 "certificate horizon is exact when no later physical commit exists"
6008 );
6009
6010 let tail_page = sample_page(0x45);
6011 backend
6012 .append_frame(&cx, 2, &tail_page, 2)
6013 .expect("append later ordinary commit marker");
6014 backend
6015 .sync(&cx)
6016 .expect("sync later ordinary commit marker");
6017 backend
6018 .begin_transaction(&cx)
6019 .expect("repin reader after ordinary tail commit");
6020 let pinned = backend
6021 .pinned_read_snapshot()
6022 .expect("reader snapshot includes ordinary tail commit");
6023 let logical = backend
6024 .pinned_logical_read_snapshot(&cx)
6025 .expect("inspect reader horizon with ordinary tail")
6026 .expect("current certificate remains reader-authoritative");
6027 assert_eq!(logical.generation, pinned.generation);
6028 assert_eq!(logical.last_commit_frame, pinned.last_commit_frame);
6029 assert_eq!(
6030 logical.visible_commit_seq.get(),
6031 certificate.commit_seq_hi.get() + 1
6032 );
6033 }
6034
6035 fn open_wal_file(vfs: &MemoryVfs, cx: &Cx) -> <MemoryVfs as Vfs>::File {
6036 let flags = VfsOpenFlags::READWRITE | VfsOpenFlags::CREATE | VfsOpenFlags::WAL;
6037 let (file, _) = vfs
6038 .open(cx, Some(std::path::Path::new("test.db-wal")), flags)
6039 .expect("open WAL file");
6040 file
6041 }
6042
6043 fn make_adapter(vfs: &MemoryVfs, cx: &Cx) -> WalBackendAdapter<<MemoryVfs as Vfs>::File> {
6044 let file = open_wal_file(vfs, cx);
6045 let wal = WalFile::create(cx, file, PAGE_SIZE, 0, test_salts()).expect("create WAL");
6046 WalBackendAdapter::new(wal)
6047 }
6048
6049 fn make_fault_adapter(
6051 vfs: &CheckpointHandoffFaultVfs,
6052 cx: &Cx,
6053 ) -> WalBackendAdapter<<CheckpointHandoffFaultVfs as Vfs>::File> {
6054 let flags = VfsOpenFlags::READWRITE | VfsOpenFlags::CREATE | VfsOpenFlags::WAL;
6055 let (file, _) = vfs
6056 .open(cx, Some(std::path::Path::new("test.db-wal")), flags)
6057 .expect("open fault WAL file");
6058 let wal = WalFile::create(cx, file, PAGE_SIZE, 0, test_salts()).expect("create fault WAL");
6059 WalBackendAdapter::new(wal)
6060 }
6061
6062 #[test]
6065 fn test_adapter_append_and_frame_count() {
6066 let cx = test_cx();
6067 let vfs = MemoryVfs::new();
6068 let mut adapter = make_adapter(&vfs, &cx);
6069
6070 assert_eq!(adapter.frame_count(), 0);
6071
6072 let page = sample_page(0x42);
6073 adapter
6074 .append_frame(&cx, 1, &page, 0)
6075 .expect("append frame");
6076 assert_eq!(adapter.frame_count(), 1);
6077
6078 adapter
6079 .append_frame(&cx, 2, &sample_page(0x43), 2)
6080 .expect("append commit frame");
6081 assert_eq!(adapter.frame_count(), 2);
6082 }
6083
6084 #[test]
6085 fn test_adapter_read_page_found() {
6086 let cx = test_cx();
6087 let vfs = MemoryVfs::new();
6088 let mut adapter = make_adapter(&vfs, &cx);
6089
6090 let page1 = sample_page(0x10);
6091 let page2 = sample_page(0x20);
6092 adapter.append_frame(&cx, 1, &page1, 0).expect("append");
6093 adapter
6094 .append_frame(&cx, 2, &page2, 2)
6095 .expect("append commit");
6096
6097 assert_eq!(
6100 adapter.read_page(&cx, 1).expect("read staged page 1"),
6101 None,
6102 "staged frames must stay invisible before publication"
6103 );
6104 adapter.sync(&cx).expect("publish staged frames");
6105
6106 let result = adapter.read_page(&cx, 1).expect("read page 1");
6107 assert_eq!(result, Some(page1));
6108
6109 let result = adapter.read_page(&cx, 2).expect("read page 2");
6110 assert_eq!(result, Some(page2));
6111 }
6112
6113 #[test]
6114 fn test_adapter_read_page_not_found() {
6115 let cx = test_cx();
6116 let vfs = MemoryVfs::new();
6117 let mut adapter = make_adapter(&vfs, &cx);
6118
6119 adapter
6120 .append_frame(&cx, 1, &sample_page(0x10), 1)
6121 .expect("append");
6122
6123 let result = adapter.read_page(&cx, 99).expect("read missing page");
6124 assert_eq!(result, None);
6125 }
6126
6127 #[test]
6128 fn test_adapter_read_page_returns_latest_version() {
6129 let cx = test_cx();
6130 let vfs = MemoryVfs::new();
6131 let mut adapter = make_adapter(&vfs, &cx);
6132
6133 let old_data = sample_page(0xAA);
6134 let new_data = sample_page(0xBB);
6135
6136 adapter
6138 .append_frame(&cx, 5, &old_data, 0)
6139 .expect("append old");
6140 adapter
6141 .append_frame(&cx, 5, &new_data, 1)
6142 .expect("append new (commit)");
6143
6144 adapter.sync(&cx).expect("publish staged frames");
6146
6147 let result = adapter.read_page(&cx, 5).expect("read page 5");
6148 assert_eq!(
6149 result,
6150 Some(new_data),
6151 "adapter should return the latest WAL version"
6152 );
6153 }
6154
6155 #[test]
6156 fn test_adapter_refreshes_cross_handle_visibility_and_append_position() {
6157 let cx = test_cx();
6158 let vfs = MemoryVfs::new();
6159
6160 let file1 = open_wal_file(&vfs, &cx);
6161 let wal1 = WalFile::create(&cx, file1, PAGE_SIZE, 0, test_salts()).expect("create WAL");
6162 let mut adapter1 = WalBackendAdapter::new(wal1);
6163
6164 let file2 = open_wal_file(&vfs, &cx);
6165 let wal2 = WalFile::open(&cx, file2).expect("open WAL");
6166 let mut adapter2 = WalBackendAdapter::new(wal2);
6167
6168 let page1 = sample_page(0x11);
6169 adapter1
6170 .append_frame(&cx, 1, &page1, 1)
6171 .expect("adapter1 append commit");
6172 adapter1.sync(&cx).expect("adapter1 sync");
6173 adapter2
6174 .begin_transaction(&cx)
6175 .expect("adapter2 begin transaction");
6176 assert_eq!(
6177 adapter2.read_page(&cx, 1).expect("adapter2 read page1"),
6178 Some(page1.clone()),
6179 "adapter2 should observe adapter1 commit at transaction begin"
6180 );
6181
6182 let page2 = sample_page(0x22);
6183 adapter2
6184 .append_frame(&cx, 2, &page2, 2)
6185 .expect("adapter2 append commit");
6186 adapter2.sync(&cx).expect("adapter2 sync");
6187 adapter1
6188 .begin_transaction(&cx)
6189 .expect("adapter1 begin transaction");
6190 assert_eq!(
6191 adapter1.read_page(&cx, 2).expect("adapter1 read page2"),
6192 Some(page2.clone()),
6193 "adapter1 should observe adapter2 commit at transaction begin"
6194 );
6195
6196 assert_eq!(
6198 adapter1.frame_count(),
6199 2,
6200 "shared WAL should contain both commit frames"
6201 );
6202 assert_eq!(
6203 adapter2.frame_count(),
6204 2,
6205 "shared WAL should contain both commit frames"
6206 );
6207 }
6208
6209 #[test]
6210 fn test_path_refresh_rejects_replacement_wal_page_size_mismatch() {
6211 let cx = test_cx();
6212 let vfs = MemoryVfs::new();
6213 let wal_path = std::path::Path::new("test.db-wal");
6214
6215 let file = open_wal_file(&vfs, &cx);
6216 let wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create WAL");
6217 let mut backend = PathRefreshingWalBackend::new(
6218 vfs.clone(),
6219 std::path::Path::new("test.db"),
6220 wal_path,
6221 PAGE_SIZE,
6222 wal,
6223 true,
6224 #[cfg(all(feature = "native", any(unix, windows)))]
6225 None,
6226 );
6227
6228 backend
6229 .append_frame(&cx, 1, &sample_page(0x31), 1)
6230 .expect("append through live backend");
6231 backend.sync(&cx).expect("sync live backend");
6232
6233 vfs.delete(&cx, wal_path, false)
6234 .expect("remove path-visible WAL");
6235 let flags = VfsOpenFlags::READWRITE | VfsOpenFlags::CREATE | VfsOpenFlags::WAL;
6236 let (replacement_file, _) = vfs
6237 .open(&cx, Some(wal_path), flags)
6238 .expect("open replacement WAL path");
6239 let replacement_page_size = PAGE_SIZE
6240 .checked_mul(2)
6241 .expect("test replacement page size fits u32");
6242 let replacement_wal = WalFile::create(
6243 &cx,
6244 replacement_file,
6245 replacement_page_size,
6246 0,
6247 test_salts(),
6248 )
6249 .expect("create mismatched replacement WAL");
6250 replacement_wal.close(&cx).expect("close replacement WAL");
6251
6252 let err = backend
6253 .begin_transaction(&cx)
6254 .expect_err("path refresh should reject mismatched WAL page size");
6255 assert!(
6256 matches!(
6257 err,
6258 FrankenError::WalCorrupt { ref detail }
6259 if detail.contains("does not match database page size")
6260 && detail.contains("during path refresh")
6261 ),
6262 "unexpected error: {err:?}"
6263 );
6264 }
6265
6266 #[test]
6267 fn test_generation_change_allows_identical_full_page_baseline() {
6268 let cx = test_cx();
6269 let vfs = MemoryVfs::new();
6270 let (mut backend, snapshot, page_two) = make_generation_transition_backend(&vfs, &cx);
6271 let baseline = TransactionConflictPageBaseline {
6272 page_number: 2,
6273 page_hash: *blake3::hash(&page_two).as_bytes(),
6274 };
6275
6276 let conflicts = backend
6277 .conflicting_pages_since_snapshot(&cx, snapshot, &[2], &[baseline])
6278 .expect("validate checkpoint-only generation transition");
6279 assert!(
6280 conflicts.is_empty(),
6281 "byte-identical checkpoint-only reset must not create a false conflict"
6282 );
6283 }
6284
6285 #[test]
6292 fn test_generation_change_cached_verification_fd_reads_live_main_db_content() {
6293 let cx = test_cx();
6294 let vfs = MemoryVfs::new();
6295 let (mut backend, snapshot, page_two) = make_generation_transition_backend(&vfs, &cx);
6296 let baseline = TransactionConflictPageBaseline {
6297 page_number: 2,
6298 page_hash: *blake3::hash(&page_two).as_bytes(),
6299 };
6300
6301 let first = backend
6304 .conflicting_pages_since_snapshot(&cx, snapshot, &[2], &[baseline])
6305 .expect("first (cache-populating) generation-change verification");
6306 assert!(
6307 first.is_empty(),
6308 "identical baseline must not conflict on the first check"
6309 );
6310
6311 write_main_db_pages(
6315 &vfs,
6316 &cx,
6317 &[
6318 sqlite_page_one(u16::try_from(PAGE_SIZE).expect("page size fits u16")),
6319 sample_page(0x33),
6320 ],
6321 );
6322
6323 let second = backend
6324 .conflicting_pages_since_snapshot(&cx, snapshot, &[2], &[baseline])
6325 .expect("second generation-change verification reuses the cached fd");
6326 assert_eq!(
6327 second,
6328 vec![2],
6329 "cached verification fd must read the live changed page, not stale cached bytes"
6330 );
6331 }
6332
6333 #[test]
6334 fn test_generation_change_rejects_changed_candidate_page() {
6335 let cx = test_cx();
6336 let vfs = MemoryVfs::new();
6337 let (mut backend, snapshot, page_two) = make_generation_transition_backend(&vfs, &cx);
6338 let changed_page_two = sample_page(0x33);
6339 write_main_db_pages(
6340 &vfs,
6341 &cx,
6342 &[
6343 sqlite_page_one(u16::try_from(PAGE_SIZE).expect("page size fits u16")),
6344 changed_page_two,
6345 ],
6346 );
6347 let baseline = TransactionConflictPageBaseline {
6348 page_number: 2,
6349 page_hash: *blake3::hash(&page_two).as_bytes(),
6350 };
6351
6352 let conflicts = backend
6353 .conflicting_pages_since_snapshot(&cx, snapshot, &[2], &[baseline])
6354 .expect("validate changed page across generation transition");
6355 assert_eq!(conflicts, vec![2]);
6356 }
6357
6358 #[test]
6359 fn test_generation_change_rejects_changed_candidate_from_replacement_wal() {
6360 let cx = test_cx();
6361 let vfs = MemoryVfs::new();
6362 let (mut backend, snapshot, page_two) = make_generation_transition_backend(&vfs, &cx);
6363 append_replacement_wal_page(&vfs, &cx, 2, &sample_page(0x44), 2);
6364 let baseline = TransactionConflictPageBaseline {
6365 page_number: 2,
6366 page_hash: *blake3::hash(&page_two).as_bytes(),
6367 };
6368
6369 let conflicts = backend
6370 .conflicting_pages_since_snapshot(&cx, snapshot, &[2], &[baseline])
6371 .expect("replacement WAL page must take precedence over identical main page");
6372 assert_eq!(conflicts, vec![2]);
6373 }
6374
6375 #[test]
6376 fn test_generation_change_rejects_missing_baseline() {
6377 let cx = test_cx();
6378 let vfs = MemoryVfs::new();
6379 let (mut backend, snapshot, _) = make_generation_transition_backend(&vfs, &cx);
6380
6381 let conflicts = backend
6382 .conflicting_pages_since_snapshot(&cx, snapshot, &[2], &[])
6383 .expect("missing baseline must fail closed");
6384 assert_eq!(conflicts, vec![2]);
6385 }
6386
6387 #[test]
6388 fn test_generation_change_rejects_conflicting_duplicate_baselines() {
6389 let cx = test_cx();
6390 let vfs = MemoryVfs::new();
6391 let (mut backend, snapshot, page_two) = make_generation_transition_backend(&vfs, &cx);
6392 let baselines = [
6393 TransactionConflictPageBaseline {
6394 page_number: 2,
6395 page_hash: *blake3::hash(&page_two).as_bytes(),
6396 },
6397 TransactionConflictPageBaseline {
6398 page_number: 2,
6399 page_hash: *blake3::hash(&sample_page(0x55)).as_bytes(),
6400 },
6401 ];
6402
6403 let conflicts = backend
6404 .conflicting_pages_since_snapshot(&cx, snapshot, &[2], &baselines)
6405 .expect("conflicting duplicate baselines must fail closed");
6406 assert_eq!(conflicts, vec![2]);
6407 }
6408
6409 #[test]
6410 fn test_generation_change_rejects_short_candidate_page() {
6411 let cx = test_cx();
6412 let vfs = MemoryVfs::new();
6413 let (mut backend, snapshot, page_two) = make_generation_transition_backend(&vfs, &cx);
6414 write_main_db_pages(
6415 &vfs,
6416 &cx,
6417 &[sqlite_page_one(
6418 u16::try_from(PAGE_SIZE).expect("page size fits u16"),
6419 )],
6420 );
6421 let baseline = TransactionConflictPageBaseline {
6422 page_number: 2,
6423 page_hash: *blake3::hash(&page_two).as_bytes(),
6424 };
6425
6426 let conflicts = backend
6427 .conflicting_pages_since_snapshot(&cx, snapshot, &[2], &[baseline])
6428 .expect("short page must fail closed");
6429 assert_eq!(conflicts, vec![2]);
6430 }
6431
6432 #[test]
6433 fn test_generation_change_rejects_database_page_size_change() {
6434 let cx = test_cx();
6435 let vfs = MemoryVfs::new();
6436 let (mut backend, snapshot, page_two) = make_generation_transition_backend(&vfs, &cx);
6437 write_main_db_pages(&vfs, &cx, &[sqlite_page_one(8192), page_two.clone()]);
6438 let baseline = TransactionConflictPageBaseline {
6439 page_number: 2,
6440 page_hash: *blake3::hash(&page_two).as_bytes(),
6441 };
6442
6443 let conflicts = backend
6444 .conflicting_pages_since_snapshot(&cx, snapshot, &[2], &[baseline])
6445 .expect("page-size change must fail closed");
6446 assert_eq!(conflicts, vec![2]);
6447 }
6448
6449 #[test]
6450 fn test_generation_change_decodes_64k_database_header_sentinel() {
6451 assert_eq!(
6452 sqlite_database_header_page_size(&sqlite_page_one(1)),
6453 Some(65_536)
6454 );
6455 }
6456
6457 #[test]
6458 fn test_adapter_batch_append_checksum_chain_matches_single_append() {
6459 let cx = test_cx();
6460 let vfs_single = MemoryVfs::new();
6461 let vfs_batch = MemoryVfs::new();
6462
6463 let mut adapter_single = make_adapter(&vfs_single, &cx);
6464 let mut adapter_batch = make_adapter(&vfs_batch, &cx);
6465
6466 let pages: Vec<Vec<u8>> = (0..4u8).map(sample_page).collect();
6467 let commit_sizes = [0_u32, 0, 0, 4];
6468
6469 for (index, page) in pages.iter().enumerate() {
6470 adapter_single
6471 .append_frame(
6472 &cx,
6473 u32::try_from(index + 1).expect("page number fits u32"),
6474 page,
6475 commit_sizes[index],
6476 )
6477 .expect("single append");
6478 }
6479
6480 let batch_frames: Vec<_> = pages
6481 .iter()
6482 .enumerate()
6483 .map(|(index, page)| WalFrameRef {
6484 page_number: u32::try_from(index + 1).expect("page number fits u32"),
6485 page_data: page,
6486 db_size_if_commit: commit_sizes[index],
6487 })
6488 .collect();
6489 adapter_batch
6490 .append_frames(&cx, &batch_frames)
6491 .expect("batch append");
6492
6493 assert_eq!(
6494 adapter_single.frame_count(),
6495 adapter_batch.frame_count(),
6496 "batch adapter append must preserve frame count"
6497 );
6498 assert_eq!(
6499 adapter_single.wal.running_checksum(),
6500 adapter_batch.wal.running_checksum(),
6501 "batch adapter append must preserve checksum chain"
6502 );
6503
6504 for frame_index in 0..pages.len() {
6505 let (single_header, single_data) = adapter_single
6506 .wal
6507 .read_frame(&cx, frame_index)
6508 .expect("read single frame");
6509 let (batch_header, batch_data) = adapter_batch
6510 .wal
6511 .read_frame(&cx, frame_index)
6512 .expect("read batch frame");
6513 assert_eq!(
6514 single_header, batch_header,
6515 "frame header {frame_index} must match"
6516 );
6517 assert_eq!(
6518 single_data, batch_data,
6519 "frame payload {frame_index} must match"
6520 );
6521 }
6522 }
6523
6524 #[test]
6525 fn test_adapter_prepared_batch_append_checksum_chain_matches_single_append() {
6526 let cx = test_cx();
6527 let vfs_single = MemoryVfs::new();
6528 let vfs_prepared = MemoryVfs::new();
6529
6530 let mut adapter_single = make_adapter(&vfs_single, &cx);
6531 let mut adapter_prepared = make_adapter(&vfs_prepared, &cx);
6532
6533 let pages: Vec<Vec<u8>> = (0..4u8).map(sample_page).collect();
6534 let commit_sizes = [0_u32, 0, 0, 4];
6535
6536 for (index, page) in pages.iter().enumerate() {
6537 adapter_single
6538 .append_frame(
6539 &cx,
6540 u32::try_from(index + 1).expect("page number fits u32"),
6541 page,
6542 commit_sizes[index],
6543 )
6544 .expect("single append");
6545 }
6546
6547 let batch_frames: Vec<_> = pages
6548 .iter()
6549 .enumerate()
6550 .map(|(index, page)| WalFrameRef {
6551 page_number: u32::try_from(index + 1).expect("page number fits u32"),
6552 page_data: page,
6553 db_size_if_commit: commit_sizes[index],
6554 })
6555 .collect();
6556 let mut prepared = adapter_prepared
6557 .prepare_append_frames(&batch_frames)
6558 .expect("prepare append")
6559 .expect("prepared batch");
6560 adapter_prepared
6561 .append_prepared_frames(&cx, &mut prepared)
6562 .expect("append prepared");
6563
6564 assert_eq!(
6565 adapter_single.frame_count(),
6566 adapter_prepared.frame_count(),
6567 "prepared adapter append must preserve frame count"
6568 );
6569 assert_eq!(
6570 adapter_single.wal.running_checksum(),
6571 adapter_prepared.wal.running_checksum(),
6572 "prepared adapter append must preserve checksum chain"
6573 );
6574
6575 for frame_index in 0..pages.len() {
6576 let (single_header, single_data) = adapter_single
6577 .wal
6578 .read_frame(&cx, frame_index)
6579 .expect("read single frame");
6580 let (prepared_header, prepared_data) = adapter_prepared
6581 .wal
6582 .read_frame(&cx, frame_index)
6583 .expect("read prepared frame");
6584 assert_eq!(
6585 single_header, prepared_header,
6586 "frame header {frame_index} must match"
6587 );
6588 assert_eq!(
6589 single_data, prepared_data,
6590 "frame payload {frame_index} must match"
6591 );
6592 }
6593 }
6594
6595 #[test]
6596 fn test_adapter_pre_finalize_reused_when_append_window_is_stable() {
6597 let cx = test_cx();
6598 let vfs_single = MemoryVfs::new();
6599 let vfs_prepared = MemoryVfs::new();
6600
6601 let mut adapter_single = make_adapter(&vfs_single, &cx);
6602 let mut adapter_prepared = make_adapter(&vfs_prepared, &cx);
6603
6604 let pages: Vec<Vec<u8>> = (0..3u8).map(sample_page).collect();
6605 let commit_sizes = [0_u32, 0, 3];
6606
6607 for (index, page) in pages.iter().enumerate() {
6608 adapter_single
6609 .append_frame(
6610 &cx,
6611 u32::try_from(index + 1).expect("page number fits u32"),
6612 page,
6613 commit_sizes[index],
6614 )
6615 .expect("single append");
6616 }
6617
6618 let batch_frames: Vec<_> = pages
6619 .iter()
6620 .enumerate()
6621 .map(|(index, page)| WalFrameRef {
6622 page_number: u32::try_from(index + 1).expect("page number fits u32"),
6623 page_data: page,
6624 db_size_if_commit: commit_sizes[index],
6625 })
6626 .collect();
6627 let mut prepared = adapter_prepared
6628 .prepare_append_frames(&batch_frames)
6629 .expect("prepare append")
6630 .expect("prepared batch");
6631 adapter_prepared
6632 .finalize_prepared_frames(&cx, &mut prepared)
6633 .expect("pre-finalize prepared batch");
6634 let finalized_for = prepared.finalized_for.expect("finalization state");
6635 let finalized_running_checksum = prepared
6636 .finalized_running_checksum
6637 .expect("finalized checksum");
6638
6639 adapter_prepared
6640 .append_prepared_frames(&cx, &mut prepared)
6641 .expect("append prepared");
6642
6643 assert_eq!(
6644 prepared.finalized_for,
6645 Some(finalized_for),
6646 "stable append window should reuse the pre-lock finalization state"
6647 );
6648 assert_eq!(
6649 prepared.finalized_running_checksum,
6650 Some(finalized_running_checksum),
6651 "stable append window should reuse the pre-lock finalized checksum"
6652 );
6653 assert_eq!(
6654 adapter_single.wal.running_checksum(),
6655 adapter_prepared.wal.running_checksum(),
6656 "stable reuse path must preserve checksum chain"
6657 );
6658 }
6659
6660 #[test]
6661 fn test_adapter_pre_finalize_reseeds_after_intervening_external_append() {
6662 let cx = test_cx();
6663 let baseline_vfs = MemoryVfs::new();
6664 let shared_vfs = MemoryVfs::new();
6665
6666 let mut baseline = make_adapter(&baseline_vfs, &cx);
6667 let mut prepared_writer = make_adapter(&shared_vfs, &cx);
6668 let intruder_file = open_wal_file(&shared_vfs, &cx);
6669 let intruder_wal = WalFile::open(&cx, intruder_file).expect("open shared WAL");
6670 let mut intruder = WalBackendAdapter::new(intruder_wal);
6671
6672 let pages: Vec<Vec<u8>> = (0..3u8).map(sample_page).collect();
6673 let commit_sizes = [0_u32, 0, 3];
6674 let intruder_page = sample_page(0xEE);
6675
6676 baseline
6677 .append_frame(&cx, 99, &intruder_page, 1)
6678 .expect("baseline intruder append");
6679 for (index, page) in pages.iter().enumerate() {
6680 baseline
6681 .append_frame(
6682 &cx,
6683 u32::try_from(index + 1).expect("page number fits u32"),
6684 page,
6685 commit_sizes[index],
6686 )
6687 .expect("baseline append");
6688 }
6689
6690 let batch_frames: Vec<_> = pages
6691 .iter()
6692 .enumerate()
6693 .map(|(index, page)| WalFrameRef {
6694 page_number: u32::try_from(index + 1).expect("page number fits u32"),
6695 page_data: page,
6696 db_size_if_commit: commit_sizes[index],
6697 })
6698 .collect();
6699 let mut prepared = prepared_writer
6700 .prepare_append_frames(&batch_frames)
6701 .expect("prepare append")
6702 .expect("prepared batch");
6703 prepared_writer
6704 .finalize_prepared_frames(&cx, &mut prepared)
6705 .expect("pre-finalize prepared batch");
6706 let stale_finalization_state = prepared.finalized_for;
6707
6708 intruder
6709 .append_frame(&cx, 99, &intruder_page, 1)
6710 .expect("intruder append");
6711 intruder.sync(&cx).expect("intruder sync");
6712
6713 prepared_writer
6714 .append_prepared_frames(&cx, &mut prepared)
6715 .expect("append prepared after external growth");
6716
6717 assert_ne!(
6718 prepared.finalized_for, stale_finalization_state,
6719 "intervening external growth should force prepared batch reseeding"
6720 );
6721 assert_eq!(
6722 baseline.wal.running_checksum(),
6723 prepared_writer.wal.running_checksum(),
6724 "reseeding path must preserve checksum chain"
6725 );
6726 assert_eq!(
6727 baseline.frame_count(),
6728 prepared_writer.frame_count(),
6729 "reseeding path must preserve frame count"
6730 );
6731 }
6732
6733 #[test]
6734 fn test_adapter_pins_read_snapshot_until_next_begin() {
6735 init_wal_publication_test_tracing();
6736 let cx = test_cx();
6737 let vfs = MemoryVfs::new();
6738
6739 let file_writer = open_wal_file(&vfs, &cx);
6740 let wal_writer =
6741 WalFile::create(&cx, file_writer, PAGE_SIZE, 0, test_salts()).expect("create WAL");
6742 let mut writer = WalBackendAdapter::new(wal_writer);
6743
6744 let file_reader = open_wal_file(&vfs, &cx);
6745 let wal_reader = WalFile::open(&cx, file_reader).expect("open WAL");
6746 let mut reader = WalBackendAdapter::new(wal_reader);
6747
6748 let v1 = sample_page(0x41);
6749 writer.append_frame(&cx, 3, &v1, 3).expect("append v1");
6750 writer.sync(&cx).expect("sync v1");
6751
6752 reader
6753 .begin_transaction(&cx)
6754 .expect("begin reader snapshot 1");
6755 let pinned_v1 = reader
6756 .pinned_read_snapshot()
6757 .expect("reader pins publication snapshot");
6758 assert_eq!(pinned_v1.last_commit_frame, Some(0));
6759 assert_eq!(pinned_v1.commit_count, 1);
6760 assert_eq!(pinned_v1.latest_frame_entries, 1);
6761 assert!(pinned_v1.lookup_contract_is_authoritative());
6762 assert_eq!(
6763 reader.read_page(&cx, 3).expect("reader sees v1"),
6764 Some(v1.clone())
6765 );
6766
6767 let v2 = sample_page(0x42);
6768 writer.append_frame(&cx, 3, &v2, 3).expect("append v2");
6769 writer.sync(&cx).expect("sync v2");
6770
6771 assert_eq!(
6773 reader
6774 .read_page(&cx, 3)
6775 .expect("reader remains on pinned snapshot"),
6776 Some(v1.clone())
6777 );
6778 assert_eq!(
6779 reader
6780 .pinned_read_snapshot()
6781 .expect("reader keeps the same pinned snapshot"),
6782 pinned_v1,
6783 "pinned publication metadata must stay stable until the next begin"
6784 );
6785
6786 reader
6788 .begin_transaction(&cx)
6789 .expect("begin reader snapshot 2");
6790 let pinned_v2 = reader
6791 .pinned_read_snapshot()
6792 .expect("reader repins publication snapshot");
6793 assert!(pinned_v2.publication_seq > pinned_v1.publication_seq);
6794 assert_eq!(pinned_v2.commit_count, 2);
6795 assert_eq!(pinned_v2.latest_frame_entries, 1);
6796 assert_eq!(reader.read_page(&cx, 3).expect("reader sees v2"), Some(v2));
6797 }
6798
6799 #[test]
6800 fn test_adapter_read_page_hides_uncommitted_frames() {
6801 let cx = test_cx();
6802 let vfs = MemoryVfs::new();
6803 let mut adapter = make_adapter(&vfs, &cx);
6804
6805 let committed = sample_page(0x31);
6806 let uncommitted = sample_page(0x32);
6807
6808 adapter
6809 .append_frame(&cx, 7, &committed, 7)
6810 .expect("append committed frame");
6811 adapter.sync(&cx).expect("publish committed frame");
6814 adapter
6815 .append_frame(&cx, 7, &uncommitted, 0)
6816 .expect("append uncommitted frame");
6817
6818 let result = adapter.read_page(&cx, 7).expect("read committed page");
6819 assert_eq!(
6820 result,
6821 Some(committed),
6822 "reader must ignore uncommitted (and unpublished) tail frames"
6823 );
6824 }
6825
6826 #[test]
6827 fn test_adapter_read_page_none_when_wal_has_no_commit_frame() {
6828 let cx = test_cx();
6829 let vfs = MemoryVfs::new();
6830 let mut adapter = make_adapter(&vfs, &cx);
6831
6832 adapter
6833 .append_frame(&cx, 3, &sample_page(0x44), 0)
6834 .expect("append uncommitted frame");
6835
6836 let result = adapter.read_page(&cx, 3).expect("read page");
6837 assert_eq!(result, None, "uncommitted WAL frames must stay invisible");
6838 }
6839
6840 #[test]
6841 fn test_adapter_read_page_empty_wal() {
6842 let cx = test_cx();
6843 let vfs = MemoryVfs::new();
6844 let mut adapter = make_adapter(&vfs, &cx);
6845
6846 let result = adapter.read_page(&cx, 1).expect("read from empty WAL");
6847 assert_eq!(result, None);
6848 }
6849
6850 #[test]
6851 fn test_adapter_sync() {
6852 let cx = test_cx();
6853 let vfs = MemoryVfs::new();
6854 let mut adapter = make_adapter(&vfs, &cx);
6855
6856 adapter
6857 .append_frame(&cx, 1, &sample_page(0), 1)
6858 .expect("append");
6859 adapter.sync(&cx).expect("sync should not fail");
6860 }
6861
6862 #[test]
6863 fn test_adapter_into_inner_fails_closed_until_sync() {
6864 let cx = test_cx();
6865 let staged_vfs = MemoryVfs::new();
6866 let mut staged = make_adapter(&staged_vfs, &cx);
6867
6868 staged
6869 .append_frame(&cx, 1, &sample_page(0), 1)
6870 .expect("append");
6871 assert!(
6872 matches!(staged.into_inner(), Err(FrankenError::Busy)),
6873 "an unsynced commit must prevent consuming the adapter"
6874 );
6875
6876 let synced_vfs = MemoryVfs::new();
6877 let mut synced = make_adapter(&synced_vfs, &cx);
6878 synced
6879 .append_frame(&cx, 1, &sample_page(0), 1)
6880 .expect("append");
6881 synced.sync(&cx).expect("sync staged commit");
6882
6883 assert_eq!(synced.inner().frame_count(), 1);
6884
6885 let wal = synced.into_inner().expect("sync drained the staged frames");
6886 assert_eq!(wal.frame_count(), 1);
6887 }
6888
6889 #[test]
6890 fn test_adapter_as_dyn_wal_backend() {
6891 let cx = test_cx();
6892 let vfs = MemoryVfs::new();
6893 let mut adapter = make_adapter(&vfs, &cx);
6894
6895 let backend: &mut dyn WalBackend = &mut adapter;
6897 backend
6898 .append_frame(&cx, 1, &sample_page(0x77), 1)
6899 .expect("append via dyn");
6900 assert_eq!(backend.frame_count(), 1);
6901
6902 backend.sync(&cx).expect("publish via dyn");
6904 let page = backend.read_page(&cx, 1).expect("read via dyn");
6905 assert_eq!(page, Some(sample_page(0x77)));
6906 }
6907
6908 #[test]
6909 fn test_publication_snapshots_are_visible_through_wal_backend_trait() {
6910 init_wal_publication_test_tracing();
6911 let cx = test_cx();
6912 let vfs = MemoryVfs::new();
6913
6914 let file_writer = open_wal_file(&vfs, &cx);
6915 let wal_writer =
6916 WalFile::create(&cx, file_writer, PAGE_SIZE, 0, test_salts()).expect("create WAL");
6917 let mut writer = WalBackendAdapter::new(wal_writer);
6918
6919 writer
6920 .append_frame(&cx, 4, &sample_page(0x84), 4)
6921 .expect("append committed frame");
6922 writer.sync(&cx).expect("sync committed frame");
6923
6924 let file_reader = open_wal_file(&vfs, &cx);
6925 let wal_reader = WalFile::open(&cx, file_reader).expect("open WAL");
6926 let mut reader = WalBackendAdapter::new(wal_reader);
6927 let backend: &mut dyn WalBackend = &mut reader;
6928
6929 let published_before = backend
6930 .published_snapshot()
6931 .expect("trait should expose the adapter publication summary");
6932 assert_eq!(published_before.last_commit_frame, None);
6933 assert_eq!(published_before.commit_count, 0);
6934
6935 let refreshed = backend
6936 .refresh_published_snapshot(&cx)
6937 .expect("refresh through trait should succeed")
6938 .expect("adapter should republish an existing committed prefix");
6939 assert_eq!(refreshed.last_commit_frame, Some(0));
6940 assert_eq!(refreshed.commit_count, 1);
6941 assert_eq!(refreshed.latest_frame_entries, 1);
6942
6943 backend
6944 .begin_transaction(&cx)
6945 .expect("begin_transaction through trait should pin snapshot");
6946 let pinned = backend
6947 .pinned_read_snapshot()
6948 .expect("trait should expose the pinned read snapshot");
6949 assert_eq!(pinned, refreshed);
6950 }
6951
6952 #[test]
6955 fn test_page_index_returns_correct_data() {
6956 let cx = test_cx();
6958 let vfs = MemoryVfs::new();
6959 let mut adapter = make_adapter(&vfs, &cx);
6960
6961 let page1 = sample_page(0x01);
6962 let page2 = sample_page(0x02);
6963 let page3 = sample_page(0x03);
6964
6965 adapter.append_frame(&cx, 1, &page1, 0).expect("append");
6966 adapter.append_frame(&cx, 2, &page2, 0).expect("append");
6967 adapter
6968 .append_frame(&cx, 3, &page3, 3)
6969 .expect("append commit");
6970 adapter.sync(&cx).expect("publish staged frames");
6971
6972 assert_eq!(adapter.read_page(&cx, 1).expect("read"), Some(page1));
6974 assert_eq!(adapter.read_page(&cx, 2).expect("read"), Some(page2));
6975 assert_eq!(adapter.read_page(&cx, 3).expect("read"), Some(page3));
6976
6977 assert_eq!(adapter.read_page(&cx, 99).expect("read"), None);
6979 }
6980
6981 #[test]
6982 fn test_page_index_returns_latest_version() {
6983 let cx = test_cx();
6985 let vfs = MemoryVfs::new();
6986 let mut adapter = make_adapter(&vfs, &cx);
6987
6988 let old_data = sample_page(0xAA);
6989 let new_data = sample_page(0xBB);
6990
6991 adapter
6992 .append_frame(&cx, 5, &old_data, 0)
6993 .expect("append old");
6994 adapter
6995 .append_frame(&cx, 5, &new_data, 1)
6996 .expect("append new (commit)");
6997 adapter.sync(&cx).expect("publish staged frames");
6998
6999 assert_eq!(
7000 adapter.read_page(&cx, 5).expect("read"),
7001 Some(new_data),
7002 "page index must return the latest frame for a page"
7003 );
7004 }
7005
7006 #[test]
7007 fn test_page_index_invalidated_on_wal_reset() {
7008 let cx = test_cx();
7011 let vfs = MemoryVfs::new();
7012 let mut adapter = make_adapter(&vfs, &cx);
7013
7014 let old_data = sample_page(0x11);
7015 adapter
7016 .append_frame(&cx, 1, &old_data, 1)
7017 .expect("append commit");
7018 adapter.sync(&cx).expect("publish staged frames");
7019
7020 assert_eq!(adapter.read_page(&cx, 1).expect("read old"), Some(old_data));
7022
7023 let new_salts = WalSalts {
7025 salt1: 0xAAAA_BBBB,
7026 salt2: 0xCCCC_DDDD,
7027 };
7028 adapter
7029 .inner_mut()
7030 .expect("no staged batch blocks inner access")
7031 .reset(&cx, 1, new_salts, false)
7032 .expect("WAL reset");
7033
7034 let new_data = sample_page(0x22);
7036 adapter
7037 .append_frame(&cx, 1, &new_data, 1)
7038 .expect("append new generation commit");
7039 adapter.sync(&cx).expect("publish new generation commit");
7040
7041 let result = adapter.read_page(&cx, 1).expect("read after reset");
7043 assert_eq!(
7044 result,
7045 Some(new_data),
7046 "after WAL reset, page index must return new-generation data, not stale cached data"
7047 );
7048
7049 let old_only = sample_page(0x33);
7051 assert_eq!(
7053 adapter.read_page(&cx, 99).expect("read non-existent"),
7054 None,
7055 "pages from old WAL generation must not appear after reset"
7056 );
7057 drop(old_only);
7059 }
7060
7061 #[test]
7062 fn test_page_index_invalidated_on_same_salt_generation_change() {
7063 init_wal_publication_test_tracing();
7064 let cx = test_cx();
7067 let vfs = MemoryVfs::new();
7068 let mut adapter = make_adapter(&vfs, &cx);
7069
7070 let reused_salts = adapter.inner().header().salts;
7071 let old_data = sample_page(0x11);
7072 adapter
7073 .append_frame(&cx, 1, &old_data, 1)
7074 .expect("append commit");
7075 adapter.sync(&cx).expect("publish staged frames");
7076 assert_eq!(adapter.read_page(&cx, 1).expect("read old"), Some(old_data));
7077
7078 adapter
7079 .inner_mut()
7080 .expect("no staged batch blocks inner access")
7081 .reset(&cx, 1, reused_salts, false)
7082 .expect("reset with same salts");
7083 let new_data = sample_page(0x22);
7084 adapter
7085 .append_frame(&cx, 2, &new_data, 2)
7086 .expect("append new generation commit");
7087 adapter.sync(&cx).expect("publish new generation commit");
7088 let refreshed = adapter
7089 .refresh_published_snapshot(&cx)
7090 .expect("refresh published snapshot after same-salt reset");
7091 assert_eq!(refreshed.generation.checkpoint_seq, 1);
7092 assert_eq!(refreshed.generation.salts, reused_salts);
7093 assert_eq!(refreshed.last_commit_frame, Some(0));
7094 assert_eq!(refreshed.commit_count, 1);
7095 assert_eq!(refreshed.latest_frame_entries, 1);
7096
7097 assert_eq!(
7098 adapter.read_page(&cx, 1).expect("old page should be gone"),
7099 None,
7100 "cached index entries from the previous generation must be invalidated"
7101 );
7102 assert_eq!(
7103 adapter.read_page(&cx, 2).expect("read new page"),
7104 Some(new_data),
7105 "adapter must resolve pages from the new generation even when salts are reused"
7106 );
7107 }
7108
7109 #[test]
7110 fn test_refresh_published_snapshot_materializes_existing_committed_prefix() {
7111 init_wal_publication_test_tracing();
7112 let cx = test_cx();
7113 let vfs = MemoryVfs::new();
7114
7115 let file_writer = open_wal_file(&vfs, &cx);
7116 let wal_writer =
7117 WalFile::create(&cx, file_writer, PAGE_SIZE, 0, test_salts()).expect("create WAL");
7118 let mut writer = WalBackendAdapter::new(wal_writer);
7119
7120 let p1 = sample_page(0x71);
7121 let p2 = sample_page(0x72);
7122 writer.append_frame(&cx, 1, &p1, 0).expect("append p1");
7123 writer
7124 .append_frame(&cx, 2, &p2, 2)
7125 .expect("append p2 commit");
7126 writer.sync(&cx).expect("sync writer");
7127
7128 let file_reader = open_wal_file(&vfs, &cx);
7129 let wal_reader = WalFile::open(&cx, file_reader).expect("open reader WAL");
7130 let mut reader = WalBackendAdapter::new(wal_reader);
7131
7132 let before = reader.published_snapshot();
7133 assert_eq!(before.last_commit_frame, None);
7134 assert_eq!(before.commit_count, 0);
7135 assert_eq!(before.latest_frame_entries, 0);
7136
7137 let refreshed = reader
7138 .refresh_published_snapshot(&cx)
7139 .expect("refresh published snapshot");
7140 assert_eq!(refreshed.last_commit_frame, Some(1));
7141 assert_eq!(refreshed.commit_count, 1);
7142 assert_eq!(refreshed.latest_frame_entries, 2);
7143 assert!(refreshed.lookup_contract_is_authoritative());
7144 assert_eq!(reader.read_page(&cx, 1).expect("read p1"), Some(p1));
7145 assert_eq!(reader.read_page(&cx, 2).expect("read p2"), Some(p2));
7146 }
7147
7148 #[test]
7149 fn test_page_index_incremental_extend_after_durable_sync() {
7150 let cx = test_cx();
7153 let vfs = MemoryVfs::new();
7154 let mut adapter = make_adapter(&vfs, &cx);
7155
7156 let page1 = sample_page(0x10);
7157 adapter
7158 .append_frame(&cx, 1, &page1, 1)
7159 .expect("append commit 1");
7160 adapter.sync(&cx).expect("durably publish commit 1");
7161
7162 assert_eq!(
7164 adapter.read_page(&cx, 1).expect("read"),
7165 Some(page1.clone())
7166 );
7167
7168 let page2 = sample_page(0x20);
7170 let page1_v2 = sample_page(0x30);
7171 adapter
7172 .append_frame(&cx, 2, &page2, 0)
7173 .expect("append page 2");
7174 adapter
7175 .append_frame(&cx, 1, &page1_v2, 3)
7176 .expect("append page 1 v2 (commit)");
7177 adapter.sync(&cx).expect("durably publish commit 2");
7178
7179 assert_eq!(
7181 adapter.read_page(&cx, 1).expect("read page 1 v2"),
7182 Some(page1_v2),
7183 "incremental index extend should pick up the updated page"
7184 );
7185 assert_eq!(adapter.read_page(&cx, 2).expect("read page 2"), Some(page2));
7186 }
7187
7188 fn commit_batch_pages() -> (Vec<u8>, Vec<u8>) {
7190 (sample_page(0x71), sample_page(0x72))
7191 }
7192
7193 fn assert_publication_unchanged(adapter: &WalBackendAdapter<impl VfsFile>, context: &str) {
7195 assert_eq!(
7196 adapter.published_snapshot.last_commit_frame, None,
7197 "{context}: publication must not advance before a successful sync"
7198 );
7199 assert_eq!(
7200 adapter.published_snapshot.commit_count, 0,
7201 "{context}: commit count must not advance before a successful sync"
7202 );
7203 assert!(
7204 adapter.published_snapshot.page_index.is_empty(),
7205 "{context}: no page may be visible before a successful sync"
7206 );
7207 }
7208
7209 #[test]
7210 fn test_append_frame_without_sync_leaves_publication_unchanged() {
7211 let cx = test_cx();
7212 let vfs = MemoryVfs::new();
7213 let mut adapter = make_adapter(&vfs, &cx);
7214
7215 let (p1, p2) = commit_batch_pages();
7216 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
7217 adapter.append_frame(&cx, 2, &p2, 2).expect("append commit");
7218
7219 assert_publication_unchanged(&adapter, "append_frame");
7220 assert_eq!(
7221 adapter.pending_publication_commit,
7222 Some(1),
7223 "append_frame must stage the commit horizon for a later sync"
7224 );
7225 }
7226
7227 #[test]
7228 fn test_append_frames_without_sync_leaves_publication_unchanged() {
7229 let cx = test_cx();
7230 let vfs = MemoryVfs::new();
7231 let mut adapter = make_adapter(&vfs, &cx);
7232
7233 let (p1, p2) = commit_batch_pages();
7234 let frames = [
7235 WalFrameRef {
7236 page_number: 1,
7237 page_data: &p1,
7238 db_size_if_commit: 0,
7239 },
7240 WalFrameRef {
7241 page_number: 2,
7242 page_data: &p2,
7243 db_size_if_commit: 2,
7244 },
7245 ];
7246 adapter
7247 .append_frames(&cx, &frames)
7248 .expect("append frames batch");
7249
7250 assert_publication_unchanged(&adapter, "append_frames");
7251 assert_eq!(
7252 adapter.pending_publication_commit,
7253 Some(1),
7254 "append_frames must stage the commit horizon for a later sync"
7255 );
7256 }
7257
7258 #[test]
7259 fn test_append_frames_tracked_without_sync_leaves_publication_unchanged() {
7260 let cx = test_cx();
7261 let vfs = MemoryVfs::new();
7262 let mut adapter = make_adapter(&vfs, &cx);
7263
7264 let (p1, p2) = commit_batch_pages();
7265 let frames = [
7266 WalFrameRef {
7267 page_number: 1,
7268 page_data: &p1,
7269 db_size_if_commit: 0,
7270 },
7271 WalFrameRef {
7272 page_number: 2,
7273 page_data: &p2,
7274 db_size_if_commit: 2,
7275 },
7276 ];
7277 adapter
7278 .append_frames_tracked(&cx, &frames, VfsWriteCompletion::new())
7279 .expect("append tracked frames batch");
7280
7281 assert_publication_unchanged(&adapter, "append_frames_tracked");
7282 assert_eq!(
7283 adapter.pending_publication_commit,
7284 Some(1),
7285 "append_frames_tracked must stage the commit horizon for a later sync"
7286 );
7287 }
7288
7289 #[test]
7290 fn test_append_prepared_frames_without_sync_leaves_publication_unchanged() {
7291 let cx = test_cx();
7292 let vfs = MemoryVfs::new();
7293 let mut adapter = make_adapter(&vfs, &cx);
7294
7295 let (p1, p2) = commit_batch_pages();
7296 let frames = [
7297 WalFrameRef {
7298 page_number: 1,
7299 page_data: &p1,
7300 db_size_if_commit: 0,
7301 },
7302 WalFrameRef {
7303 page_number: 2,
7304 page_data: &p2,
7305 db_size_if_commit: 2,
7306 },
7307 ];
7308 let mut prepared = adapter
7309 .prepare_append_frames(&frames)
7310 .expect("prepare append")
7311 .expect("prepared batch");
7312 adapter
7313 .append_prepared_frames(&cx, &mut prepared)
7314 .expect("append prepared");
7315
7316 assert_publication_unchanged(&adapter, "append_prepared_frames");
7317 assert_eq!(
7318 adapter.pending_publication_commit,
7319 Some(1),
7320 "append_prepared_frames must stage the commit horizon for a later sync"
7321 );
7322 }
7323
7324 #[test]
7325 fn test_append_prepared_frames_tracked_without_sync_leaves_publication_unchanged() {
7326 let cx = test_cx();
7327 let vfs = MemoryVfs::new();
7328 let mut adapter = make_adapter(&vfs, &cx);
7329
7330 let (p1, p2) = commit_batch_pages();
7331 let frames = [
7332 WalFrameRef {
7333 page_number: 1,
7334 page_data: &p1,
7335 db_size_if_commit: 0,
7336 },
7337 WalFrameRef {
7338 page_number: 2,
7339 page_data: &p2,
7340 db_size_if_commit: 2,
7341 },
7342 ];
7343 let mut prepared = adapter
7344 .prepare_append_frames(&frames)
7345 .expect("prepare append")
7346 .expect("prepared batch");
7347 adapter
7348 .append_prepared_frames_tracked(&cx, &mut prepared, VfsWriteCompletion::new())
7349 .expect("append prepared tracked");
7350
7351 assert_publication_unchanged(&adapter, "append_prepared_frames_tracked");
7352 assert_eq!(
7353 adapter.pending_publication_commit,
7354 Some(1),
7355 "append_prepared_frames_tracked must stage the commit horizon for a later sync"
7356 );
7357 }
7358
7359 #[test]
7360 fn test_successful_sync_publishes_staged_commit_horizon() {
7361 let cx = test_cx();
7362 let vfs = MemoryVfs::new();
7363 let mut adapter = make_adapter(&vfs, &cx);
7364
7365 let (p1, p2) = commit_batch_pages();
7366 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
7367 adapter.append_frame(&cx, 2, &p2, 2).expect("append commit");
7368 assert_publication_unchanged(&adapter, "before sync");
7369
7370 adapter.sync(&cx).expect("sync must succeed");
7371
7372 assert_eq!(
7373 adapter.published_snapshot.last_commit_frame,
7374 Some(1),
7375 "a successful sync must publish the staged commit horizon"
7376 );
7377 assert_eq!(
7378 adapter.published_snapshot.commit_count, 1,
7379 "a successful sync must publish the staged commit count"
7380 );
7381 assert_eq!(
7382 adapter.published_snapshot.page_index.len(),
7383 2,
7384 "a successful sync must publish every staged page"
7385 );
7386 assert_eq!(
7387 adapter.pending_publication_commit, None,
7388 "a published batch must no longer be staged"
7389 );
7390 assert!(
7391 adapter.pending_publication_frames.is_empty(),
7392 "a published batch must drain its staged frames"
7393 );
7394 }
7395
7396 #[test]
7397 fn test_failed_sync_advances_no_publication_and_retry_publishes() {
7398 let cx = test_cx();
7399 let vfs = CheckpointHandoffFaultVfs::new();
7400 let mut adapter = make_fault_adapter(&vfs, &cx);
7401
7402 let (p1, p2) = commit_batch_pages();
7403 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
7404 adapter.append_frame(&cx, 2, &p2, 2).expect("append commit");
7405
7406 vfs.fail_next_wal_sync();
7407 let failure = adapter
7408 .sync(&cx)
7409 .expect_err("injected WAL sync failure must surface");
7410 assert!(
7411 failure.to_string().contains("injected WAL sync failure"),
7412 "sync must report the injected durability failure, got: {failure}"
7413 );
7414
7415 assert_publication_unchanged(&adapter, "after failed sync");
7416 assert_eq!(
7417 adapter.pending_publication_commit,
7418 Some(1),
7419 "a failed sync must preserve the staged horizon for retry"
7420 );
7421 assert!(
7422 !adapter.pending_publication_frames.is_empty(),
7423 "a failed sync must preserve staged frames for retry"
7424 );
7425
7426 adapter.sync(&cx).expect("retry sync must succeed");
7428
7429 assert_eq!(
7430 adapter.published_snapshot.last_commit_frame,
7431 Some(1),
7432 "retrying sync must publish the preserved commit horizon"
7433 );
7434 assert_eq!(
7435 adapter.published_snapshot.commit_count, 1,
7436 "retrying sync must publish the preserved commit count"
7437 );
7438 assert_eq!(
7439 adapter.published_snapshot.page_index.len(),
7440 2,
7441 "retrying sync must publish every preserved page"
7442 );
7443 assert_eq!(
7444 adapter.pending_publication_commit, None,
7445 "a retried publication must clear the staged horizon"
7446 );
7447 }
7448
7449 #[test]
7450 fn test_failed_sync_then_append_cannot_drop_or_publish_pending() {
7451 let cx = test_cx();
7452 let vfs = CheckpointHandoffFaultVfs::new();
7453 let mut adapter = make_fault_adapter(&vfs, &cx);
7454
7455 let (p1, p2) = commit_batch_pages();
7456 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
7457 adapter.append_frame(&cx, 2, &p2, 2).expect("append commit");
7458
7459 vfs.fail_next_wal_sync();
7460 adapter
7461 .sync(&cx)
7462 .expect_err("injected WAL sync failure must surface");
7463
7464 let staged_after_failure = adapter.pending_publication_commit;
7465 let staged_frames_after_failure = adapter.pending_publication_frames.len();
7466 assert_eq!(
7467 staged_after_failure,
7468 Some(1),
7469 "failed sync must preserve the staged horizon"
7470 );
7471
7472 let p3 = sample_page(0x73);
7475 adapter
7476 .append_frame(&cx, 3, &p3, 3)
7477 .expect("append after failed sync");
7478
7479 assert_publication_unchanged(&adapter, "append after failed sync");
7480 assert!(
7481 adapter.pending_publication_frames.len() > staged_frames_after_failure,
7482 "append after a failed sync must extend, never discard, the staged batch"
7483 );
7484 assert_eq!(
7485 adapter.pending_publication_commit,
7486 Some(2),
7487 "append after a failed sync must carry the staged horizon forward"
7488 );
7489
7490 adapter.sync(&cx).expect("sync after failed attempt");
7492 assert_eq!(
7493 adapter.published_snapshot.last_commit_frame,
7494 Some(2),
7495 "recovered sync must publish the full preserved horizon"
7496 );
7497 assert_eq!(
7498 adapter.pending_publication_commit, None,
7499 "recovered sync must clear the staged horizon"
7500 );
7501 }
7502
7503 #[test]
7504 fn test_failed_sync_then_begin_transaction_then_append_fails_closed() {
7505 let cx = test_cx();
7506 let vfs = CheckpointHandoffFaultVfs::new();
7507 let mut adapter = make_fault_adapter(&vfs, &cx);
7508
7509 let (p1, p2) = commit_batch_pages();
7510 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
7511 adapter.append_frame(&cx, 2, &p2, 2).expect("append commit");
7512
7513 vfs.fail_next_wal_sync();
7514 adapter
7515 .sync(&cx)
7516 .expect_err("injected WAL sync failure must surface");
7517 assert_eq!(
7518 adapter.pending_publication_commit,
7519 Some(1),
7520 "failed sync must preserve the staged horizon"
7521 );
7522
7523 let begin_error = adapter
7527 .begin_transaction(&cx)
7528 .expect_err("begin_transaction must fail closed while frames are staged");
7529 assert!(
7530 matches!(begin_error, FrankenError::Busy),
7531 "staged-state rejection must be retryable Busy, not corruption: {begin_error:?}"
7532 );
7533 assert_publication_unchanged(&adapter, "begin_transaction refused after failed sync");
7534 assert_eq!(
7535 adapter.pending_publication_commit,
7536 Some(1),
7537 "a refused begin_transaction must not drop the staged horizon"
7538 );
7539 assert!(
7540 adapter.pinned_read_snapshot().is_none(),
7541 "a refused begin_transaction must not pin a read snapshot"
7542 );
7543
7544 adapter.refresh_before_append = true;
7547 let p3 = sample_page(0x74);
7548 let append_error = adapter
7549 .append_frame(&cx, 3, &p3, 3)
7550 .expect_err("append must fail closed while frames are staged");
7551 assert!(
7552 matches!(append_error, FrankenError::Busy),
7553 "append rejection must be retryable Busy: {append_error:?}"
7554 );
7555 assert_publication_unchanged(&adapter, "append refused after failed sync");
7556 assert_eq!(
7557 adapter.pending_publication_commit,
7558 Some(1),
7559 "a refused append must leave the staged horizon intact"
7560 );
7561 assert!(
7562 !adapter.pending_publication_frames.is_empty(),
7563 "a refused append must leave the staged frames intact"
7564 );
7565 adapter.refresh_before_append = false;
7566
7567 adapter.sync(&cx).expect("sync after failed attempt");
7569 assert_eq!(
7570 adapter.published_snapshot.last_commit_frame,
7571 Some(1),
7572 "recovered sync must publish the preserved horizon"
7573 );
7574 }
7575
7576 #[test]
7577 fn test_failed_sync_then_checkpoint_fails_closed_and_preserves_state() {
7578 let cx = test_cx();
7579 let vfs = CheckpointHandoffFaultVfs::new();
7580 let mut adapter = make_fault_adapter(&vfs, &cx);
7581
7582 let (p1, p2) = commit_batch_pages();
7583 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
7584 adapter.append_frame(&cx, 2, &p2, 2).expect("append commit");
7585
7586 vfs.fail_next_wal_sync();
7587 adapter
7588 .sync(&cx)
7589 .expect_err("injected WAL sync failure must surface");
7590
7591 let frames_before = adapter.frame_count();
7592 let staged_before = adapter.pending_publication_commit;
7593 let staged_frame_count_before = adapter.pending_publication_frames.len();
7594
7595 let mut writer = MockCheckpointPageWriter;
7599 let checkpoint_error = adapter
7600 .checkpoint(&cx, CheckpointMode::Passive, &mut writer, 0, None)
7601 .expect_err("checkpoint must fail closed while frames are staged");
7602 assert!(
7603 matches!(checkpoint_error, FrankenError::CheckpointFailed { .. }),
7604 "checkpoint rejection must be CheckpointFailed, not corruption: {checkpoint_error:?}"
7605 );
7606
7607 assert_eq!(
7608 adapter.frame_count(),
7609 frames_before,
7610 "a refused checkpoint must not mutate WAL bytes"
7611 );
7612 assert_publication_unchanged(&adapter, "checkpoint refused");
7613 assert_eq!(
7614 adapter.pending_publication_commit, staged_before,
7615 "a refused checkpoint must preserve the staged horizon"
7616 );
7617 assert_eq!(
7618 adapter.pending_publication_frames.len(),
7619 staged_frame_count_before,
7620 "a refused checkpoint must preserve the staged frames"
7621 );
7622
7623 adapter.sync(&cx).expect("retry sync must succeed");
7625 assert_eq!(
7626 adapter.published_snapshot.last_commit_frame,
7627 Some(1),
7628 "retry sync must publish the preserved horizon"
7629 );
7630 assert_eq!(
7631 adapter.pending_publication_commit, None,
7632 "a published batch must no longer be staged"
7633 );
7634 }
7635
7636 #[test]
7637 fn test_midtransaction_sync_preserves_uncommitted_frames_and_allows_continuation() {
7638 let cx = test_cx();
7639 let vfs = MemoryVfs::new();
7640 let mut adapter = make_adapter(&vfs, &cx);
7641
7642 let (p1, p2) = commit_batch_pages();
7643
7644 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
7647 assert_eq!(
7648 adapter.pending_publication_commit, None,
7649 "a non-commit append stages no commit horizon"
7650 );
7651 adapter
7652 .sync(&cx)
7653 .expect("mid-transaction sync must succeed");
7654
7655 assert_publication_unchanged(&adapter, "sync of uncommitted frames");
7656 assert!(
7657 !adapter.pending_publication_frames.is_empty(),
7658 "a mid-transaction sync must preserve durable-but-uncommitted frames"
7659 );
7660
7661 adapter
7663 .append_frame(&cx, 2, &p2, 2)
7664 .expect("commit append after mid-transaction sync must be allowed");
7665 assert_eq!(
7666 adapter.pending_publication_commit,
7667 Some(1),
7668 "the commit append must stage the horizon for the whole batch"
7669 );
7670 assert_publication_unchanged(&adapter, "commit staged but not yet synced");
7671
7672 adapter.sync(&cx).expect("commit sync must succeed");
7673
7674 assert_eq!(
7675 adapter.published_snapshot.last_commit_frame,
7676 Some(1),
7677 "the commit sync must publish the whole batch"
7678 );
7679 assert_eq!(
7680 adapter.published_snapshot.commit_count, 1,
7681 "the batch must publish exactly one commit"
7682 );
7683 assert_eq!(
7684 adapter.published_snapshot.page_index.len(),
7685 2,
7686 "both pages must be published exactly once"
7687 );
7688 assert_eq!(
7689 adapter.published_snapshot.page_index.get(&1),
7690 Some(&0),
7691 "page 1 must map to its frame from before the mid-transaction sync"
7692 );
7693 assert_eq!(
7694 adapter.published_snapshot.page_index.get(&2),
7695 Some(&1),
7696 "page 2 must map to the commit frame"
7697 );
7698 assert!(
7699 !adapter.has_pending_publication(),
7700 "a published batch must leave nothing staged"
7701 );
7702 }
7703
7704 #[test]
7705 fn test_inner_mut_fails_closed_while_batch_is_staged() {
7706 let cx = test_cx();
7707 let vfs = CheckpointHandoffFaultVfs::new();
7708 let mut adapter = make_fault_adapter(&vfs, &cx);
7709
7710 let (p1, p2) = commit_batch_pages();
7711 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
7712 adapter.append_frame(&cx, 2, &p2, 2).expect("append commit");
7713
7714 assert!(
7715 adapter.has_pending_publication(),
7716 "an appended-but-unsynced batch must report as pending"
7717 );
7718 assert!(
7721 matches!(adapter.inner_mut(), Err(FrankenError::Busy)),
7722 "inner_mut must fail closed with retryable Busy while frames are staged"
7723 );
7724 assert_eq!(
7725 adapter.pending_publication_commit,
7726 Some(1),
7727 "a refused inner_mut must preserve the staged horizon"
7728 );
7729
7730 adapter.sync(&cx).expect("sync staged batch");
7732 assert!(
7733 !adapter.has_pending_publication(),
7734 "a published batch must clear the pending flag"
7735 );
7736 adapter
7737 .inner_mut()
7738 .expect("inner_mut must succeed once the batch is drained");
7739 }
7740
7741 #[test]
7742 fn test_unpinned_refresh_does_not_expose_staged_horizon_before_sync() {
7743 let cx = test_cx();
7744 let vfs = MemoryVfs::new();
7745 let mut adapter = make_adapter(&vfs, &cx);
7746
7747 let (p1, p2) = commit_batch_pages();
7748 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
7749 adapter.append_frame(&cx, 2, &p2, 2).expect("append commit");
7750
7751 adapter
7754 .refresh_published_snapshot(&cx)
7755 .expect("refresh published snapshot");
7756 assert_publication_unchanged(&adapter, "refresh with staged frames");
7757 assert_eq!(
7758 adapter.pending_publication_commit,
7759 Some(1),
7760 "refresh must leave the staged horizon intact"
7761 );
7762
7763 adapter.sync(&cx).expect("sync staged batch");
7764 assert_eq!(
7765 adapter.published_snapshot.last_commit_frame,
7766 Some(1),
7767 "sync must publish once the staged batch is durable"
7768 );
7769 }
7770
7771 #[test]
7772 fn test_authorized_deferred_commit_publishes_without_claiming_fsync() {
7773 let cx = test_cx();
7774 let vfs = MemoryVfs::new();
7775 let mut adapter = make_adapter(&vfs, &cx);
7776
7777 let (p1, p2) = commit_batch_pages();
7778 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
7779 adapter.append_frame(&cx, 2, &p2, 2).expect("append commit");
7780 let fsynced_before = adapter.wal.last_fsynced_frame_count();
7781
7782 adapter
7783 .publish_authorized_deferred_commit(&cx)
7784 .expect("parallel-WAL authorization must publish the deferred commit");
7785
7786 assert_eq!(
7787 adapter.published_snapshot.last_commit_frame,
7788 Some(1),
7789 "the authorized commit marker must become visible"
7790 );
7791 assert_eq!(
7792 adapter.published_snapshot.commit_count, 1,
7793 "the authorized batch must publish exactly one commit"
7794 );
7795 assert!(
7796 !adapter.has_pending_publication(),
7797 "authorization must drain the staged publication horizon"
7798 );
7799 assert_eq!(
7800 adapter.wal.last_fsynced_frame_count(),
7801 fsynced_before,
7802 "deferred authorization must not claim or force an fsync"
7803 );
7804 adapter
7805 .begin_transaction(&cx)
7806 .expect("the next transaction must not see a stale Busy");
7807 }
7808
7809 #[test]
7810 fn test_commit_append_publishes_visibility_snapshot() {
7811 init_wal_publication_test_tracing();
7812 let cx = test_cx();
7813 let vfs = MemoryVfs::new();
7814 let mut adapter = make_adapter(&vfs, &cx);
7815
7816 let p1 = sample_page(0x41);
7817 let p2 = sample_page(0x42);
7818 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
7819 adapter.append_frame(&cx, 2, &p2, 2).expect("append commit");
7820 adapter.sync(&cx).expect("sync commit batch");
7823
7824 assert_eq!(
7825 adapter.published_snapshot.last_commit_frame,
7826 Some(1),
7827 "synced commit should publish the visible commit horizon"
7828 );
7829 assert_eq!(
7830 adapter.published_snapshot.commit_count, 1,
7831 "synced commit should track the visible WAL commit count"
7832 );
7833 assert_eq!(
7834 adapter.published_snapshot.page_index.len(),
7835 2,
7836 "published snapshot should track both committed pages"
7837 );
7838 assert_eq!(
7839 adapter.published_snapshot.page_index.get(&2),
7840 Some(&1),
7841 "published snapshot must map each page to its latest committed frame"
7842 );
7843 }
7844
7845 #[test]
7846 fn test_prepared_append_publishes_visibility_snapshot() {
7847 init_wal_publication_test_tracing();
7848 let cx = test_cx();
7849 let vfs = MemoryVfs::new();
7850 let mut adapter = make_adapter(&vfs, &cx);
7851
7852 let p1 = sample_page(0x51);
7853 let p2 = sample_page(0x52);
7854 let frames = [
7855 WalFrameRef {
7856 page_number: 1,
7857 page_data: &p1,
7858 db_size_if_commit: 0,
7859 },
7860 WalFrameRef {
7861 page_number: 2,
7862 page_data: &p2,
7863 db_size_if_commit: 2,
7864 },
7865 ];
7866 let mut prepared = adapter
7867 .prepare_append_frames(&frames)
7868 .expect("prepare append")
7869 .expect("prepared batch");
7870 adapter
7871 .append_prepared_frames(&cx, &mut prepared)
7872 .expect("append prepared");
7873 adapter.sync(&cx).expect("sync prepared commit batch");
7875
7876 assert_eq!(
7877 adapter.published_snapshot.last_commit_frame,
7878 Some(1),
7879 "synced prepared commit should publish the visible commit horizon"
7880 );
7881 assert_eq!(
7882 adapter.published_snapshot.commit_count, 1,
7883 "synced prepared commit should track the visible WAL commit count"
7884 );
7885 assert_eq!(
7886 adapter.published_snapshot.page_index.len(),
7887 2,
7888 "synced prepared commit should publish all committed pages"
7889 );
7890 assert_eq!(
7891 adapter.published_snapshot.page_index.get(&2),
7892 Some(&1),
7893 "prepared commit append must map each page to its latest committed frame"
7894 );
7895 }
7896
7897 #[test]
7898 fn test_commit_publication_refreshes_external_prefix_before_local_commit() {
7899 let cx = test_cx();
7900 let vfs = MemoryVfs::new();
7901
7902 let file_writer = open_wal_file(&vfs, &cx);
7903 let wal_writer =
7904 WalFile::create(&cx, file_writer, PAGE_SIZE, 0, test_salts()).expect("create WAL");
7905 let mut writer = WalBackendAdapter::new(wal_writer);
7906
7907 let file_follower = open_wal_file(&vfs, &cx);
7908 let wal_follower = WalFile::open(&cx, file_follower).expect("open WAL");
7909 let mut follower = WalBackendAdapter::new(wal_follower);
7910
7911 let p1 = sample_page(0x61);
7912 writer
7913 .append_frame(&cx, 1, &p1, 1)
7914 .expect("writer commit 1");
7915 writer.sync(&cx).expect("sync writer commit 1");
7916
7917 let p2 = sample_page(0x62);
7918 writer
7919 .append_frame(&cx, 2, &p2, 2)
7920 .expect("writer commit 2");
7921 writer.sync(&cx).expect("sync writer commit 2");
7922
7923 let p3 = sample_page(0x63);
7924 follower
7925 .append_frame(&cx, 3, &p3, 3)
7926 .expect("follower local commit");
7927
7928 assert_eq!(
7932 follower.published_snapshot.last_commit_frame,
7933 Some(1),
7934 "refresh-before-append must publish the external prefix only"
7935 );
7936 assert_eq!(
7937 follower.published_snapshot.commit_count, 2,
7938 "the staged local commit must not count until publication"
7939 );
7940 assert_eq!(
7941 follower.published_snapshot.page_index.get(&1),
7942 Some(&0),
7943 "refresh-before-append should preserve earlier committed pages"
7944 );
7945 assert_eq!(
7946 follower.published_snapshot.page_index.get(&2),
7947 Some(&1),
7948 "refresh-before-append should publish externally committed pages"
7949 );
7950 assert_eq!(
7951 follower.published_snapshot.page_index.get(&3),
7952 None,
7953 "the staged local page must stay out of the published map"
7954 );
7955
7956 follower.sync(&cx).expect("publish follower local commit");
7957 assert_eq!(
7958 follower.published_snapshot.last_commit_frame,
7959 Some(2),
7960 "publication must extend the map with the local commit"
7961 );
7962 assert_eq!(follower.published_snapshot.commit_count, 3);
7963 assert_eq!(
7964 follower.published_snapshot.page_index.get(&3),
7965 Some(&2),
7966 "published local commit extends the WAL visibility map"
7967 );
7968 assert_eq!(follower.read_page(&cx, 1).expect("read p1"), Some(p1));
7969 assert_eq!(follower.read_page(&cx, 2).expect("read p2"), Some(p2));
7970 assert_eq!(follower.read_page(&cx, 3).expect("read p3"), Some(p3));
7971 }
7972
7973 #[test]
7974 fn test_truncate_checkpoint_republishes_empty_generation_snapshot() {
7975 init_wal_publication_test_tracing();
7976 let cx = test_cx();
7977 let vfs = MemoryVfs::new();
7978 let mut adapter = make_adapter(&vfs, &cx);
7979 let mut writer = MockCheckpointPageWriter;
7980
7981 adapter
7982 .append_frame(&cx, 1, &sample_page(0x61), 1)
7983 .expect("append committed frame");
7984 adapter.sync(&cx).expect("sync committed frame");
7988 let before = adapter.published_snapshot();
7989 assert_eq!(before.last_commit_frame, Some(0));
7990 assert_eq!(before.commit_count, 1);
7991 assert_eq!(before.latest_frame_entries, 1);
7992
7993 let result = adapter
7994 .checkpoint(&cx, CheckpointMode::Truncate, &mut writer, 0, None)
7995 .expect("truncate checkpoint");
7996 assert!(result.completed);
7997 assert!(result.wal_was_reset);
7998
7999 let after = adapter.published_snapshot();
8000 assert_ne!(
8001 before.generation, after.generation,
8002 "truncate checkpoint should publish a new WAL generation"
8003 );
8004 assert_eq!(after.last_commit_frame, None);
8005 assert_eq!(after.commit_count, 0);
8006 assert_eq!(after.latest_frame_entries, 0);
8007 assert!(after.lookup_contract_is_authoritative());
8008 }
8009
8010 #[test]
8013 fn test_partial_index_falls_back_to_linear_scan() {
8014 init_wal_publication_test_tracing();
8015 let cx = test_cx();
8018 let vfs = MemoryVfs::new();
8019 let mut adapter = make_adapter(&vfs, &cx);
8020
8021 adapter.set_page_index_cap(2);
8024
8025 let p1 = sample_page(0x01);
8028 let p2 = sample_page(0x02);
8029 let p3 = sample_page(0x03);
8030 let p4 = sample_page(0x04);
8031 let p5 = sample_page(0x05);
8032
8033 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
8034 adapter.append_frame(&cx, 2, &p2, 0).expect("append p2");
8035 adapter.append_frame(&cx, 3, &p3, 0).expect("append p3");
8036 adapter.append_frame(&cx, 4, &p4, 0).expect("append p4");
8037 adapter
8038 .append_frame(&cx, 5, &p5, 5)
8039 .expect("append p5 (commit)");
8040 adapter.sync(&cx).expect("publish staged frames");
8041
8042 assert_eq!(
8044 adapter.read_page(&cx, 1).expect("read p1"),
8045 Some(p1),
8046 "indexed page should be found via HashMap"
8047 );
8048 assert_eq!(
8049 adapter.read_page(&cx, 2).expect("read p2"),
8050 Some(p2),
8051 "indexed page should be found via HashMap"
8052 );
8053
8054 assert_eq!(
8057 adapter.read_page(&cx, 3).expect("read p3"),
8058 Some(p3),
8059 "non-indexed page must be found via linear scan fallback"
8060 );
8061 assert_eq!(
8062 adapter.read_page(&cx, 4).expect("read p4"),
8063 Some(p4),
8064 "non-indexed page must be found via linear scan fallback"
8065 );
8066 assert_eq!(
8067 adapter.read_page(&cx, 5).expect("read p5"),
8068 Some(p5),
8069 "non-indexed page must be found via linear scan fallback"
8070 );
8071
8072 assert_eq!(
8074 adapter.read_page(&cx, 99).expect("read non-existent"),
8075 None,
8076 "non-existent page must return None even with partial index"
8077 );
8078
8079 assert!(
8081 adapter.published_snapshot.index_is_partial,
8082 "index_is_partial should be true when cap is exceeded"
8083 );
8084 }
8085
8086 #[test]
8087 fn test_partial_index_returns_latest_version_via_fallback() {
8088 let cx = test_cx();
8092 let vfs = MemoryVfs::new();
8093 let mut adapter = make_adapter(&vfs, &cx);
8094
8095 adapter.set_page_index_cap(1);
8097
8098 let old_p2 = sample_page(0xAA);
8099 let new_p2 = sample_page(0xBB);
8100
8101 adapter
8103 .append_frame(&cx, 1, &sample_page(0x01), 0)
8104 .expect("append p1");
8105 adapter
8107 .append_frame(&cx, 2, &old_p2, 0)
8108 .expect("append p2 old");
8109 adapter
8112 .append_frame(&cx, 2, &new_p2, 3)
8113 .expect("append p2 new (commit)");
8114 adapter.sync(&cx).expect("publish staged frames");
8115
8116 assert_eq!(
8118 adapter.read_page(&cx, 2).expect("read p2"),
8119 Some(new_p2),
8120 "backwards scan must return the most recent frame for the page"
8121 );
8122 }
8123
8124 #[test]
8125 fn test_lookup_contract_distinguishes_authoritative_and_fallback_paths() {
8126 init_wal_publication_test_tracing();
8127 let cx = test_cx();
8128 let vfs = MemoryVfs::new();
8129 let mut adapter = make_adapter(&vfs, &cx);
8130 adapter.set_page_index_cap(1);
8131
8132 let p1 = sample_page(0x01);
8133 let p2 = sample_page(0x02);
8134 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
8135 adapter
8136 .append_frame(&cx, 2, &p2, 2)
8137 .expect("append p2 commit");
8138 adapter.sync(&cx).expect("publish staged frames");
8139
8140 let last_commit = adapter
8141 .inner_mut()
8142 .expect("no staged batch blocks inner access")
8143 .last_commit_frame(&cx)
8144 .expect("last commit")
8145 .expect("commit exists");
8146 adapter
8147 .publish_visible_snapshot(&cx, Some(last_commit), "lookup_contract_test")
8148 .expect("build published snapshot");
8149 let snapshot = adapter.published_snapshot.clone();
8150
8151 assert_eq!(
8152 adapter
8153 .resolve_visible_frame(&cx, &snapshot, 1)
8154 .expect("resolve indexed page"),
8155 WalPageLookupResolution::AuthoritativeHit { frame_index: 0 }
8156 );
8157 assert_eq!(
8158 adapter
8159 .resolve_visible_frame(&cx, &snapshot, 2)
8160 .expect("resolve fallback page"),
8161 WalPageLookupResolution::PartialIndexFallbackHit { frame_index: 1 }
8162 );
8163 assert_eq!(
8164 adapter
8165 .resolve_visible_frame(&cx, &snapshot, 99)
8166 .expect("resolve missing page"),
8167 WalPageLookupResolution::PartialIndexFallbackMiss
8168 );
8169 }
8170
8171 #[test]
8172 fn test_lookup_contract_is_authoritative_by_default() {
8173 let cx = test_cx();
8174 let vfs = MemoryVfs::new();
8175 let mut adapter = make_adapter(&vfs, &cx);
8176
8177 let p1 = sample_page(0x11);
8178 let p2 = sample_page(0x22);
8179 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
8180 adapter
8181 .append_frame(&cx, 2, &p2, 2)
8182 .expect("append p2 commit");
8183 adapter.sync(&cx).expect("publish staged frames");
8184
8185 let last_commit = adapter
8186 .inner_mut()
8187 .expect("no staged batch blocks inner access")
8188 .last_commit_frame(&cx)
8189 .expect("last commit")
8190 .expect("commit exists");
8191 adapter
8192 .publish_visible_snapshot(&cx, Some(last_commit), "lookup_contract_default")
8193 .expect("build published snapshot");
8194 let snapshot = adapter.published_snapshot.clone();
8195
8196 assert!(
8197 !snapshot.index_is_partial,
8198 "default index should be authoritative"
8199 );
8200 assert_eq!(
8201 adapter
8202 .resolve_visible_frame(&cx, &snapshot, 1)
8203 .expect("resolve page 1"),
8204 WalPageLookupResolution::AuthoritativeHit { frame_index: 0 }
8205 );
8206 assert_eq!(
8207 adapter
8208 .resolve_visible_frame(&cx, &snapshot, 2)
8209 .expect("resolve page 2"),
8210 WalPageLookupResolution::AuthoritativeHit { frame_index: 1 }
8211 );
8212 assert_eq!(
8213 adapter
8214 .resolve_visible_frame(&cx, &snapshot, 99)
8215 .expect("resolve missing page"),
8216 WalPageLookupResolution::AuthoritativeMiss
8217 );
8218 }
8219
8220 #[test]
8221 fn test_committed_txns_since_page_uses_visible_frame_horizon() {
8222 let cx = test_cx();
8223 let vfs = MemoryVfs::new();
8224 let mut adapter = make_adapter(&vfs, &cx);
8225
8226 let p1 = sample_page(0x31);
8227 let p2 = sample_page(0x32);
8228 let p3 = sample_page(0x33);
8229
8230 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
8231 adapter.append_frame(&cx, 2, &p2, 2).expect("commit tx1");
8232 adapter.append_frame(&cx, 3, &p3, 0).expect("append p3");
8233 adapter.append_frame(&cx, 2, &p2, 3).expect("commit tx2");
8234 adapter.sync(&cx).expect("publish staged commits");
8237
8238 assert_eq!(
8239 adapter
8240 .committed_txns_since_page(&cx, 1)
8241 .expect("count txns since page 1"),
8242 1
8243 );
8244 assert_eq!(
8245 adapter
8246 .committed_txns_since_page(&cx, 2)
8247 .expect("count txns since page 2"),
8248 0
8249 );
8250 assert_eq!(
8251 adapter
8252 .committed_txns_since_page(&cx, 99)
8253 .expect("count txns since missing page"),
8254 2
8255 );
8256 assert_eq!(
8257 adapter
8258 .committed_txn_count(&cx)
8259 .expect("count visible transactions"),
8260 2
8261 );
8262 }
8263
8264 #[test]
8265 fn test_conflicting_pages_since_snapshot_detects_later_wal_writes() {
8266 let cx = test_cx();
8267 let vfs = MemoryVfs::new();
8268 let mut adapter = make_adapter(&vfs, &cx);
8269
8270 let p1 = sample_page(0x41);
8271 let p2_before = sample_page(0x42);
8272 let p2_after = sample_page(0x43);
8273 let p3 = sample_page(0x44);
8274
8275 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
8276 adapter
8277 .append_frame(&cx, 2, &p2_before, 2)
8278 .expect("commit tx1");
8279 adapter.sync(&cx).expect("publish staged frames");
8280 adapter
8281 .begin_transaction(&cx)
8282 .expect("pin transaction snapshot");
8283 let pinned = adapter
8284 .pinned_read_snapshot()
8285 .expect("transaction should expose pinned WAL snapshot");
8286 let conflict_snapshot = TransactionConflictSnapshot {
8287 generation: pinned.generation,
8288 last_commit_frame: pinned.last_commit_frame,
8289 commit_count: pinned.commit_count,
8290 snapshot_db_size: 0,
8291 };
8292
8293 adapter
8294 .append_frame(&cx, 3, &p3, 0)
8295 .expect("append unrelated later page");
8296 adapter
8297 .append_frame(&cx, 2, &p2_after, 3)
8298 .expect("commit later page 2 update");
8299 adapter.sync(&cx).expect("publish later commit");
8302
8303 let conflicts = adapter
8304 .conflicting_pages_since_snapshot(&cx, conflict_snapshot, &[2, 99], &[])
8305 .expect("conflict check should scan later committed frames");
8306 assert_eq!(conflicts, vec![2]);
8307
8308 let unrelated = adapter
8309 .conflicting_pages_since_snapshot(&cx, conflict_snapshot, &[99], &[])
8310 .expect("unrelated page should stay conflict-free");
8311 assert!(unrelated.is_empty());
8312 }
8313
8314 #[test]
8317 fn test_checkpoint_adapter_write_page() {
8318 let cx = test_cx();
8319 let mut writer = MockCheckpointPageWriter;
8320 let mut adapter = CheckpointTargetAdapterRef {
8321 writer: &mut writer,
8322 };
8323
8324 let page_no = PageNumber::new(1).expect("valid page number");
8325 adapter
8326 .write_page(&cx, page_no, &[0u8; 4096])
8327 .expect("write_page");
8328 }
8329
8330 #[test]
8331 fn test_checkpoint_adapter_truncate_db() {
8332 let cx = test_cx();
8333 let mut writer = MockCheckpointPageWriter;
8334 let mut adapter = CheckpointTargetAdapterRef {
8335 writer: &mut writer,
8336 };
8337
8338 adapter.truncate_db(&cx, 10).expect("truncate_db");
8339 }
8340
8341 #[test]
8342 fn test_checkpoint_adapter_sync_db() {
8343 let cx = test_cx();
8344 let mut writer = MockCheckpointPageWriter;
8345 let mut adapter = CheckpointTargetAdapterRef {
8346 writer: &mut writer,
8347 };
8348
8349 adapter.sync_db(&cx).expect("sync_db");
8350 }
8351
8352 #[test]
8353 fn test_checkpoint_adapter_as_dyn_target() {
8354 let cx = test_cx();
8355 let mut writer = MockCheckpointPageWriter;
8356 let mut adapter = CheckpointTargetAdapterRef {
8357 writer: &mut writer,
8358 };
8359
8360 let target: &mut dyn CheckpointTarget = &mut adapter;
8362 let page_no = PageNumber::new(3).expect("valid page number");
8363 target
8364 .write_page(&cx, page_no, &[0u8; 4096])
8365 .expect("write via dyn");
8366 target.truncate_db(&cx, 5).expect("truncate via dyn");
8367 target.sync_db(&cx).expect("sync via dyn");
8368 }
8369}