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,
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::{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() {
768 let durable_frames = self.wal.last_fsynced_frame_count();
769 last_commit_frame.filter(|frame| {
770 frame
771 .checked_add(1)
772 .is_some_and(|frame_count| frame_count <= durable_frames)
773 })
774 } else {
775 last_commit_frame
776 };
777 self.publish_visible_snapshot(cx, last_commit_frame, scenario_id)
778 .await
779 }
780
781 async fn synchronize_publication_before_append(
782 &mut self,
783 cx: &Cx,
784 scenario_id: &'static str,
785 ) -> Result<()> {
786 if self.has_pending_publication() {
794 return Err(FrankenError::Busy);
795 }
796 self.wal.refresh(cx).await?;
797 self.discard_pending_publication();
798 self.publish_latest_committed_snapshot(cx, scenario_id)
799 .await
800 }
801
802 fn discard_pending_publication(&mut self) {
804 self.pending_publication_frames.clear();
805 self.pending_publication_commit = None;
806 self.pending_publication_generation = None;
807 }
808
809 fn stage_pending_commit_publication(&mut self, last_commit_frame: usize) -> Result<()> {
816 let generation = self.wal.generation_identity();
817 if self
821 .pending_publication_generation
822 .is_some_and(|staged| staged != generation)
823 {
824 return Err(FrankenError::WalCorrupt {
825 detail: "cannot stage a commit horizon across differing WAL generations".to_owned(),
826 });
827 }
828 let staged = self
829 .pending_publication_commit
830 .map_or(last_commit_frame, |staged| staged.max(last_commit_frame));
831 self.pending_publication_commit = Some(staged);
832 self.pending_publication_generation = Some(generation);
833 Ok(())
834 }
835
836 fn assert_pending_horizon_matches_wal(
844 &mut self,
845 cx: &Cx,
846 last_commit_frame: usize,
847 ) -> Result<()> {
848 let generation = self.wal.generation_identity();
849 if self
850 .pending_publication_generation
851 .is_some_and(|staged| staged != generation)
852 {
853 return Err(FrankenError::WalCorrupt {
854 detail: "WAL generation changed before the staged commit horizon was published"
855 .to_owned(),
856 });
857 }
858
859 let frame_count = self.wal.frame_count();
860 if last_commit_frame >= frame_count {
861 return Err(FrankenError::WalCorrupt {
862 detail: format!(
863 "staged commit horizon {last_commit_frame} exceeds WAL frame count {frame_count}"
864 ),
865 });
866 }
867
868 let live_last_commit = self.wal.last_commit_frame(cx)?;
869 if live_last_commit.is_none_or(|live| live < last_commit_frame) {
870 return Err(FrankenError::WalCorrupt {
871 detail: format!(
872 "WAL does not report staged commit horizon {last_commit_frame} as committed"
873 ),
874 });
875 }
876
877 if self
878 .pending_publication_frames
879 .iter()
880 .any(|frame| frame.frame_index >= frame_count)
881 {
882 return Err(FrankenError::WalCorrupt {
883 detail: format!(
884 "a staged publication frame lies beyond WAL frame count {frame_count}"
885 ),
886 });
887 }
888
889 Ok(())
890 }
891
892 fn assert_publish_safe(&mut self, cx: &Cx, last_commit_frame: usize) -> Result<()> {
893 self.assert_pending_horizon_matches_wal(cx, last_commit_frame)?;
894
895 let publish_frame_count =
898 last_commit_frame
899 .checked_add(1)
900 .ok_or_else(|| FrankenError::WalCorrupt {
901 detail: "staged commit horizon overflows the publishable frame count"
902 .to_owned(),
903 })?;
904 self.wal.assert_publish_safe(publish_frame_count)?;
905
906 Ok(())
907 }
908
909 fn publish_authorized_deferred_commit(&mut self, cx: &Cx) -> Result<()> {
916 let Some(last_commit_frame) = self.pending_publication_commit else {
917 return Ok(());
918 };
919 self.assert_pending_horizon_matches_wal(cx, last_commit_frame)?;
920 self.publish_pending_commit_snapshot(cx, last_commit_frame, "authorized_deferred_commit");
921 self.pending_publication_commit = None;
922 self.pending_publication_generation = None;
923 Ok(())
924 }
925
926 fn publish_pending_after_sync(&mut self, cx: &Cx) -> Result<()> {
933 let Some(last_commit_frame) = self.pending_publication_commit else {
934 return Ok(());
935 };
936 self.assert_publish_safe(cx, last_commit_frame)?;
937 self.publish_pending_commit_snapshot(cx, last_commit_frame, "sync_publish_commit");
938 self.pending_publication_commit = None;
939 self.pending_publication_generation = None;
940 Ok(())
941 }
942
943 fn record_appended_frames<I>(&mut self, start_frame_index: usize, frames: I) -> Option<usize>
944 where
945 I: IntoIterator<Item = (u32, u32)>,
946 {
947 let mut last_commit_frame = None;
948 for (offset, (page_number, db_size_if_commit)) in frames.into_iter().enumerate() {
949 let frame_index = start_frame_index.saturating_add(offset);
950 self.pending_publication_frames
951 .push(PendingPublicationFrame {
952 page_number,
953 frame_index,
954 is_commit: db_size_if_commit != 0,
955 });
956 if db_size_if_commit != 0 {
957 last_commit_frame = Some(frame_index);
958 }
959 }
960 last_commit_frame
961 }
962
963 fn publish_pending_commit_snapshot(
970 &mut self,
971 cx: &Cx,
972 last_commit_frame: usize,
973 scenario_id: &'static str,
974 ) {
975 let generation = self.wal.generation_identity();
976 let previous_last_commit = self.published_snapshot.last_commit_frame;
977 let can_extend_previous = self.published_snapshot.generation == generation
978 && self
979 .published_snapshot
980 .last_commit_frame
981 .is_none_or(|previous_last_commit| previous_last_commit < last_commit_frame);
982 let mut page_index = if can_extend_previous {
983 std::mem::replace(
984 &mut self.published_snapshot.page_index,
985 Arc::new(HashMap::new()),
986 )
987 } else {
988 Arc::new(HashMap::new())
989 };
990 let mut index_is_partial = if can_extend_previous {
991 self.published_snapshot.index_is_partial
992 } else {
993 false
994 };
995 let previous_last_commit = if can_extend_previous {
996 previous_last_commit
997 } else {
998 None
999 };
1000 let previous_commit_count = if can_extend_previous {
1001 self.published_snapshot.commit_count
1002 } else {
1003 0
1004 };
1005
1006 let mut frame_delta_count = 0_usize;
1007 let mut commit_delta_count = 0_u64;
1008 for frame in &self.pending_publication_frames {
1009 if previous_last_commit
1010 .is_some_and(|previous_last_commit| frame.frame_index <= previous_last_commit)
1011 || frame.frame_index > last_commit_frame
1012 {
1013 continue;
1014 }
1015
1016 frame_delta_count = frame_delta_count.saturating_add(1);
1017 let page_index_map = Arc::make_mut(&mut page_index);
1018 if page_index_map.len() < self.page_index_cap
1019 || page_index_map.contains_key(&frame.page_number)
1020 {
1021 page_index_map.insert(frame.page_number, frame.frame_index);
1022 } else {
1023 index_is_partial = true;
1024 }
1025 if frame.is_commit {
1026 commit_delta_count = commit_delta_count.saturating_add(1);
1027 }
1028 }
1029
1030 if frame_delta_count == 0 {
1031 if can_extend_previous {
1038 self.published_snapshot.page_index = page_index;
1039 }
1040 self.pending_publication_frames.clear();
1041 return;
1042 }
1043
1044 let publication_seq = self.next_publication_seq;
1045 self.next_publication_seq = self.next_publication_seq.saturating_add(1);
1046 let latest_frame_entries = page_index.len();
1047 self.published_snapshot = WalPublishedSnapshot {
1048 publication_seq,
1049 generation,
1050 last_commit_frame: Some(last_commit_frame),
1051 commit_count: previous_commit_count.saturating_add(commit_delta_count),
1052 page_index,
1053 index_is_partial,
1054 };
1055 self.pending_publication_frames.clear();
1056
1057 tracing::trace!(
1058 target: "fsqlite.wal_publication",
1059 trace_id = cx.trace_id(),
1060 run_id = "wal-publication",
1061 scenario_id,
1062 wal_generation = generation.checkpoint_seq,
1063 wal_salt1 = generation.salts.salt1,
1064 wal_salt2 = generation.salts.salt2,
1065 publication_seq,
1066 frame_delta_count,
1067 latest_frame_entries,
1068 snapshot_age = 0_u64,
1069 lookup_mode = "published_visibility_map",
1070 fallback_reason = if index_is_partial {
1071 "partial_index_cap"
1072 } else {
1073 "none"
1074 },
1075 "published WAL visibility snapshot from commit path"
1076 );
1077 }
1078}
1079
1080fn to_wal_mode(mode: CheckpointMode) -> WalCheckpointMode {
1082 match mode {
1083 CheckpointMode::Passive => WalCheckpointMode::Passive,
1084 CheckpointMode::Full => WalCheckpointMode::Full,
1085 CheckpointMode::Restart => WalCheckpointMode::Restart,
1086 CheckpointMode::Truncate => WalCheckpointMode::Truncate,
1087 }
1088}
1089
1090impl<F: VfsFile> WalBackend for WalBackendAdapter<F> {
1091 fn begin_transaction<'a>(&'a mut self, cx: &'a Cx) -> WalFuture<'a, ()> {
1092 Box::pin(async move {
1093 if self.has_pending_publication() {
1099 return Err(FrankenError::Busy);
1100 }
1101 self.wal.refresh(cx).await?;
1104 self.publish_latest_committed_snapshot(cx, "begin_transaction")
1105 .await?;
1106 self.read_snapshot = Some(self.published_snapshot.clone());
1107 self.refresh_before_append = true;
1108 Ok(())
1109 })
1110 }
1111
1112 fn published_snapshot(&self) -> Option<WalPublicationSnapshot> {
1113 Some(Self::published_snapshot(self))
1114 }
1115
1116 fn pinned_read_snapshot(&self) -> Option<WalPublicationSnapshot> {
1117 Self::pinned_read_snapshot(self)
1118 }
1119
1120 fn refresh_published_snapshot<'a>(
1121 &'a mut self,
1122 cx: &'a Cx,
1123 ) -> WalFuture<'a, Option<WalPublicationSnapshot>> {
1124 Box::pin(async move { Self::refresh_published_snapshot(self, cx).await.map(Some) })
1125 }
1126
1127 fn publish_authorized_deferred_commit<'a>(&'a mut self, cx: &'a Cx) -> WalFuture<'a, ()> {
1128 Box::pin(async move { Self::publish_authorized_deferred_commit(self, cx) })
1129 }
1130
1131 fn append_frame<'a>(
1132 &'a mut self,
1133 cx: &'a Cx,
1134 page_number: u32,
1135 page_data: &'a [u8],
1136 db_size_if_commit: u32,
1137 ) -> WalFuture<'a, ()> {
1138 Box::pin(async move {
1139 if self.refresh_before_append {
1140 self.synchronize_publication_before_append(cx, "append_frame_pre_refresh")
1144 .await?;
1145 }
1146 let start_frame_index = self.wal.frame_count();
1147 self.wal
1148 .append_frame(cx, page_number, page_data, db_size_if_commit)
1149 .await?;
1150 self.refresh_before_append = false;
1151 let last_commit_frame =
1152 self.record_appended_frames(start_frame_index, [(page_number, db_size_if_commit)]);
1153
1154 #[cfg(all(not(target_arch = "wasm32"), feature = "native"))]
1157 if let Some(hook) = &mut self.fec_hook {
1158 match hook.on_frame(cx, page_number, page_data, db_size_if_commit) {
1159 Ok(Some(result)) => {
1160 debug!(
1161 pages = result.page_numbers.len(),
1162 k_source = result.k_source,
1163 symbols = result.symbols.len(),
1164 "FEC commit group encoded"
1165 );
1166 self.fec_pending.push(result);
1167 }
1168 Ok(None) => {}
1169 Err(e) => {
1170 warn!(error = %e, "FEC encoding failed; commit proceeds without repair symbols");
1172 }
1173 }
1174 }
1175
1176 if let Some(last_commit_frame) = last_commit_frame {
1177 self.stage_pending_commit_publication(last_commit_frame)?;
1180 }
1181
1182 Ok(())
1183 })
1184 }
1185
1186 fn append_frames<'a>(
1187 &'a mut self,
1188 cx: &'a Cx,
1189 frames: &'a [WalFrameRef<'a>],
1190 ) -> WalFuture<'a, ()> {
1191 Box::pin(async move {
1192 if frames.is_empty() {
1193 return Ok(());
1194 }
1195
1196 if self.refresh_before_append {
1197 self.synchronize_publication_before_append(cx, "append_frames_pre_refresh")
1198 .await?;
1199 }
1200
1201 let start_frame_index = self.wal.frame_count();
1202 let mut wal_frames = Vec::with_capacity(frames.len());
1203 for frame in frames {
1204 wal_frames.push(WalAppendFrameRef {
1205 page_number: frame.page_number,
1206 page_data: frame.page_data,
1207 db_size_if_commit: frame.db_size_if_commit,
1208 });
1209 }
1210 self.wal.append_frames(cx, &wal_frames).await?;
1211 self.refresh_before_append = false;
1212 let last_commit_frame = self.record_appended_frames(
1213 start_frame_index,
1214 frames
1215 .iter()
1216 .map(|frame| (frame.page_number, frame.db_size_if_commit)),
1217 );
1218
1219 #[cfg(all(not(target_arch = "wasm32"), feature = "native"))]
1220 if let Some(hook) = &mut self.fec_hook {
1221 for frame in frames {
1222 match hook.on_frame(
1223 cx,
1224 frame.page_number,
1225 frame.page_data,
1226 frame.db_size_if_commit,
1227 ) {
1228 Ok(Some(result)) => {
1229 debug!(
1230 pages = result.page_numbers.len(),
1231 k_source = result.k_source,
1232 symbols = result.symbols.len(),
1233 "FEC commit group encoded"
1234 );
1235 self.fec_pending.push(result);
1236 }
1237 Ok(None) => {}
1238 Err(e) => {
1239 warn!(
1240 error = %e,
1241 "FEC encoding failed; commit proceeds without repair symbols"
1242 );
1243 }
1244 }
1245 }
1246 }
1247
1248 if let Some(last_commit_frame) = last_commit_frame {
1249 self.stage_pending_commit_publication(last_commit_frame)?;
1251 }
1252
1253 Ok(())
1254 })
1255 }
1256
1257 fn append_frames_tracked<'a>(
1258 &'a mut self,
1259 cx: &'a Cx,
1260 frames: &'a [WalFrameRef<'a>],
1261 completion: VfsWriteCompletion,
1262 ) -> WalFuture<'a, ()> {
1263 Box::pin(async move {
1264 let mut preflight = WalWriteCompletionPreflight::new(Some(&completion));
1265 if frames.is_empty() {
1266 completion.complete_success();
1267 preflight.hand_off();
1268 return Ok(());
1269 }
1270
1271 if self.refresh_before_append {
1272 self.synchronize_publication_before_append(cx, "append_frames_pre_refresh")
1273 .await?;
1274 }
1275
1276 let start_frame_index = self.wal.frame_count();
1277 let mut wal_frames = Vec::with_capacity(frames.len());
1278 for frame in frames {
1279 wal_frames.push(WalAppendFrameRef {
1280 page_number: frame.page_number,
1281 page_data: frame.page_data,
1282 db_size_if_commit: frame.db_size_if_commit,
1283 });
1284 }
1285 preflight.hand_off();
1286 drop(preflight);
1287 self.wal
1288 .append_frames_tracked(cx, &wal_frames, completion)
1289 .await?;
1290 self.refresh_before_append = false;
1291 let last_commit_frame = self.record_appended_frames(
1292 start_frame_index,
1293 frames
1294 .iter()
1295 .map(|frame| (frame.page_number, frame.db_size_if_commit)),
1296 );
1297
1298 #[cfg(all(not(target_arch = "wasm32"), feature = "native"))]
1299 if let Some(hook) = &mut self.fec_hook {
1300 for frame in frames {
1301 match hook.on_frame(
1302 cx,
1303 frame.page_number,
1304 frame.page_data,
1305 frame.db_size_if_commit,
1306 ) {
1307 Ok(Some(result)) => {
1308 debug!(
1309 pages = result.page_numbers.len(),
1310 k_source = result.k_source,
1311 symbols = result.symbols.len(),
1312 "FEC commit group encoded"
1313 );
1314 self.fec_pending.push(result);
1315 }
1316 Ok(None) => {}
1317 Err(e) => {
1318 warn!(
1319 error = %e,
1320 "FEC encoding failed; commit proceeds without repair symbols"
1321 );
1322 }
1323 }
1324 }
1325 }
1326
1327 if let Some(last_commit_frame) = last_commit_frame {
1328 self.stage_pending_commit_publication(last_commit_frame)?;
1330 }
1331
1332 Ok(())
1333 })
1334 }
1335
1336 fn prepare_append_frames(
1337 &self,
1338 frames: &[WalFrameRef<'_>],
1339 ) -> Result<Option<PreparedWalFrameBatch>> {
1340 if frames.is_empty() {
1341 return Ok(None);
1342 }
1343
1344 let mut frame_bytes = Vec::new();
1345 let mut checksum_transforms = Vec::new();
1346 let last_commit_frame_offset = self.wal.prepare_frame_bytes_with_transforms_into(
1347 frames.len(),
1348 frames.iter().map(|frame| WalAppendFrameRef {
1349 page_number: frame.page_number,
1350 page_data: frame.page_data,
1351 db_size_if_commit: frame.db_size_if_commit,
1352 }),
1353 &mut frame_bytes,
1354 &mut checksum_transforms,
1355 )?;
1356 let frame_metas = frames
1357 .iter()
1358 .map(|frame| PreparedWalFrameMeta {
1359 page_number: frame.page_number,
1360 db_size_if_commit: frame.db_size_if_commit,
1361 })
1362 .collect();
1363
1364 Ok(Some(PreparedWalFrameBatch {
1365 frame_size: self.wal.frame_size(),
1366 page_data_offset: WAL_FRAME_HEADER_SIZE,
1367 big_endian_checksum: self.wal.big_endian_checksum(),
1368 frame_metas,
1369 checksum_transforms,
1370 frame_bytes,
1371 last_commit_frame_offset,
1372 finalized_for: None,
1373 finalized_running_checksum: None,
1374 }))
1375 }
1376
1377 fn finalize_prepared_frames(
1378 &self,
1379 _cx: &Cx,
1380 prepared: &mut PreparedWalFrameBatch,
1381 ) -> Result<()> {
1382 if prepared.frame_count() == 0 {
1383 return Ok(());
1384 }
1385 self.finalize_prepared_batch_against_current_state(prepared)
1389 }
1390
1391 fn append_prepared_frames<'a>(
1392 &'a mut self,
1393 cx: &'a Cx,
1394 prepared: &'a mut PreparedWalFrameBatch,
1395 ) -> WalFuture<'a, ()> {
1396 Box::pin(async move {
1397 if prepared.frame_count() == 0 {
1398 return Ok(());
1399 }
1400
1401 let can_reuse_prelock_finalize = self.refresh_before_append
1402 && self.prepared_batch_matches_current_state(prepared)
1403 && self.prepared_batch_matches_disk_state(cx, prepared).await?;
1404 if self.refresh_before_append && !can_reuse_prelock_finalize {
1405 self.synchronize_publication_before_append(cx, "append_prepared_pre_refresh")
1406 .await?;
1407 }
1408
1409 if !self.prepared_batch_matches_current_state(prepared) {
1410 self.finalize_prepared_batch_against_current_state(prepared)?;
1411 }
1412
1413 let start_frame_index = self.wal.frame_count();
1414 self.wal
1415 .append_finalized_prepared_frame_bytes(
1416 cx,
1417 &prepared.frame_bytes,
1418 prepared.frame_count(),
1419 Self::finalized_running_checksum(prepared)?,
1420 prepared.last_commit_frame_offset,
1421 )
1422 .await?;
1423 self.refresh_before_append = false;
1424 let last_commit_frame = self.record_appended_frames(
1425 start_frame_index,
1426 prepared
1427 .frame_metas
1428 .iter()
1429 .map(|frame| (frame.page_number, frame.db_size_if_commit)),
1430 );
1431
1432 #[cfg(all(not(target_arch = "wasm32"), feature = "native"))]
1433 if let Some(hook) = &mut self.fec_hook {
1434 for (index, frame) in prepared.frame_metas.iter().enumerate() {
1435 match hook.on_frame(
1436 cx,
1437 frame.page_number,
1438 prepared.page_data(index),
1439 frame.db_size_if_commit,
1440 ) {
1441 Ok(Some(result)) => {
1442 debug!(
1443 pages = result.page_numbers.len(),
1444 k_source = result.k_source,
1445 symbols = result.symbols.len(),
1446 "FEC commit group encoded"
1447 );
1448 self.fec_pending.push(result);
1449 }
1450 Ok(None) => {}
1451 Err(e) => {
1452 warn!(
1453 error = %e,
1454 "FEC encoding failed; commit proceeds without repair symbols"
1455 );
1456 }
1457 }
1458 }
1459 }
1460
1461 if let Some(last_commit_frame) = last_commit_frame {
1462 self.stage_pending_commit_publication(last_commit_frame)?;
1464 }
1465
1466 Ok(())
1467 })
1468 }
1469
1470 fn append_prepared_frames_tracked<'a>(
1471 &'a mut self,
1472 cx: &'a Cx,
1473 prepared: &'a mut PreparedWalFrameBatch,
1474 completion: VfsWriteCompletion,
1475 ) -> WalFuture<'a, ()> {
1476 Box::pin(async move {
1477 let mut preflight = WalWriteCompletionPreflight::new(Some(&completion));
1478 if prepared.frame_count() == 0 {
1479 completion.complete_success();
1480 preflight.hand_off();
1481 return Ok(());
1482 }
1483
1484 let can_reuse_prelock_finalize = self.refresh_before_append
1485 && self.prepared_batch_matches_current_state(prepared)
1486 && self.prepared_batch_matches_disk_state(cx, prepared).await?;
1487 if self.refresh_before_append && !can_reuse_prelock_finalize {
1488 self.synchronize_publication_before_append(cx, "append_prepared_pre_refresh")
1489 .await?;
1490 }
1491
1492 if !self.prepared_batch_matches_current_state(prepared) {
1493 self.finalize_prepared_batch_against_current_state(prepared)?;
1494 }
1495
1496 let start_frame_index = self.wal.frame_count();
1497 let final_running_checksum = Self::finalized_running_checksum(prepared)?;
1498 preflight.hand_off();
1499 drop(preflight);
1500 self.wal
1501 .append_finalized_prepared_frame_bytes_tracked(
1502 cx,
1503 &prepared.frame_bytes,
1504 prepared.frame_count(),
1505 final_running_checksum,
1506 prepared.last_commit_frame_offset,
1507 completion,
1508 )
1509 .await?;
1510 self.refresh_before_append = false;
1511 let last_commit_frame = self.record_appended_frames(
1512 start_frame_index,
1513 prepared
1514 .frame_metas
1515 .iter()
1516 .map(|frame| (frame.page_number, frame.db_size_if_commit)),
1517 );
1518
1519 #[cfg(all(not(target_arch = "wasm32"), feature = "native"))]
1520 if let Some(hook) = &mut self.fec_hook {
1521 for (index, frame) in prepared.frame_metas.iter().enumerate() {
1522 match hook.on_frame(
1523 cx,
1524 frame.page_number,
1525 prepared.page_data(index),
1526 frame.db_size_if_commit,
1527 ) {
1528 Ok(Some(result)) => {
1529 debug!(
1530 pages = result.page_numbers.len(),
1531 k_source = result.k_source,
1532 symbols = result.symbols.len(),
1533 "FEC commit group encoded"
1534 );
1535 self.fec_pending.push(result);
1536 }
1537 Ok(None) => {}
1538 Err(e) => {
1539 warn!(
1540 error = %e,
1541 "FEC encoding failed; commit proceeds without repair symbols"
1542 );
1543 }
1544 }
1545 }
1546 }
1547
1548 if let Some(last_commit_frame) = last_commit_frame {
1549 self.stage_pending_commit_publication(last_commit_frame)?;
1551 }
1552
1553 Ok(())
1554 })
1555 }
1556
1557 fn read_page<'a>(&'a mut self, cx: &'a Cx, page_number: u32) -> WalFuture<'a, Option<Vec<u8>>> {
1558 Box::pin(async move {
1559 let snapshot = if let Some(snapshot) = self.read_snapshot.clone() {
1560 snapshot
1561 } else {
1562 self.publish_latest_committed_snapshot(cx, "read_page_unpinned")
1563 .await?;
1564 self.published_snapshot.clone()
1565 };
1566 if snapshot.last_commit_frame.is_none() {
1567 return Ok(None);
1568 }
1569 let snapshot_age = self
1570 .published_snapshot
1571 .publication_seq
1572 .saturating_sub(snapshot.publication_seq);
1573
1574 let resolution = self
1575 .resolve_visible_frame(cx, &snapshot, page_number)
1576 .await?;
1577 let Some(frame_index) = resolution.frame_index() else {
1578 debug!(
1579 page_number,
1580 wal_checkpoint_seq = snapshot.generation.checkpoint_seq,
1581 wal_salt1 = snapshot.generation.salts.salt1,
1582 wal_salt2 = snapshot.generation.salts.salt2,
1583 publication_seq = snapshot.publication_seq,
1584 snapshot_age,
1585 lookup_mode = resolution.lookup_mode(),
1586 fallback_reason = resolution.fallback_reason(),
1587 "WAL adapter: page absent from current generation"
1588 );
1589 return Ok(None);
1590 };
1591
1592 let mut frame_buf = vec![0u8; self.wal.frame_size()];
1594 let header = self
1595 .wal
1596 .read_frame_into(cx, frame_index, &mut frame_buf)
1597 .await?;
1598
1599 if header.page_number != page_number {
1602 return Err(FrankenError::WalCorrupt {
1603 detail: format!(
1604 "WAL page index integrity failure: expected page {page_number} \
1605 at frame {frame_index}, found page {}",
1606 header.page_number
1607 ),
1608 });
1609 }
1610
1611 let header_size = fsqlite_wal::checksum::WAL_FRAME_HEADER_SIZE;
1623 let page_size = self.wal.page_size();
1624 frame_buf.copy_within(header_size.., 0);
1625 frame_buf.truncate(page_size);
1626 debug!(
1627 page_number,
1628 frame_index,
1629 wal_checkpoint_seq = snapshot.generation.checkpoint_seq,
1630 wal_salt1 = snapshot.generation.salts.salt1,
1631 wal_salt2 = snapshot.generation.salts.salt2,
1632 publication_seq = snapshot.publication_seq,
1633 snapshot_age,
1634 lookup_mode = resolution.lookup_mode(),
1635 fallback_reason = resolution.fallback_reason(),
1636 "WAL adapter: resolved page from current WAL generation"
1637 );
1638 Ok(Some(frame_buf))
1639 })
1640 }
1641
1642 fn read_page_pinned<'a>(
1644 &'a self,
1645 cx: &'a Cx,
1646 page_number: u32,
1647 ) -> WalFuture<'a, Option<Vec<u8>>> {
1648 Box::pin(async move {
1649 let snapshot = self.read_snapshot.as_ref().ok_or_else(|| {
1650 FrankenError::internal(
1651 "read_page_pinned called without a pinned read snapshot; \
1652 use read_page(&mut self) or call begin_transaction first",
1653 )
1654 })?;
1655 if snapshot.last_commit_frame.is_none() {
1656 return Ok(None);
1657 }
1658
1659 let resolution = self
1660 .resolve_visible_frame(cx, snapshot, page_number)
1661 .await?;
1662 let Some(frame_index) = resolution.frame_index() else {
1663 return Ok(None);
1664 };
1665
1666 let mut frame_buf = vec![0u8; self.wal.frame_size()];
1667 let header = self
1668 .wal
1669 .read_frame_into(cx, frame_index, &mut frame_buf)
1670 .await?;
1671
1672 if header.page_number != page_number {
1673 return Err(FrankenError::WalCorrupt {
1674 detail: format!(
1675 "WAL page index integrity failure: expected page {page_number} \
1676 at frame {frame_index}, found page {}",
1677 header.page_number
1678 ),
1679 });
1680 }
1681
1682 let header_size = fsqlite_wal::checksum::WAL_FRAME_HEADER_SIZE;
1694 let page_size = self.wal.page_size();
1695 frame_buf.copy_within(header_size.., 0);
1696 frame_buf.truncate(page_size);
1697 Ok(Some(frame_buf))
1698 })
1699 }
1700
1701 fn supports_pinned_reads(&self) -> bool {
1702 self.read_snapshot.is_some()
1703 }
1704
1705 fn committed_txns_since_page<'a>(
1706 &'a mut self,
1707 cx: &'a Cx,
1708 page_number: u32,
1709 ) -> WalFuture<'a, u64> {
1710 Box::pin(async move {
1711 let snapshot = if let Some(snapshot) = self.read_snapshot.clone() {
1712 snapshot
1713 } else {
1714 self.publish_latest_committed_snapshot(cx, "committed_txns_since_page")
1715 .await?;
1716 self.published_snapshot.clone()
1717 };
1718 let Some(last_commit_frame) = snapshot.last_commit_frame else {
1719 return Ok(0);
1720 };
1721
1722 let resolution = self
1723 .resolve_visible_frame(cx, &snapshot, page_number)
1724 .await?;
1725 let Some(last_page_frame) = resolution.frame_index() else {
1726 let mut total_commits = 0_u64;
1727 for frame_index in 0..=last_commit_frame {
1728 if self
1729 .wal
1730 .read_frame_header(cx, frame_index)
1731 .await?
1732 .is_commit()
1733 {
1734 total_commits = total_commits.saturating_add(1);
1735 }
1736 }
1737 return Ok(total_commits);
1738 };
1739
1740 let mut page_commit_frame = None;
1741 for frame_index in last_page_frame..=last_commit_frame {
1742 if self
1743 .wal
1744 .read_frame_header(cx, frame_index)
1745 .await?
1746 .is_commit()
1747 {
1748 page_commit_frame = Some(frame_index);
1749 break;
1750 }
1751 }
1752
1753 let Some(page_commit_frame) = page_commit_frame else {
1754 return Ok(0);
1755 };
1756
1757 let mut committed_txns_after_page = 0_u64;
1758 for frame_index in page_commit_frame.saturating_add(1)..=last_commit_frame {
1759 if self
1760 .wal
1761 .read_frame_header(cx, frame_index)
1762 .await?
1763 .is_commit()
1764 {
1765 committed_txns_after_page = committed_txns_after_page.saturating_add(1);
1766 }
1767 }
1768
1769 Ok(committed_txns_after_page)
1770 })
1771 }
1772
1773 fn conflicting_pages_since_snapshot<'a>(
1774 &'a mut self,
1775 cx: &'a Cx,
1776 snapshot: TransactionConflictSnapshot,
1777 page_numbers: &'a [u32],
1778 _page_baselines: &'a [TransactionConflictPageBaseline],
1779 ) -> WalFuture<'a, Vec<u32>> {
1780 Box::pin(async move {
1781 if page_numbers.is_empty() {
1782 return Ok(Vec::new());
1783 }
1784
1785 let mut candidates = page_numbers
1786 .iter()
1787 .copied()
1788 .filter(|page| *page != 0)
1789 .collect::<Vec<_>>();
1790 candidates.sort_unstable();
1791 candidates.dedup();
1792 if candidates.is_empty() {
1793 return Ok(Vec::new());
1794 }
1795
1796 self.wal.refresh(cx).await?;
1797 self.publish_latest_committed_snapshot(cx, "conflicting_pages_since_snapshot")
1798 .await?;
1799 let latest = self.published_snapshot();
1800 if latest.commit_count <= snapshot.commit_count
1801 && latest.generation == snapshot.generation
1802 && latest.last_commit_frame <= snapshot.last_commit_frame
1803 {
1804 return Ok(Vec::new());
1805 }
1806
1807 if latest.generation != snapshot.generation {
1808 return Ok(candidates);
1809 }
1810
1811 let Some(latest_last_commit_frame) = latest.last_commit_frame else {
1812 return Ok(Vec::new());
1813 };
1814 let start_frame = snapshot
1815 .last_commit_frame
1816 .map_or(0, |frame| frame.saturating_add(1));
1817 if start_frame > latest_last_commit_frame {
1818 return Ok(Vec::new());
1819 }
1820
1821 let candidate_set = candidates.iter().copied().collect::<HashSet<_>>();
1822 let mut conflicts = HashSet::<u32>::new();
1823 for frame_index in start_frame..=latest_last_commit_frame {
1824 let header = self.wal.read_frame_header(cx, frame_index).await?;
1825 if candidate_set.contains(&header.page_number) {
1826 conflicts.insert(header.page_number);
1827 }
1828 }
1829
1830 let mut conflicts = conflicts.into_iter().collect::<Vec<_>>();
1831 conflicts.sort_unstable();
1832 Ok(conflicts)
1833 })
1834 }
1835
1836 fn committed_txn_count<'a>(&'a mut self, cx: &'a Cx) -> WalFuture<'a, u64> {
1837 Box::pin(async move {
1838 let snapshot = if let Some(snapshot) = self.read_snapshot.clone() {
1839 snapshot
1840 } else {
1841 self.publish_latest_committed_snapshot(cx, "committed_txn_count")
1842 .await?;
1843 self.published_snapshot.clone()
1844 };
1845 Ok(snapshot.commit_count)
1846 })
1847 }
1848
1849 fn sync(&mut self, cx: &Cx) -> Result<()> {
1850 self.wal.sync(cx, SyncFlags::NORMAL)?;
1860 self.publish_pending_after_sync(cx)?;
1861 if !self.has_pending_publication() {
1871 self.refresh_before_append = true;
1872 }
1873 Ok(())
1874 }
1875
1876 fn frame_count(&self) -> usize {
1877 self.wal.frame_count()
1878 }
1879
1880 fn checkpoint<'a>(
1881 &'a mut self,
1882 cx: &'a Cx,
1883 mode: CheckpointMode,
1884 writer: &'a mut dyn CheckpointPageWriter,
1885 backfilled_frames: u32,
1886 oldest_reader_frame: Option<u32>,
1887 ) -> WalFuture<'a, CheckpointResult> {
1888 Box::pin(async move {
1889 if self.has_pending_publication() {
1896 return Err(FrankenError::CheckpointFailed {
1897 detail: "staged, unpublished frames remain; a successful commit sync must \
1898 drain them before checkpointing"
1899 .to_owned(),
1900 });
1901 }
1902 self.wal.refresh(cx).await?;
1904 self.refresh_before_append = true;
1905 let total_frames = u32::try_from(self.wal.frame_count()).unwrap_or(u32::MAX);
1906
1907 let state = CheckpointState {
1909 total_frames,
1910 backfilled_frames,
1911 oldest_reader_frame,
1912 };
1913
1914 let mut target = CheckpointTargetAdapterRef { writer };
1916
1917 let result =
1919 execute_checkpoint(cx, &mut self.wal, to_wal_mode(mode), state, &mut target)
1920 .await?;
1921
1922 #[cfg(all(not(target_arch = "wasm32"), feature = "native"))]
1926 if result.frames_backfilled > 0 {
1927 let drained = self.fec_pending.len();
1928 self.fec_pending.clear();
1929 if drained > 0 {
1930 debug!(
1931 drained_groups = drained,
1932 frames_backfilled = result.frames_backfilled,
1933 "FEC symbols reclaimed after checkpoint"
1934 );
1935 }
1936 }
1937
1938 #[cfg(all(not(target_arch = "wasm32"), feature = "native"))]
1941 if result.wal_was_reset {
1942 self.fec_discard();
1943 }
1944 if result.wal_was_reset {
1945 self.invalidate_publication();
1946 }
1947
1948 self.publish_latest_committed_snapshot(cx, "checkpoint")
1949 .await?;
1950
1951 Ok(CheckpointResult {
1952 total_frames,
1953 frames_backfilled: result.frames_backfilled,
1954 completed: result.plan.completes_checkpoint(),
1955 wal_was_reset: result.wal_was_reset,
1956 requested_mode: mode,
1957 effective_mode: mode,
1958 })
1959 })
1960 }
1961}
1962
1963const MIN_DURABLE_CERTIFICATE_RECORD_SIZE: usize =
1964 ParallelWalDurableCertificateRecord::MIN_ENCODED_SIZE;
1965const DURABLE_CERTIFICATE_RECORD_HEADER_SIZE: usize = 14;
1966const MAX_ORPHAN_CERTIFICATE_LOOKBACK: usize = 64;
1967
1968fn durable_certificate_declared_len(bytes: &[u8]) -> Option<usize> {
1969 let length_bytes = bytes.get(10..DURABLE_CERTIFICATE_RECORD_HEADER_SIZE)?;
1970 usize::try_from(u32::from_le_bytes([
1971 length_bytes[0],
1972 length_bytes[1],
1973 length_bytes[2],
1974 length_bytes[3],
1975 ]))
1976 .ok()
1977}
1978
1979fn durable_certificate_declares_len(bytes: &[u8], expected: usize) -> bool {
1980 durable_certificate_declared_len(bytes).is_some_and(|actual| actual.cmp(&expected).is_eq())
1981}
1982
1983fn decode_durable_certificate_record(
1984 bytes: &[u8],
1985 location: &str,
1986) -> Result<ParallelWalDurableCertificateRecord> {
1987 ParallelWalDurableCertificateRecord::from_bytes(bytes).map_err(|error| {
1988 FrankenError::WalCorrupt {
1989 detail: format!("parallel WAL certificate {location} is invalid: {error}"),
1990 }
1991 })
1992}
1993
1994fn validate_incomplete_certificate_suffix(bytes: &[u8], anchored: bool) -> Result<()> {
1995 if bytes.is_empty() {
1996 return Ok(());
1997 }
1998 if bytes.len() > PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE {
1999 return Err(FrankenError::WalCorrupt {
2000 detail: format!(
2001 "parallel WAL certificate torn suffix exceeds {} bytes",
2002 PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE
2003 ),
2004 });
2005 }
2006
2007 if bytes.len() < PARALLEL_WAL_DURABLE_CERTIFICATE_MAGIC.len() {
2008 if anchored || PARALLEL_WAL_DURABLE_CERTIFICATE_MAGIC.starts_with(bytes) {
2013 return Ok(());
2014 }
2015 return Err(FrankenError::WalCorrupt {
2016 detail: "parallel WAL certificate sidecar starts with non-record garbage".to_owned(),
2017 });
2018 }
2019 if !bytes.starts_with(&PARALLEL_WAL_DURABLE_CERTIFICATE_MAGIC) {
2020 return Err(FrankenError::WalCorrupt {
2021 detail: "parallel WAL certificate suffix does not start at a record boundary"
2022 .to_owned(),
2023 });
2024 }
2025 if bytes.len() < 10 {
2026 return Ok(());
2027 }
2028 let version = u16::from_le_bytes([bytes[8], bytes[9]]);
2029 if version != fsqlite_wal::PARALLEL_WAL_DURABLE_CERTIFICATE_RECORD_VERSION {
2030 return Err(FrankenError::WalCorrupt {
2031 detail: format!(
2032 "parallel WAL certificate suffix has unsupported record version {version}"
2033 ),
2034 });
2035 }
2036 if bytes.len() < DURABLE_CERTIFICATE_RECORD_HEADER_SIZE {
2037 return Ok(());
2038 }
2039 let declared_len =
2040 durable_certificate_declared_len(bytes).ok_or_else(|| FrankenError::WalCorrupt {
2041 detail: "parallel WAL certificate suffix length exceeds usize".to_owned(),
2042 })?;
2043 if !(MIN_DURABLE_CERTIFICATE_RECORD_SIZE..=PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE)
2044 .contains(&declared_len)
2045 {
2046 return Err(FrankenError::WalCorrupt {
2047 detail: format!(
2048 "parallel WAL certificate suffix declares invalid record length {declared_len}"
2049 ),
2050 });
2051 }
2052 if bytes.len() < declared_len {
2053 return Ok(());
2054 }
2055
2056 decode_durable_certificate_record(&bytes[..declared_len], "suffix")?;
2061 Err(FrankenError::WalCorrupt {
2062 detail:
2063 "parallel WAL certificate sidecar contains a complete record outside the footer chain"
2064 .to_owned(),
2065 })
2066}
2067
2068fn combine_sidecar_io_results<const N: usize>(
2069 context: &str,
2070 results: [(&str, Result<()>); N],
2071) -> Result<()> {
2072 let failures = results
2073 .into_iter()
2074 .filter_map(|(stage, result)| result.err().map(|error| (stage, error)))
2075 .collect::<Vec<_>>();
2076 if failures.is_empty() {
2077 return Ok(());
2078 }
2079 if failures.len() == 1 {
2080 return failures
2081 .into_iter()
2082 .next()
2083 .map_or(Ok(()), |(_, error)| Err(error));
2084 }
2085 let details = failures
2086 .iter()
2087 .map(|(stage, error)| format!("{stage}={error}"))
2088 .collect::<Vec<_>>()
2089 .join("; ");
2090 Err(FrankenError::internal(format!("{context}: {details}")))
2091}
2092
2093pub struct PathRefreshingWalBackend<V: Vfs>
2103where
2104 V::File: Send + Sync + 'static,
2105{
2106 vfs: V,
2107 db_path: PathBuf,
2108 wal_path: PathBuf,
2109 page_size: u32,
2110 create_missing: bool,
2111 #[cfg(all(feature = "native", any(unix, windows)))]
2112 namespace_binding: Option<Arc<DatabaseNamespaceBinding>>,
2113 inner: WalBackendAdapter<V::File>,
2114}
2115
2116impl<V> PathRefreshingWalBackend<V>
2117where
2118 V: Vfs + 'static,
2119 V::File: Send + Sync + 'static,
2120{
2121 #[must_use]
2122 pub fn new(
2123 vfs: V,
2124 db_path: impl AsRef<Path>,
2125 wal_path: impl AsRef<Path>,
2126 page_size: u32,
2127 wal: WalFile<V::File>,
2128 create_missing: bool,
2129 #[cfg(all(feature = "native", any(unix, windows)))] namespace_binding: Option<
2130 Arc<DatabaseNamespaceBinding>,
2131 >,
2132 ) -> Self {
2133 Self {
2134 vfs,
2135 db_path: db_path.as_ref().to_path_buf(),
2136 wal_path: wal_path.as_ref().to_path_buf(),
2137 page_size,
2138 create_missing,
2139 #[cfg(all(feature = "native", any(unix, windows)))]
2140 namespace_binding,
2141 inner: WalBackendAdapter::new(wal),
2142 }
2143 }
2144
2145 #[must_use]
2146 pub fn into_inner(self) -> WalBackendAdapter<V::File> {
2147 self.inner
2148 }
2149
2150 fn replace_inner(&mut self, cx: &Cx, wal: WalFile<V::File>) -> Result<()> {
2158 if self.inner.has_pending_publication() {
2159 let cleanup_cx = cx.create_child();
2160 let _cleanup_mask = cleanup_cx.masked();
2161 let _ = wal.close(&cleanup_cx);
2162 return Err(FrankenError::Busy);
2163 }
2164 let old = std::mem::replace(&mut self.inner, WalBackendAdapter::new(wal));
2165 let old_wal = old.into_inner()?;
2166 let _ = old_wal.close(cx);
2167 Ok(())
2168 }
2169
2170 async fn create_replacement_wal(&self, cx: &Cx) -> Result<WalFile<V::File>> {
2171 let flags = VfsOpenFlags::READWRITE | VfsOpenFlags::CREATE | VfsOpenFlags::WAL;
2172 let (file, _) = self.vfs.open(cx, Some(&self.wal_path), flags)?;
2173 let wal = WalFile::create(cx, file, self.page_size, 0, WalSalts::generate()).await?;
2176 if let Err(error) = self.vfs.sync_parent_directory(cx, &self.wal_path) {
2177 let cleanup_cx = cx.create_child();
2178 let _cleanup_mask = cleanup_cx.masked();
2179 let _ = wal.close(&cleanup_cx);
2180 return Err(error);
2181 }
2182 Ok(wal)
2183 }
2184
2185 async fn replace_with_created_wal(&mut self, cx: &Cx) -> Result<()> {
2186 let wal = self.create_replacement_wal(cx).await?;
2187 self.replace_inner(cx, wal)
2188 }
2189
2190 async fn open_replacement_wal(&self, cx: &Cx, path_file: V::File) -> Result<WalFile<V::File>> {
2191 let wal = WalFile::open(cx, path_file).await?;
2192 if u32::try_from(wal.page_size()).ok() != Some(self.page_size) {
2193 let actual_page_size = wal.page_size();
2194 let expected_page_size = self.page_size;
2195 let _ = wal.close(cx);
2196 return Err(FrankenError::WalCorrupt {
2197 detail: format!(
2198 "WAL page size {actual_page_size} does not match database page size {expected_page_size} during path refresh"
2199 ),
2200 });
2201 }
2202 Ok(wal)
2203 }
2204
2205 async fn path_header_matches_current_handle(
2206 &self,
2207 cx: &Cx,
2208 path_file: &V::File,
2209 ) -> Result<bool> {
2210 let mut header_buf = [0_u8; WAL_HEADER_SIZE];
2211 let bytes_read = path_file.read(cx, &mut header_buf, 0).await?;
2212 if bytes_read < WAL_HEADER_SIZE {
2213 return Ok(false);
2214 }
2215
2216 let path_header = WalHeader::from_bytes(&header_buf)?;
2217 if !validate_wal_header_checksum(&header_buf, path_header.big_endian_checksum())? {
2218 return Err(FrankenError::WalCorrupt {
2219 detail: "WAL header checksum mismatch during path refresh".to_owned(),
2220 });
2221 }
2222
2223 let current_header = self.inner.inner().header();
2224 Ok(path_header.magic == current_header.magic
2225 && path_header.format_version == current_header.format_version
2226 && path_header.page_size == current_header.page_size
2227 && path_header.checkpoint_seq == current_header.checkpoint_seq
2228 && path_header.salts == current_header.salts)
2229 }
2230
2231 async fn conflicts_after_generation_change(
2247 &mut self,
2248 cx: &Cx,
2249 page_numbers: &[u32],
2250 page_baselines: &[TransactionConflictPageBaseline],
2251 ) -> Vec<u32> {
2252 let mut candidates = page_numbers
2253 .iter()
2254 .copied()
2255 .filter(|page| *page != 0)
2256 .collect::<Vec<_>>();
2257 candidates.sort_unstable();
2258 candidates.dedup();
2259 if candidates.is_empty() {
2260 return Vec::new();
2261 }
2262
2263 let mut baselines = HashMap::<u32, [u8; 32]>::new();
2264 let mut ambiguous_baselines = HashSet::<u32>::new();
2265 for baseline in page_baselines {
2266 if baseline.page_number == 0 {
2267 continue;
2268 }
2269 if let Some(previous) = baselines.insert(baseline.page_number, baseline.page_hash)
2270 && previous != baseline.page_hash
2271 {
2272 ambiguous_baselines.insert(baseline.page_number);
2273 }
2274 }
2275
2276 let main_db_flags = VfsOpenFlags::READWRITE | VfsOpenFlags::MAIN_DB;
2277 let (mut db_file, _) = match self.vfs.open(cx, Some(&self.db_path), main_db_flags) {
2278 Ok(opened) => opened,
2279 Err(_) => return candidates,
2280 };
2281 let page_size = match usize::try_from(self.page_size) {
2282 Ok(page_size) if page_size > 0 => page_size,
2283 _ => {
2284 let _ = db_file.close(cx);
2285 return candidates;
2286 }
2287 };
2288
2289 let mut page_one = vec![0_u8; page_size];
2292 let page_one_read = match db_file.read(cx, &mut page_one, 0).await {
2293 Ok(bytes_read) => bytes_read,
2294 Err(_) => {
2295 let _ = db_file.close(cx);
2296 return candidates;
2297 }
2298 };
2299 let header_page_size =
2300 (page_one_read == page_size).then(|| sqlite_database_header_page_size(&page_one));
2301 if header_page_size.flatten() != Some(self.page_size) {
2302 let _ = db_file.close(cx);
2303 return candidates;
2304 }
2305
2306 let mut conflicts = Vec::new();
2307 for &page_number in &candidates {
2308 let Some(expected_hash) = baselines.get(&page_number).copied() else {
2309 conflicts.push(page_number);
2310 continue;
2311 };
2312 if ambiguous_baselines.contains(&page_number) {
2313 conflicts.push(page_number);
2314 continue;
2315 }
2316
2317 let current_page = match self.inner.read_page(cx, page_number).await {
2318 Ok(Some(page)) if page.len() == page_size => page,
2319 Ok(Some(_)) | Err(_) => {
2320 conflicts.push(page_number);
2321 continue;
2322 }
2323 Ok(None) => {
2324 let mut page = vec![0_u8; page_size];
2325 let page_offset = u64::from(page_number.saturating_sub(1))
2326 .saturating_mul(u64::from(self.page_size));
2327 match db_file.read(cx, &mut page, page_offset).await {
2328 Ok(bytes_read) if bytes_read == page_size => page,
2329 Ok(_) | Err(_) => {
2330 conflicts.push(page_number);
2331 continue;
2332 }
2333 }
2334 }
2335 };
2336 let current_hash = *blake3::hash(¤t_page).as_bytes();
2337 if current_hash != expected_hash {
2338 conflicts.push(page_number);
2339 }
2340 }
2341
2342 if db_file.close(cx).is_err() {
2343 return candidates;
2344 }
2345 conflicts.sort_unstable();
2346 conflicts.dedup();
2347 conflicts
2348 }
2349
2350 async fn ensure_current_wal_path(&mut self, cx: &Cx) -> Result<()> {
2351 #[cfg(all(feature = "native", any(unix, windows)))]
2352 if let Some(binding) = &self.namespace_binding {
2353 binding.validate_path_identity()?;
2354 }
2355 if !self.vfs.access(cx, &self.wal_path, AccessFlags::EXISTS)? {
2356 if self.create_missing {
2357 return self.replace_with_created_wal(cx).await;
2358 }
2359 return Ok(());
2360 }
2361
2362 let flags = VfsOpenFlags::READWRITE | VfsOpenFlags::WAL;
2363 let (mut path_file, _) = self.vfs.open(cx, Some(&self.wal_path), flags)?;
2364 let path_size = path_file.file_size(cx)?;
2365 if path_size < u64::try_from(WAL_HEADER_SIZE).unwrap_or(32) {
2366 let _ = path_file.close(cx);
2367 if self.create_missing {
2368 return self.replace_with_created_wal(cx).await;
2369 }
2370 return Ok(());
2371 }
2372
2373 let current_size = self.inner.inner().file().file_size(cx).unwrap_or(u64::MAX);
2374 let path_matches_current = if path_size == current_size {
2375 match self
2376 .path_header_matches_current_handle(cx, &path_file)
2377 .await
2378 {
2379 Ok(matches) => matches,
2380 Err(err) => {
2381 let _ = path_file.close(cx);
2382 return Err(err);
2383 }
2384 }
2385 } else {
2386 false
2387 };
2388 if !path_matches_current {
2389 let wal = self.open_replacement_wal(cx, path_file).await?;
2390 self.replace_inner(cx, wal)?;
2391 } else {
2392 let _ = path_file.close(cx);
2393 }
2394 Ok(())
2395 }
2396
2397 fn certificate_sidecar_path(&self) -> PathBuf {
2398 let mut path = self.wal_path.as_os_str().to_owned();
2399 path.push("-cert");
2400 PathBuf::from(path)
2401 }
2402
2403 fn certificate_checkpoint_handoff_path(&self) -> PathBuf {
2404 let mut path = self.wal_path.as_os_str().to_owned();
2405 path.push("-cert-head");
2406 PathBuf::from(path)
2407 }
2408
2409 async fn read_certificate_sidecar_exact(
2410 file: &V::File,
2411 cx: &Cx,
2412 offset: u64,
2413 len: usize,
2414 location: &str,
2415 ) -> Result<Vec<u8>> {
2416 let mut bytes = vec![0_u8; len];
2417 let bytes_read = file.read(cx, &mut bytes, offset).await?;
2418 if bytes_read != len {
2419 return Err(FrankenError::WalCorrupt {
2420 detail: format!(
2421 "parallel WAL certificate {location} at offset {offset} was short-read: got {bytes_read} of {len}"
2422 ),
2423 });
2424 }
2425 Ok(bytes)
2426 }
2427
2428 async fn read_certificate_record_ending_at(
2429 file: &V::File,
2430 cx: &Cx,
2431 record_end: u64,
2432 ) -> Result<(u64, ParallelWalDurableCertificateRecord)> {
2433 let footer_size =
2434 u64::try_from(ParallelWalDurableCertificateRecord::LENGTH_FOOTER_SIZE).unwrap_or(4);
2435 let footer_offset = record_end.checked_sub(footer_size).ok_or_else(|| {
2436 FrankenError::WalCorrupt {
2437 detail: format!(
2438 "parallel WAL certificate record ending at {record_end} has no length footer"
2439 ),
2440 }
2441 })?;
2442 let footer = Self::read_certificate_sidecar_exact(
2443 file,
2444 cx,
2445 footer_offset,
2446 ParallelWalDurableCertificateRecord::LENGTH_FOOTER_SIZE,
2447 "length footer",
2448 )
2449 .await?;
2450 let record_len = usize::try_from(u32::from_le_bytes([
2451 footer[0], footer[1], footer[2], footer[3],
2452 ]))
2453 .map_err(|_| FrankenError::WalCorrupt {
2454 detail: "parallel WAL certificate footer length exceeds usize".to_owned(),
2455 })?;
2456 if !(MIN_DURABLE_CERTIFICATE_RECORD_SIZE..=PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE)
2457 .contains(&record_len)
2458 {
2459 return Err(FrankenError::WalCorrupt {
2460 detail: format!(
2461 "parallel WAL certificate footer declares invalid record length {record_len}"
2462 ),
2463 });
2464 }
2465 let record_len_u64 = u64::try_from(record_len).map_err(|_| FrankenError::WalCorrupt {
2466 detail: "parallel WAL certificate record length exceeds u64".to_owned(),
2467 })?;
2468 let record_start =
2469 record_end
2470 .checked_sub(record_len_u64)
2471 .ok_or_else(|| FrankenError::WalCorrupt {
2472 detail: format!(
2473 "parallel WAL certificate record length {record_len} exceeds end offset {record_end}"
2474 ),
2475 })?;
2476 let bytes =
2477 Self::read_certificate_sidecar_exact(file, cx, record_start, record_len, "record")
2478 .await?;
2479 let record = decode_durable_certificate_record(&bytes, "record")?;
2480 Ok((record_start, record))
2481 }
2482
2483 async fn prepare_certificate_sidecar_for_append(file: &mut V::File, cx: &Cx) -> Result<u64> {
2490 let file_size = file.file_size(cx)?;
2491 if file_size == 0 {
2492 return Ok(0);
2493 }
2494
2495 let footer_size =
2496 u64::try_from(ParallelWalDurableCertificateRecord::LENGTH_FOOTER_SIZE).unwrap_or(4);
2497 if file_size >= footer_size {
2498 let footer = Self::read_certificate_sidecar_exact(
2499 file,
2500 cx,
2501 file_size - footer_size,
2502 ParallelWalDurableCertificateRecord::LENGTH_FOOTER_SIZE,
2503 "append-boundary length footer",
2504 )
2505 .await?;
2506 let record_len = usize::try_from(u32::from_le_bytes([
2507 footer[0], footer[1], footer[2], footer[3],
2508 ]))
2509 .unwrap_or(usize::MAX);
2510 let record_len_u64 = u64::try_from(record_len).unwrap_or(u64::MAX);
2511 if (MIN_DURABLE_CERTIFICATE_RECORD_SIZE
2512 ..=PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE)
2513 .contains(&record_len)
2514 && record_len_u64 <= file_size
2515 {
2516 let record_start = file_size - record_len_u64;
2517 let bytes = Self::read_certificate_sidecar_exact(
2518 file,
2519 cx,
2520 record_start,
2521 record_len,
2522 "append-boundary record",
2523 )
2524 .await?;
2525 if bytes.starts_with(&PARALLEL_WAL_DURABLE_CERTIFICATE_MAGIC)
2526 || durable_certificate_declares_len(&bytes, record_len)
2527 {
2528 decode_durable_certificate_record(&bytes, "append-boundary record")?;
2529 return Ok(file_size);
2530 }
2531 }
2532 }
2533
2534 let recovery_window_size =
2538 PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE.saturating_mul(2);
2539 let recovery_window_size_u64 = u64::try_from(recovery_window_size).unwrap_or(u64::MAX);
2540 let tail_offset = file_size.saturating_sub(recovery_window_size_u64);
2541 let tail_len =
2542 usize::try_from(file_size - tail_offset).map_err(|_| FrankenError::WalCorrupt {
2543 detail: "parallel WAL certificate append-repair window exceeds usize".to_owned(),
2544 })?;
2545 let tail = Self::read_certificate_sidecar_exact(
2546 file,
2547 cx,
2548 tail_offset,
2549 tail_len,
2550 "append-repair window",
2551 )
2552 .await?;
2553 let minimum_candidate_end = tail
2554 .len()
2555 .saturating_sub(PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE)
2556 .max(ParallelWalDurableCertificateRecord::LENGTH_FOOTER_SIZE);
2557 let mut anchor_end = None;
2558 for candidate_end in (minimum_candidate_end..tail.len()).rev() {
2559 let footer_start =
2560 candidate_end - ParallelWalDurableCertificateRecord::LENGTH_FOOTER_SIZE;
2561 let footer = &tail[footer_start..candidate_end];
2562 let record_len = usize::try_from(u32::from_le_bytes([
2563 footer[0], footer[1], footer[2], footer[3],
2564 ]))
2565 .unwrap_or(usize::MAX);
2566 if !(MIN_DURABLE_CERTIFICATE_RECORD_SIZE
2567 ..=PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE)
2568 .contains(&record_len)
2569 || record_len > candidate_end
2570 {
2571 continue;
2572 }
2573 let record_start = candidate_end - record_len;
2574 let record_bytes = &tail[record_start..candidate_end];
2575 if !record_bytes.starts_with(&PARALLEL_WAL_DURABLE_CERTIFICATE_MAGIC)
2576 || !durable_certificate_declares_len(record_bytes, record_len)
2577 {
2578 continue;
2579 }
2580 if ParallelWalDurableCertificateRecord::from_bytes(record_bytes).is_ok() {
2581 anchor_end = Some(candidate_end);
2582 break;
2583 }
2584 }
2585
2586 let safe_end = if let Some(anchor_end) = anchor_end {
2587 validate_incomplete_certificate_suffix(&tail[anchor_end..], true)?;
2588 tail_offset
2589 .checked_add(u64::try_from(anchor_end).unwrap_or(u64::MAX))
2590 .ok_or_else(|| FrankenError::WalCorrupt {
2591 detail: "parallel WAL certificate append-repair boundary overflow".to_owned(),
2592 })?
2593 } else {
2594 if file_size
2595 > u64::try_from(PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE)
2596 .unwrap_or(u64::MAX)
2597 {
2598 return Err(FrankenError::WalCorrupt {
2599 detail: format!(
2600 "parallel WAL certificate sidecar has no valid append boundary within its bounded {}-byte recovery suffix",
2601 PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE
2602 ),
2603 });
2604 }
2605 validate_incomplete_certificate_suffix(&tail, false)?;
2606 0
2607 };
2608
2609 if safe_end < file_size {
2610 file.truncate(cx, safe_end)?;
2611 }
2612 Ok(safe_end)
2613 }
2614
2615 async fn append_durable_certificate_record(
2616 &self,
2617 cx: &Cx,
2618 certificate: &ParallelWalCommitCertificate,
2619 wal_frame_start: u64,
2620 wal_frame_end: u64,
2621 sync: bool,
2622 ) -> Result<()> {
2623 self.append_durable_certificate_record_with_completion(
2624 cx,
2625 certificate,
2626 wal_frame_start,
2627 wal_frame_end,
2628 sync,
2629 None,
2630 )
2631 .await
2632 }
2633
2634 async fn append_durable_certificate_record_with_completion(
2635 &self,
2636 cx: &Cx,
2637 certificate: &ParallelWalCommitCertificate,
2638 wal_frame_start: u64,
2639 wal_frame_end: u64,
2640 sync: bool,
2641 completion: Option<&VfsWriteCompletion>,
2642 ) -> Result<()> {
2643 let mut preflight = WalWriteCompletionPreflight::new(completion);
2644 let expected_frame_start = u64::try_from(self.inner.frame_count())
2645 .unwrap_or(u64::MAX)
2646 .saturating_add(1);
2647 if wal_frame_start != expected_frame_start {
2648 return Err(FrankenError::internal(format!(
2649 "parallel WAL certificate starts at frame {wal_frame_start}, expected {expected_frame_start}"
2650 )));
2651 }
2652 let record = ParallelWalDurableCertificateRecord::new(
2653 self.inner.inner().generation_identity(),
2654 wal_frame_start,
2655 wal_frame_end,
2656 certificate.clone(),
2657 )
2658 .map_err(|error| {
2659 FrankenError::internal(format!(
2660 "could not encode parallel WAL durability certificate: {error}"
2661 ))
2662 })?;
2663 let record_bytes = record.to_bytes();
2664 if record_bytes.len() > PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE {
2665 return Err(FrankenError::WalCorrupt {
2666 detail: format!(
2667 "parallel WAL certificate record is {} bytes; maximum is {}",
2668 record_bytes.len(),
2669 PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE
2670 ),
2671 });
2672 }
2673 let certificate_path = self.certificate_sidecar_path();
2674 let existed = self
2675 .vfs
2676 .access(cx, &certificate_path, AccessFlags::EXISTS)?;
2677 let flags = VfsOpenFlags::READWRITE | VfsOpenFlags::CREATE | VfsOpenFlags::WAL;
2678 let (mut file, _) = self.vfs.open(cx, Some(&certificate_path), flags)?;
2679 let append_offset = Self::prepare_certificate_sidecar_for_append(&mut file, cx).await?;
2680 preflight.hand_off();
2681 drop(preflight);
2682 let write_result = if let Some(completion) = completion {
2683 file.write_tracked(cx, &record_bytes, append_offset, completion.clone())
2684 .await
2685 } else {
2686 file.write(cx, &record_bytes, append_offset).await
2687 };
2688 if let Err(write_error) = write_result {
2689 let cleanup_cx = cx.create_child();
2694 let _cleanup_mask = cleanup_cx.masked();
2695 let cleanup_result = file.truncate(&cleanup_cx, append_offset);
2696 let close_result = file.close(&cleanup_cx);
2697 return combine_sidecar_io_results(
2698 "parallel WAL certificate append cleanup failed",
2699 [
2700 ("write", Err(write_error)),
2701 ("truncate", cleanup_result),
2702 ("close", close_result),
2703 ],
2704 );
2705 }
2706
2707 let finalization_cx = cx.create_child();
2711 let _finalization_mask = finalization_cx.masked();
2712 let sync_result = if sync {
2713 file.durable_sync(&finalization_cx, SyncKind::FullDurable)
2714 } else {
2715 Ok(())
2716 };
2717 let directory_sync_result = if sync && !existed && sync_result.is_ok() {
2718 self.vfs
2719 .sync_parent_directory(&finalization_cx, &certificate_path)
2720 } else {
2721 Ok(())
2722 };
2723 let close_result = file.close(&finalization_cx);
2724 combine_sidecar_io_results(
2725 "parallel WAL certificate append finalization failed",
2726 [
2727 ("file_sync", sync_result),
2728 ("directory_sync", directory_sync_result),
2729 ("close", close_result),
2730 ],
2731 )
2732 }
2733
2734 async fn reconcile_certificate_sidecar_record(
2735 &self,
2736 cx: &Cx,
2737 expected: &ParallelWalDurableCertificateRecord,
2738 remove_expected_orphan: bool,
2739 sync: bool,
2740 ) -> Result<bool> {
2741 let certificate_path = self.certificate_sidecar_path();
2742 if !self
2743 .vfs
2744 .access(cx, &certificate_path, AccessFlags::EXISTS)?
2745 {
2746 return Ok(false);
2747 }
2748
2749 let flags = VfsOpenFlags::READWRITE | VfsOpenFlags::WAL;
2750 let (mut file, _) = self.vfs.open(cx, Some(&certificate_path), flags)?;
2751 let reconciliation_result = async {
2752 let original_size = file.file_size(cx)?;
2753 let safe_end = Self::prepare_certificate_sidecar_for_append(&mut file, cx).await?;
2754 let latest = if safe_end == 0 {
2755 None
2756 } else {
2757 Some(Self::read_certificate_record_ending_at(&file, cx, safe_end).await?)
2758 };
2759 let latest_is_expected = latest
2760 .as_ref()
2761 .is_some_and(|(_, record)| record == expected);
2762 let sidecar_changed = if remove_expected_orphan
2763 && let Some((record_start, _)) = latest.as_ref()
2764 && latest_is_expected
2765 {
2766 file.truncate(cx, *record_start)?;
2767 true
2768 } else {
2769 safe_end != original_size
2770 };
2771 if sync && (latest_is_expected || sidecar_changed) {
2772 file.durable_sync(cx, SyncKind::FullDurable)?;
2773 }
2774 if sync && latest_is_expected && !remove_expected_orphan {
2775 self.vfs.sync_parent_directory(cx, &certificate_path)?;
2778 }
2779 Ok(latest_is_expected)
2780 }
2781 .await;
2782
2783 let cleanup_cx = cx.create_child();
2784 let _cleanup_mask = cleanup_cx.masked();
2785 let close_result = file.close(&cleanup_cx);
2786 match (reconciliation_result, close_result) {
2787 (Ok(latest_is_expected), Ok(())) => Ok(latest_is_expected),
2788 (Err(error), Ok(())) | (Ok(_), Err(error)) => Err(error),
2789 (Err(reconciliation_error), Err(close_error)) => Err(FrankenError::internal(format!(
2790 "parallel WAL certificate reconciliation failed and close also failed: reconciliation={reconciliation_error}; close={close_error}"
2791 ))),
2792 }
2793 }
2794
2795 async fn persist_checkpoint_certificate_handoff(
2796 &self,
2797 cx: &Cx,
2798 record: &ParallelWalDurableCertificateRecord,
2799 ) -> Result<()> {
2800 let handoff_path = self.certificate_checkpoint_handoff_path();
2801 let existed = self.vfs.access(cx, &handoff_path, AccessFlags::EXISTS)?;
2802 let flags = VfsOpenFlags::READWRITE | VfsOpenFlags::CREATE | VfsOpenFlags::WAL;
2803 let (mut file, _) = self.vfs.open(cx, Some(&handoff_path), flags)?;
2804 let record_bytes = record.to_bytes();
2805 if record_bytes.len() > PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE {
2806 let cleanup_cx = cx.create_child();
2807 let _cleanup_mask = cleanup_cx.masked();
2808 let close_result = file.close(&cleanup_cx);
2809 return combine_sidecar_io_results(
2810 "parallel WAL checkpoint certificate handoff is oversized",
2811 [
2812 (
2813 "record_size",
2814 Err(FrankenError::WalCorrupt {
2815 detail: format!(
2816 "parallel WAL checkpoint certificate handoff is {} bytes; maximum is {}",
2817 record_bytes.len(),
2818 PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE
2819 ),
2820 }),
2821 ),
2822 ("close", close_result),
2823 ],
2824 );
2825 }
2826 let mutation_cx = cx.create_child();
2831 let _mutation_mask = mutation_cx.masked();
2832 let truncate_result = file.truncate(&mutation_cx, 0);
2833 let write_result = if truncate_result.is_ok() {
2834 file.write(&mutation_cx, &record_bytes, 0).await
2835 } else {
2836 Ok(())
2837 };
2838 if let Err(write_error) = write_result {
2839 let cleanup_result = file.truncate(&mutation_cx, 0);
2840 let close_result = file.close(&mutation_cx);
2841 return combine_sidecar_io_results(
2842 "parallel WAL checkpoint certificate handoff cleanup failed",
2843 [
2844 ("truncate_before_write", truncate_result),
2845 ("write", Err(write_error)),
2846 ("truncate_after_write", cleanup_result),
2847 ("close", close_result),
2848 ],
2849 );
2850 }
2851 let sync_result = if truncate_result.is_ok() {
2852 file.durable_sync(&mutation_cx, SyncKind::FullDurable)
2853 } else {
2854 Ok(())
2855 };
2856 let directory_sync_result = if !existed && truncate_result.is_ok() && sync_result.is_ok() {
2857 self.vfs.sync_parent_directory(&mutation_cx, &handoff_path)
2858 } else {
2859 Ok(())
2860 };
2861 let close_result = file.close(&mutation_cx);
2862 combine_sidecar_io_results(
2863 "parallel WAL checkpoint certificate handoff finalization failed",
2864 [
2865 ("truncate", truncate_result),
2866 ("file_sync", sync_result),
2867 ("directory_sync", directory_sync_result),
2868 ("close", close_result),
2869 ],
2870 )
2871 }
2872
2873 async fn checkpoint_certificate_handoff(
2874 &self,
2875 cx: &Cx,
2876 ) -> Result<Option<ParallelWalCommitCertificate>> {
2877 let handoff_path = self.certificate_checkpoint_handoff_path();
2878 if !self.vfs.access(cx, &handoff_path, AccessFlags::EXISTS)? {
2879 return Ok(None);
2880 }
2881 let flags = VfsOpenFlags::READONLY | VfsOpenFlags::WAL;
2882 let (mut file, _) = self.vfs.open(cx, Some(&handoff_path), flags)?;
2883 let read_result = async {
2884 let file_size =
2885 usize::try_from(file.file_size(cx)?).map_err(|_| FrankenError::WalCorrupt {
2886 detail: "parallel WAL checkpoint certificate handoff exceeds usize".to_owned(),
2887 })?;
2888 if file_size == 0 || file_size > PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE {
2889 return Err(FrankenError::WalCorrupt {
2890 detail: format!(
2891 "parallel WAL checkpoint certificate handoff has invalid size {file_size}"
2892 ),
2893 });
2894 }
2895 let mut bytes = vec![0_u8; file_size];
2896 let bytes_read = file.read(cx, &mut bytes, 0).await?;
2897 if bytes_read != bytes.len() {
2898 return Err(FrankenError::WalCorrupt {
2899 detail: "parallel WAL checkpoint certificate handoff was short-read".to_owned(),
2900 });
2901 }
2902 let record =
2903 ParallelWalDurableCertificateRecord::from_bytes(&bytes).map_err(|error| {
2904 FrankenError::WalCorrupt {
2905 detail: format!(
2906 "parallel WAL checkpoint certificate handoff is invalid: {error}"
2907 ),
2908 }
2909 })?;
2910 Ok(Some(record.certificate))
2911 }
2912 .await;
2913 let cleanup_cx = cx.create_child();
2914 let _cleanup_mask = cleanup_cx.masked();
2915 let close_result = file.close(&cleanup_cx);
2916 match (read_result, close_result) {
2917 (Ok(certificate), Ok(())) => Ok(certificate),
2918 (Err(error), Ok(())) | (Ok(_), Err(error)) => Err(error),
2919 (Err(read_error), Err(close_error)) => Err(FrankenError::internal(format!(
2920 "parallel WAL checkpoint handoff read failed and close also failed: read={read_error}; close={close_error}"
2921 ))),
2922 }
2923 }
2924
2925 async fn wal_frame_payload_digest(
2926 &self,
2927 cx: &Cx,
2928 wal_frame_start: u64,
2929 wal_frame_end: u64,
2930 ) -> Result<[u8; 32]> {
2931 if wal_frame_start == 0 || wal_frame_end < wal_frame_start {
2932 return Err(FrankenError::WalCorrupt {
2933 detail: format!(
2934 "invalid parallel WAL digest interval {wal_frame_start}..={wal_frame_end}"
2935 ),
2936 });
2937 }
2938
2939 let mut digest = ParallelWalFramePayloadDigestBuilder::new();
2940 for frame_number in wal_frame_start..=wal_frame_end {
2941 let frame_index = usize::try_from(frame_number.saturating_sub(1)).map_err(|_| {
2942 FrankenError::WalCorrupt {
2943 detail: format!(
2944 "parallel WAL digest frame number {frame_number} exceeds usize"
2945 ),
2946 }
2947 })?;
2948 let (header, page_data) = self.inner.inner().read_frame(cx, frame_index).await?;
2949 let page_number =
2950 PageNumber::new(header.page_number).ok_or_else(|| FrankenError::WalCorrupt {
2951 detail: format!(
2952 "parallel WAL digest frame {frame_number} has invalid page number {}",
2953 header.page_number
2954 ),
2955 })?;
2956 digest.update(page_number, header.db_size, &page_data);
2957 }
2958 Ok(digest.finalize())
2959 }
2960
2961 async fn latest_authorized_durable_certificate_record(
2962 &self,
2963 cx: &Cx,
2964 ) -> Result<Option<ParallelWalDurableCertificateRecord>> {
2965 let certificate_path = self.certificate_sidecar_path();
2966 if !self
2967 .vfs
2968 .access(cx, &certificate_path, AccessFlags::EXISTS)?
2969 {
2970 return Ok(None);
2971 }
2972 let flags = VfsOpenFlags::READONLY | VfsOpenFlags::WAL;
2973 let (mut file, _) = self.vfs.open(cx, Some(&certificate_path), flags)?;
2974 let read_result = async {
2975 let file_size = file.file_size(cx)?;
2976 if file_size == 0 {
2977 return Ok(None);
2978 }
2979
2980 let footer_size =
2983 u64::try_from(ParallelWalDurableCertificateRecord::LENGTH_FOOTER_SIZE)
2984 .unwrap_or(4);
2985 let mut newest = None;
2986 if file_size >= footer_size {
2987 let footer_offset = file_size - footer_size;
2988 let footer = Self::read_certificate_sidecar_exact(
2989 &file,
2990 cx,
2991 footer_offset,
2992 ParallelWalDurableCertificateRecord::LENGTH_FOOTER_SIZE,
2993 "newest length footer",
2994 )
2995 .await?;
2996 let record_len = usize::try_from(u32::from_le_bytes([
2997 footer[0], footer[1], footer[2], footer[3],
2998 ]))
2999 .map_err(|_| FrankenError::WalCorrupt {
3000 detail: "parallel WAL certificate newest footer length exceeds usize"
3001 .to_owned(),
3002 })?;
3003 let record_len_u64 = u64::try_from(record_len).unwrap_or(u64::MAX);
3004 if (MIN_DURABLE_CERTIFICATE_RECORD_SIZE
3005 ..=PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE)
3006 .contains(&record_len)
3007 && record_len_u64 <= file_size
3008 {
3009 let record_start = file_size - record_len_u64;
3010 let bytes = Self::read_certificate_sidecar_exact(
3011 &file,
3012 cx,
3013 record_start,
3014 record_len,
3015 "newest record",
3016 )
3017 .await?;
3018 if bytes.starts_with(&PARALLEL_WAL_DURABLE_CERTIFICATE_MAGIC)
3023 || durable_certificate_declares_len(&bytes, record_len)
3024 {
3025 let record =
3026 decode_durable_certificate_record(&bytes, "newest record")?;
3027 newest = Some((record_start, record));
3028 }
3029 }
3030 }
3031
3032 if newest.is_none() {
3033 let recovery_window_size =
3040 PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE.saturating_mul(2);
3041 let recovery_window_size_u64 =
3042 u64::try_from(recovery_window_size).unwrap_or(u64::MAX);
3043 let tail_offset = file_size.saturating_sub(recovery_window_size_u64);
3044 let tail_len =
3045 usize::try_from(file_size - tail_offset).map_err(|_| {
3046 FrankenError::WalCorrupt {
3047 detail: "parallel WAL certificate recovery window exceeds usize"
3048 .to_owned(),
3049 }
3050 })?;
3051 let tail = Self::read_certificate_sidecar_exact(
3052 &file,
3053 cx,
3054 tail_offset,
3055 tail_len,
3056 "recovery window",
3057 )
3058 .await?;
3059 let minimum_candidate_end = tail
3060 .len()
3061 .saturating_sub(PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE)
3062 .max(ParallelWalDurableCertificateRecord::LENGTH_FOOTER_SIZE);
3063 let mut anchor = None;
3064 for candidate_end in (minimum_candidate_end..tail.len()).rev() {
3065 let footer_start = candidate_end
3066 - ParallelWalDurableCertificateRecord::LENGTH_FOOTER_SIZE;
3067 let footer = &tail[footer_start..candidate_end];
3068 let record_len = usize::try_from(u32::from_le_bytes([
3069 footer[0], footer[1], footer[2], footer[3],
3070 ]))
3071 .unwrap_or(usize::MAX);
3072 if !(MIN_DURABLE_CERTIFICATE_RECORD_SIZE
3073 ..=PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE)
3074 .contains(&record_len)
3075 || record_len > candidate_end
3076 {
3077 continue;
3078 }
3079 let record_start = candidate_end - record_len;
3080 let record_bytes = &tail[record_start..candidate_end];
3081 if !record_bytes.starts_with(&PARALLEL_WAL_DURABLE_CERTIFICATE_MAGIC)
3082 || !durable_certificate_declares_len(record_bytes, record_len)
3083 {
3084 continue;
3085 }
3086 if let Ok(record) =
3087 ParallelWalDurableCertificateRecord::from_bytes(record_bytes)
3088 {
3089 anchor = Some((record_start, candidate_end, record));
3090 break;
3091 }
3092 }
3093
3094 if let Some((record_start, record_end, record)) = anchor {
3095 validate_incomplete_certificate_suffix(&tail[record_end..], true)?;
3096 let absolute_start = tail_offset
3097 .checked_add(u64::try_from(record_start).unwrap_or(u64::MAX))
3098 .ok_or_else(|| FrankenError::WalCorrupt {
3099 detail:
3100 "parallel WAL certificate recovery anchor offset overflow"
3101 .to_owned(),
3102 })?;
3103 newest = Some((absolute_start, record));
3104 } else {
3105 if file_size
3106 > u64::try_from(PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE)
3107 .unwrap_or(u64::MAX)
3108 {
3109 return Err(FrankenError::WalCorrupt {
3110 detail: format!(
3111 "parallel WAL certificate sidecar has no valid record within its bounded {}-byte recovery suffix",
3112 PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE
3113 ),
3114 });
3115 }
3116 validate_incomplete_certificate_suffix(&tail, false)?;
3117 return Ok(None);
3118 }
3119 }
3120
3121 let valid_frame_count = u64::try_from(self.inner.frame_count()).unwrap_or(u64::MAX);
3122 let wal_generation = self.inner.inner().generation_identity();
3123 let (mut record_start, mut record) = newest.ok_or_else(|| {
3124 FrankenError::WalCorrupt {
3125 detail: "parallel WAL certificate recovery produced no record".to_owned(),
3126 }
3127 })?;
3128 let mut unauthorized_records = 0_usize;
3129 loop {
3130 if record.wal_generation != wal_generation {
3135 return Ok(None);
3136 }
3137 let frame_index =
3138 usize::try_from(record.wal_frame_end.saturating_sub(1)).map_err(|_| {
3139 FrankenError::WalCorrupt {
3140 detail: "parallel WAL certificate commit-marker index exceeds usize"
3141 .to_owned(),
3142 }
3143 })?;
3144 let commit_marker_frame = if frame_index < self.inner.frame_count()
3145 && self
3146 .inner
3147 .inner()
3148 .read_frame_header(cx, frame_index)
3149 .await?
3150 .is_commit()
3151 {
3152 record.wal_frame_end
3153 } else {
3154 0
3155 };
3156 let actual_wal_frame_payload_digest =
3157 if commit_marker_frame == record.wal_frame_end {
3158 Some(
3159 self.wal_frame_payload_digest(
3160 cx,
3161 record.wal_frame_start,
3162 record.wal_frame_end,
3163 )
3164 .await?,
3165 )
3166 } else {
3167 None
3168 };
3169 if actual_wal_frame_payload_digest.is_some_and(|actual_digest| {
3170 record.authorizes_wal_boundary(
3171 wal_generation,
3172 valid_frame_count,
3173 commit_marker_frame,
3174 actual_digest,
3175 )
3176 }) {
3177 return Ok(Some(record));
3178 }
3179
3180 unauthorized_records = unauthorized_records.saturating_add(1);
3181 if unauthorized_records > MAX_ORPHAN_CERTIFICATE_LOOKBACK {
3182 return Err(FrankenError::WalCorrupt {
3183 detail: format!(
3184 "parallel WAL certificate sidecar exceeded bounded orphan lookback {MAX_ORPHAN_CERTIFICATE_LOOKBACK}"
3185 ),
3186 });
3187 }
3188 tracing::debug!(
3189 target: "fsqlite::wal::durability_combiner",
3190 orphan_certificate_epoch = record.certificate.certificate_epoch,
3191 orphan_commit_seq_hi = record.certificate.commit_seq_hi.get(),
3192 orphan_wal_frame_end = record.wal_frame_end,
3193 lookback = unauthorized_records,
3194 "ignored unauthorized parallel WAL certificate tail"
3195 );
3196 if record_start == 0 {
3197 return Ok(None);
3198 }
3199 (record_start, record) =
3200 Self::read_certificate_record_ending_at(&file, cx, record_start).await?;
3201 }
3202 }
3203 .await;
3204 let cleanup_cx = cx.create_child();
3205 let _cleanup_mask = cleanup_cx.masked();
3206 let close_result = file.close(&cleanup_cx);
3207 match (read_result, close_result) {
3208 (Ok(certificate), Ok(())) => Ok(certificate),
3209 (Err(error), Ok(())) | (Ok(_), Err(error)) => Err(error),
3210 (Err(read_error), Err(close_error)) => Err(FrankenError::internal(format!(
3211 "parallel WAL certificate tail read failed and close also failed: read={read_error}; close={close_error}"
3212 ))),
3213 }
3214 }
3215}
3216
3217impl<V> WalBackend for PathRefreshingWalBackend<V>
3218where
3219 V: Vfs + 'static,
3220 V::File: Send + Sync + 'static,
3221{
3222 fn begin_transaction<'a>(&'a mut self, cx: &'a Cx) -> WalFuture<'a, ()> {
3223 Box::pin(async move {
3224 self.ensure_current_wal_path(cx).await?;
3225 self.inner.begin_transaction(cx).await
3226 })
3227 }
3228
3229 fn published_snapshot(&self) -> Option<WalPublicationSnapshot> {
3230 Some(self.inner.published_snapshot())
3231 }
3232
3233 fn pinned_read_snapshot(&self) -> Option<WalPublicationSnapshot> {
3234 self.inner.pinned_read_snapshot()
3235 }
3236
3237 fn refresh_published_snapshot<'a>(
3238 &'a mut self,
3239 cx: &'a Cx,
3240 ) -> WalFuture<'a, Option<WalPublicationSnapshot>> {
3241 Box::pin(async move {
3242 self.ensure_current_wal_path(cx).await?;
3243 self.inner.refresh_published_snapshot(cx).await.map(Some)
3244 })
3245 }
3246
3247 fn publish_authorized_deferred_commit<'a>(&'a mut self, cx: &'a Cx) -> WalFuture<'a, ()> {
3248 Box::pin(async move { self.inner.publish_authorized_deferred_commit(cx) })
3249 }
3250
3251 fn append_frame<'a>(
3252 &'a mut self,
3253 cx: &'a Cx,
3254 page_number: u32,
3255 page_data: &'a [u8],
3256 db_size_if_commit: u32,
3257 ) -> WalFuture<'a, ()> {
3258 Box::pin(async move {
3259 self.ensure_current_wal_path(cx).await?;
3260 self.inner
3261 .append_frame(cx, page_number, page_data, db_size_if_commit)
3262 .await
3263 })
3264 }
3265
3266 fn append_frames<'a>(
3267 &'a mut self,
3268 cx: &'a Cx,
3269 frames: &'a [WalFrameRef<'a>],
3270 ) -> WalFuture<'a, ()> {
3271 Box::pin(async move {
3272 self.ensure_current_wal_path(cx).await?;
3273 self.inner.append_frames(cx, frames).await
3274 })
3275 }
3276
3277 fn append_frames_tracked<'a>(
3278 &'a mut self,
3279 cx: &'a Cx,
3280 frames: &'a [WalFrameRef<'a>],
3281 completion: VfsWriteCompletion,
3282 ) -> WalFuture<'a, ()> {
3283 Box::pin(async move {
3284 let mut preflight = WalWriteCompletionPreflight::new(Some(&completion));
3285 self.ensure_current_wal_path(cx).await?;
3286 preflight.hand_off();
3287 drop(preflight);
3288 self.inner
3289 .append_frames_tracked(cx, frames, completion)
3290 .await
3291 })
3292 }
3293
3294 fn prepare_append_frames(
3295 &self,
3296 frames: &[WalFrameRef<'_>],
3297 ) -> Result<Option<PreparedWalFrameBatch>> {
3298 self.inner.prepare_append_frames(frames)
3299 }
3300
3301 fn finalize_prepared_frames(
3302 &self,
3303 cx: &Cx,
3304 prepared: &mut PreparedWalFrameBatch,
3305 ) -> Result<()> {
3306 self.inner.finalize_prepared_frames(cx, prepared)
3307 }
3308
3309 fn append_prepared_frames<'a>(
3310 &'a mut self,
3311 cx: &'a Cx,
3312 prepared: &'a mut PreparedWalFrameBatch,
3313 ) -> WalFuture<'a, ()> {
3314 Box::pin(async move {
3315 self.ensure_current_wal_path(cx).await?;
3316 self.inner.append_prepared_frames(cx, prepared).await
3317 })
3318 }
3319
3320 fn append_prepared_frames_tracked<'a>(
3321 &'a mut self,
3322 cx: &'a Cx,
3323 prepared: &'a mut PreparedWalFrameBatch,
3324 completion: VfsWriteCompletion,
3325 ) -> WalFuture<'a, ()> {
3326 Box::pin(async move {
3327 let mut preflight = WalWriteCompletionPreflight::new(Some(&completion));
3328 self.ensure_current_wal_path(cx).await?;
3329 preflight.hand_off();
3330 drop(preflight);
3331 self.inner
3332 .append_prepared_frames_tracked(cx, prepared, completion)
3333 .await
3334 })
3335 }
3336
3337 fn persist_parallel_wal_commit_certificate<'a>(
3338 &'a mut self,
3339 cx: &'a Cx,
3340 certificate: &'a ParallelWalCommitCertificate,
3341 wal_frame_start: u64,
3342 wal_frame_end: u64,
3343 sync: bool,
3344 ) -> WalFuture<'a, ()> {
3345 Box::pin(async move {
3346 self.ensure_current_wal_path(cx).await?;
3347 self.append_durable_certificate_record(
3348 cx,
3349 certificate,
3350 wal_frame_start,
3351 wal_frame_end,
3352 sync,
3353 )
3354 .await
3355 })
3356 }
3357
3358 fn persist_parallel_wal_commit_certificate_tracked<'a>(
3359 &'a mut self,
3360 cx: &'a Cx,
3361 certificate: &'a ParallelWalCommitCertificate,
3362 wal_frame_start: u64,
3363 wal_frame_end: u64,
3364 sync: bool,
3365 completion: VfsWriteCompletion,
3366 ) -> WalFuture<'a, ()> {
3367 Box::pin(async move {
3368 let mut preflight = WalWriteCompletionPreflight::new(Some(&completion));
3369 self.ensure_current_wal_path(cx).await?;
3370 preflight.hand_off();
3371 drop(preflight);
3372 self.append_durable_certificate_record_with_completion(
3373 cx,
3374 certificate,
3375 wal_frame_start,
3376 wal_frame_end,
3377 sync,
3378 Some(&completion),
3379 )
3380 .await
3381 })
3382 }
3383
3384 fn reconcile_parallel_wal_commit<'a>(
3385 &'a mut self,
3386 cx: &'a Cx,
3387 certificate: &'a ParallelWalCommitCertificate,
3388 wal_frame_start: u64,
3389 wal_frame_end: u64,
3390 sync: bool,
3391 ) -> WalFuture<'a, ParallelWalCommitReconciliation> {
3392 Box::pin(async move {
3393 self.ensure_current_wal_path(cx).await?;
3394 self.inner.wal.refresh(cx).await?;
3395 let wal_generation = self.inner.wal.generation_identity();
3396 let expected_record = ParallelWalDurableCertificateRecord::new(
3397 wal_generation,
3398 wal_frame_start,
3399 wal_frame_end,
3400 certificate.clone(),
3401 )
3402 .map_err(|error| {
3403 FrankenError::internal(format!(
3404 "could not reconstruct in-doubt parallel WAL certificate: {error}"
3405 ))
3406 })?;
3407
3408 let valid_frame_count = u64::try_from(self.inner.wal.frame_count()).unwrap_or(u64::MAX);
3409 let target_commit_present = if valid_frame_count < wal_frame_end {
3410 false
3411 } else {
3412 let target_index =
3413 usize::try_from(wal_frame_end.saturating_sub(1)).map_err(|_| {
3414 FrankenError::WalCorrupt {
3415 detail: "in-doubt WAL commit-marker index exceeds usize".to_owned(),
3416 }
3417 })?;
3418 self.inner
3419 .wal
3420 .read_frame_header(cx, target_index)
3421 .await?
3422 .is_commit()
3423 };
3424
3425 if target_commit_present {
3426 if valid_frame_count != wal_frame_end {
3427 return Err(FrankenError::WalCorrupt {
3428 detail: format!(
3429 "in-doubt parallel WAL interval ends at frame {wal_frame_end}, but the retained writer gate observed committed frame count {valid_frame_count}"
3430 ),
3431 });
3432 }
3433 let actual_wal_frame_payload_digest = self
3434 .wal_frame_payload_digest(cx, wal_frame_start, wal_frame_end)
3435 .await?;
3436 if !expected_record.authorizes_wal_boundary(
3437 wal_generation,
3438 valid_frame_count,
3439 wal_frame_end,
3440 actual_wal_frame_payload_digest,
3441 ) {
3442 return Err(FrankenError::WalCorrupt {
3443 detail: format!(
3444 "in-doubt parallel WAL interval {wal_frame_start}..={wal_frame_end} does not match its content-bound certificate"
3445 ),
3446 });
3447 }
3448 let sidecar_is_exact = self
3449 .reconcile_certificate_sidecar_record(cx, &expected_record, false, sync)
3450 .await?;
3451 if !sidecar_is_exact {
3452 return Err(FrankenError::WalCorrupt {
3453 detail: format!(
3454 "parallel WAL commit marker at frame {wal_frame_end} has no exact durable certificate"
3455 ),
3456 });
3457 }
3458 if sync {
3459 self.inner.wal.sync(cx, SyncFlags::NORMAL)?;
3460 self.vfs.sync_parent_directory(cx, &self.wal_path)?;
3461 }
3462 return Ok(ParallelWalCommitReconciliation::Authorized);
3463 }
3464
3465 let committed_prefix_before =
3466 wal_frame_start
3467 .checked_sub(1)
3468 .ok_or_else(|| FrankenError::WalCorrupt {
3469 detail: "parallel WAL recovery interval starts at frame zero".to_owned(),
3470 })?;
3471 if valid_frame_count != committed_prefix_before {
3472 return Err(FrankenError::WalCorrupt {
3473 detail: format!(
3474 "in-doubt WAL interval {wal_frame_start}..={wal_frame_end} has unexpected committed prefix {valid_frame_count}"
3475 ),
3476 });
3477 }
3478 self.reconcile_certificate_sidecar_record(cx, &expected_record, true, sync)
3483 .await?;
3484 self.inner.wal.repair_uncommitted_tail(cx)?;
3485 if sync {
3486 self.inner.wal.sync(cx, SyncFlags::NORMAL)?;
3487 self.vfs.sync_parent_directory(cx, &self.wal_path)?;
3488 }
3489 Ok(ParallelWalCommitReconciliation::NotCommitted)
3490 })
3491 }
3492
3493 fn latest_authorized_parallel_wal_commit_certificate<'a>(
3494 &'a mut self,
3495 cx: &'a Cx,
3496 ) -> WalFuture<'a, Option<ParallelWalCommitCertificate>> {
3497 Box::pin(async move {
3498 self.ensure_current_wal_path(cx).await?;
3499 if let Some(record) = self
3500 .latest_authorized_durable_certificate_record(cx)
3501 .await?
3502 {
3503 return Ok(Some(record.certificate));
3504 }
3505 self.checkpoint_certificate_handoff(cx).await
3506 })
3507 }
3508
3509 fn read_page<'a>(&'a mut self, cx: &'a Cx, page_number: u32) -> WalFuture<'a, Option<Vec<u8>>> {
3510 Box::pin(async move {
3511 self.ensure_current_wal_path(cx).await?;
3512 self.inner.read_page(cx, page_number).await
3513 })
3514 }
3515
3516 fn read_page_pinned<'a>(
3517 &'a self,
3518 cx: &'a Cx,
3519 page_number: u32,
3520 ) -> WalFuture<'a, Option<Vec<u8>>> {
3521 Box::pin(async move { self.inner.read_page_pinned(cx, page_number).await })
3522 }
3523
3524 fn supports_pinned_reads(&self) -> bool {
3525 self.inner.supports_pinned_reads()
3526 }
3527
3528 fn committed_txns_since_page<'a>(
3529 &'a mut self,
3530 cx: &'a Cx,
3531 page_number: u32,
3532 ) -> WalFuture<'a, u64> {
3533 Box::pin(async move {
3534 self.ensure_current_wal_path(cx).await?;
3535 self.inner.committed_txns_since_page(cx, page_number).await
3536 })
3537 }
3538
3539 fn conflicting_pages_since_snapshot<'a>(
3540 &'a mut self,
3541 cx: &'a Cx,
3542 snapshot: TransactionConflictSnapshot,
3543 page_numbers: &'a [u32],
3544 page_baselines: &'a [TransactionConflictPageBaseline],
3545 ) -> WalFuture<'a, Vec<u32>> {
3546 Box::pin(async move {
3547 self.ensure_current_wal_path(cx).await?;
3548 let latest = self.inner.refresh_published_snapshot(cx).await?;
3549 if latest.generation != snapshot.generation {
3550 return Ok(self
3551 .conflicts_after_generation_change(cx, page_numbers, page_baselines)
3552 .await);
3553 }
3554 self.inner
3555 .conflicting_pages_since_snapshot(cx, snapshot, page_numbers, page_baselines)
3556 .await
3557 })
3558 }
3559
3560 fn committed_txn_count<'a>(&'a mut self, cx: &'a Cx) -> WalFuture<'a, u64> {
3561 Box::pin(async move {
3562 self.ensure_current_wal_path(cx).await?;
3563 self.inner.committed_txn_count(cx).await
3564 })
3565 }
3566
3567 fn sync(&mut self, cx: &Cx) -> Result<()> {
3568 #[cfg(all(feature = "native", any(unix, windows)))]
3569 if let Some(binding) = &self.namespace_binding {
3570 binding.validate_path_identity()?;
3571 }
3572 self.inner.sync(cx)
3573 }
3574
3575 fn frame_count(&self) -> usize {
3576 self.inner.frame_count()
3577 }
3578
3579 fn checkpoint<'a>(
3580 &'a mut self,
3581 cx: &'a Cx,
3582 mode: CheckpointMode,
3583 writer: &'a mut dyn CheckpointPageWriter,
3584 backfilled_frames: u32,
3585 oldest_reader_frame: Option<u32>,
3586 ) -> WalFuture<'a, CheckpointResult> {
3587 Box::pin(async move {
3588 self.ensure_current_wal_path(cx).await?;
3589 let checkpoint_handoff = self
3590 .latest_authorized_durable_certificate_record(cx)
3591 .await?;
3592 if let Some(record) = checkpoint_handoff.as_ref() {
3593 self.persist_checkpoint_certificate_handoff(cx, record)
3600 .await?;
3601 }
3602 let result = self
3603 .inner
3604 .checkpoint(cx, mode, writer, backfilled_frames, oldest_reader_frame)
3605 .await?;
3606 Ok(result)
3607 })
3608 }
3609}
3610
3611struct CheckpointTargetAdapterRef<'a> {
3616 writer: &'a mut dyn CheckpointPageWriter,
3617}
3618
3619impl CheckpointTarget for CheckpointTargetAdapterRef<'_> {
3620 fn write_page<'a>(
3621 &'a mut self,
3622 cx: &'a Cx,
3623 page_no: PageNumber,
3624 data: &'a [u8],
3625 ) -> CheckpointTargetFuture<'a, ()> {
3626 Box::pin(async move { self.writer.write_page(cx, page_no, data).await })
3627 }
3628
3629 fn truncate_db<'a>(&'a mut self, cx: &'a Cx, n_pages: u32) -> CheckpointTargetFuture<'a, ()> {
3630 Box::pin(async move { self.writer.truncate(cx, n_pages).await })
3631 }
3632
3633 fn sync_db<'a>(&'a mut self, cx: &'a Cx) -> CheckpointTargetFuture<'a, ()> {
3634 Box::pin(async move { self.writer.sync(cx).await })
3635 }
3636}
3637
3638#[cfg(test)]
3643mod tests {
3644 use std::sync::Mutex;
3645
3646 use fsqlite_pager::MockCheckpointPageWriter;
3647 use fsqlite_pager::traits::WalFrameRef;
3648 use fsqlite_types::flags::VfsOpenFlags;
3649 use fsqlite_vfs::MemoryVfs;
3650 use fsqlite_vfs::traits::{Vfs, VfsFile};
3651 use fsqlite_wal::checksum::WalSalts;
3652
3653 use super::*;
3654
3655 const PAGE_SIZE: u32 = 4096;
3656 const CERTIFICATE_PATH: &str = "test.db-wal-cert";
3657 const CHECKPOINT_HANDOFF_PATH: &str = "test.db-wal-cert-head";
3658
3659 #[derive(Clone, Copy, Debug)]
3660 enum CheckpointHandoffWriteFault {
3661 Error,
3662 Pending,
3663 }
3664
3665 #[derive(Clone, Debug, Eq, PartialEq)]
3666 enum CertificateSyncObservation {
3667 Ordinary(PathBuf),
3668 Durable(PathBuf, SyncKind),
3669 }
3670
3671 #[derive(Debug, Default)]
3672 struct CheckpointHandoffFaultState {
3673 next_write: Option<CheckpointHandoffWriteFault>,
3674 fail_next_sync: bool,
3675 fail_next_wal_sync: bool,
3677 sync_observations: Vec<CertificateSyncObservation>,
3678 }
3679
3680 #[derive(Clone, Debug)]
3681 struct CheckpointHandoffFaultVfs {
3682 inner: MemoryVfs,
3683 faults: Arc<Mutex<CheckpointHandoffFaultState>>,
3684 }
3685
3686 impl CheckpointHandoffFaultVfs {
3687 fn new() -> Self {
3688 Self {
3689 inner: MemoryVfs::new(),
3690 faults: Arc::new(Mutex::new(CheckpointHandoffFaultState::default())),
3691 }
3692 }
3693
3694 fn fail_next_handoff_write(&self) {
3695 self.faults
3696 .lock()
3697 .unwrap_or_else(std::sync::PoisonError::into_inner)
3698 .next_write = Some(CheckpointHandoffWriteFault::Error);
3699 }
3700
3701 fn pend_next_handoff_write(&self) {
3702 self.faults
3703 .lock()
3704 .unwrap_or_else(std::sync::PoisonError::into_inner)
3705 .next_write = Some(CheckpointHandoffWriteFault::Pending);
3706 }
3707
3708 fn fail_next_handoff_sync(&self) {
3709 self.faults
3710 .lock()
3711 .unwrap_or_else(std::sync::PoisonError::into_inner)
3712 .fail_next_sync = true;
3713 }
3714
3715 fn fail_next_wal_sync(&self) {
3717 self.faults
3718 .lock()
3719 .unwrap_or_else(std::sync::PoisonError::into_inner)
3720 .fail_next_wal_sync = true;
3721 }
3722
3723 fn take_sync_observations(&self) -> Vec<CertificateSyncObservation> {
3724 std::mem::take(
3725 &mut self
3726 .faults
3727 .lock()
3728 .unwrap_or_else(std::sync::PoisonError::into_inner)
3729 .sync_observations,
3730 )
3731 }
3732 }
3733
3734 #[derive(Debug)]
3735 struct CheckpointHandoffFaultFile {
3736 inner: <MemoryVfs as Vfs>::File,
3737 faults: Arc<Mutex<CheckpointHandoffFaultState>>,
3738 path: Option<PathBuf>,
3739 is_checkpoint_handoff: bool,
3740 }
3741
3742 impl Vfs for CheckpointHandoffFaultVfs {
3743 type File = CheckpointHandoffFaultFile;
3744
3745 fn name(&self) -> &'static str {
3746 "checkpoint-handoff-fault"
3747 }
3748
3749 fn open(
3750 &self,
3751 cx: &Cx,
3752 path: Option<&Path>,
3753 flags: VfsOpenFlags,
3754 ) -> Result<(Self::File, VfsOpenFlags)> {
3755 let is_checkpoint_handoff =
3756 path.is_some_and(|candidate| candidate == Path::new(CHECKPOINT_HANDOFF_PATH));
3757 let (inner, actual_flags) = self.inner.open(cx, path, flags)?;
3758 Ok((
3759 CheckpointHandoffFaultFile {
3760 inner,
3761 faults: Arc::clone(&self.faults),
3762 path: path.map(Path::to_path_buf),
3763 is_checkpoint_handoff,
3764 },
3765 actual_flags,
3766 ))
3767 }
3768
3769 fn delete(&self, cx: &Cx, path: &Path, sync_dir: bool) -> Result<()> {
3770 self.inner.delete(cx, path, sync_dir)
3771 }
3772
3773 fn sync_parent_directory(&self, cx: &Cx, path: &Path) -> Result<()> {
3774 self.inner.sync_parent_directory(cx, path)
3775 }
3776
3777 fn access(&self, cx: &Cx, path: &Path, flags: AccessFlags) -> Result<bool> {
3778 self.inner.access(cx, path, flags)
3779 }
3780
3781 fn path_entry_exists(&self, cx: &Cx, path: &Path) -> Result<bool> {
3782 self.inner.path_entry_exists(cx, path)
3783 }
3784
3785 fn full_pathname(&self, cx: &Cx, path: &Path) -> Result<PathBuf> {
3786 self.inner.full_pathname(cx, path)
3787 }
3788
3789 fn randomness(&self, cx: &Cx, buf: &mut [u8]) {
3790 self.inner.randomness(cx, buf);
3791 }
3792
3793 fn current_time(&self, cx: &Cx) -> f64 {
3794 self.inner.current_time(cx)
3795 }
3796
3797 fn is_memory(&self) -> bool {
3798 true
3799 }
3800 }
3801
3802 impl VfsFile for CheckpointHandoffFaultFile {
3803 fn close(&mut self, cx: &Cx) -> Result<()> {
3804 self.inner.close(cx)
3805 }
3806
3807 fn file_identity(&self) -> Result<Option<fsqlite_vfs::FileIdentity>> {
3808 self.inner.file_identity()
3809 }
3810
3811 fn read<'a>(
3812 &'a self,
3813 cx: &'a Cx,
3814 buf: &'a mut [u8],
3815 offset: u64,
3816 ) -> impl std::future::Future<Output = Result<usize>> + Send + 'a {
3817 self.inner.read(cx, buf, offset)
3818 }
3819
3820 async fn write<'a>(&'a self, cx: &'a Cx, buf: &'a [u8], offset: u64) -> Result<()> {
3821 let fault = if self.is_checkpoint_handoff {
3822 self.faults
3823 .lock()
3824 .unwrap_or_else(std::sync::PoisonError::into_inner)
3825 .next_write
3826 .take()
3827 } else {
3828 None
3829 };
3830 match fault {
3831 Some(CheckpointHandoffWriteFault::Error) => Err(FrankenError::Io(
3832 std::io::Error::other("injected checkpoint handoff write failure"),
3833 )),
3834 Some(CheckpointHandoffWriteFault::Pending) => {
3835 std::future::pending::<Result<()>>().await
3836 }
3837 None => self.inner.write(cx, buf, offset).await,
3838 }
3839 }
3840
3841 fn truncate(&mut self, cx: &Cx, size: u64) -> Result<()> {
3842 self.inner.truncate(cx, size)
3843 }
3844
3845 fn sync(&mut self, cx: &Cx, flags: SyncFlags) -> Result<()> {
3846 let mut faults = self
3847 .faults
3848 .lock()
3849 .unwrap_or_else(std::sync::PoisonError::into_inner);
3850 if let Some(path) = self.path.as_ref().filter(|path| {
3851 path.as_path() == Path::new(CERTIFICATE_PATH)
3852 || path.as_path() == Path::new(CHECKPOINT_HANDOFF_PATH)
3853 }) {
3854 faults
3855 .sync_observations
3856 .push(CertificateSyncObservation::Ordinary(path.clone()));
3857 }
3858 let fail = self.is_checkpoint_handoff && std::mem::take(&mut faults.fail_next_sync);
3859 let fail_wal =
3860 !self.is_checkpoint_handoff && std::mem::take(&mut faults.fail_next_wal_sync);
3861 drop(faults);
3862 if fail {
3863 Err(FrankenError::Io(std::io::Error::other(
3864 "injected checkpoint handoff sync failure",
3865 )))
3866 } else if fail_wal {
3867 Err(FrankenError::Io(std::io::Error::other(
3868 "injected WAL sync failure",
3869 )))
3870 } else {
3871 self.inner.sync(cx, flags)
3872 }
3873 }
3874
3875 fn durable_sync(&mut self, cx: &Cx, kind: SyncKind) -> Result<()> {
3876 let mut faults = self
3877 .faults
3878 .lock()
3879 .unwrap_or_else(std::sync::PoisonError::into_inner);
3880 if let Some(path) = self.path.as_ref().filter(|path| {
3881 path.as_path() == Path::new(CERTIFICATE_PATH)
3882 || path.as_path() == Path::new(CHECKPOINT_HANDOFF_PATH)
3883 }) {
3884 faults
3885 .sync_observations
3886 .push(CertificateSyncObservation::Durable(path.clone(), kind));
3887 }
3888 let fail = self.is_checkpoint_handoff && std::mem::take(&mut faults.fail_next_sync);
3889 drop(faults);
3890 if fail {
3891 Err(FrankenError::Io(std::io::Error::other(
3892 "injected checkpoint handoff durable-sync failure",
3893 )))
3894 } else {
3895 self.inner.durable_sync(cx, kind)
3896 }
3897 }
3898
3899 fn file_size(&self, cx: &Cx) -> Result<u64> {
3900 self.inner.file_size(cx)
3901 }
3902
3903 fn lock(&mut self, cx: &Cx, level: fsqlite_types::LockLevel) -> Result<()> {
3904 self.inner.lock(cx, level)
3905 }
3906
3907 fn unlock(&mut self, cx: &Cx, level: fsqlite_types::LockLevel) -> Result<()> {
3908 self.inner.unlock(cx, level)
3909 }
3910
3911 fn lock_external_shared_snapshot(&mut self, cx: &Cx) -> Result<()> {
3912 self.inner.lock_external_shared_snapshot(cx)
3913 }
3914
3915 fn restore_external_shared_snapshot_attempt(&mut self, cx: &Cx) -> Result<()> {
3916 self.inner.restore_external_shared_snapshot_attempt(cx)
3917 }
3918
3919 fn lock_external_maintenance(&mut self, cx: &Cx, wal_mode: bool) -> Result<()> {
3920 self.inner.lock_external_maintenance(cx, wal_mode)
3921 }
3922
3923 fn restore_external_maintenance_attempt(&mut self, cx: &Cx) -> Result<()> {
3924 self.inner.restore_external_maintenance_attempt(cx)
3925 }
3926
3927 fn check_reserved_lock(&self, cx: &Cx) -> Result<bool> {
3928 self.inner.check_reserved_lock(cx)
3929 }
3930
3931 fn sector_size(&self) -> u32 {
3932 self.inner.sector_size()
3933 }
3934
3935 fn device_characteristics(&self) -> u32 {
3936 self.inner.device_characteristics()
3937 }
3938
3939 fn shm_map(
3940 &mut self,
3941 cx: &Cx,
3942 region: u32,
3943 size: u32,
3944 extend: bool,
3945 ) -> Result<fsqlite_vfs::ShmRegion> {
3946 self.inner.shm_map(cx, region, size, extend)
3947 }
3948
3949 fn shm_lock(&mut self, cx: &Cx, offset: u32, n: u32, flags: u32) -> Result<()> {
3950 self.inner.shm_lock(cx, offset, n, flags)
3951 }
3952
3953 fn shm_barrier(&self) {
3954 self.inner.shm_barrier();
3955 }
3956
3957 fn shm_unmap(&mut self, cx: &Cx, delete: bool) -> Result<()> {
3958 self.inner.shm_unmap(cx, delete)
3959 }
3960
3961 fn set_busy_timeout_ms(&mut self, ms: u64) {
3962 self.inner.set_busy_timeout_ms(ms);
3963 }
3964 }
3965
3966 fn init_wal_publication_test_tracing() {}
3982
3983 #[test]
3993 fn wal_publication_tracing_helper_installs_no_global_subscriber() {
3994 let before = tracing::dispatcher::has_been_set();
3995 init_wal_publication_test_tracing();
3996
3997 assert_eq!(
3998 before,
3999 tracing::dispatcher::has_been_set(),
4000 "init_wal_publication_test_tracing must not install or alter a global subscriber"
4001 );
4002 }
4003
4004 fn test_cx() -> Cx {
4005 Cx::default()
4006 }
4007
4008 fn test_salts() -> WalSalts {
4009 WalSalts {
4010 salt1: 0xDEAD_BEEF,
4011 salt2: 0xCAFE_BABE,
4012 }
4013 }
4014
4015 fn sample_page(seed: u8) -> Vec<u8> {
4016 let page_size = usize::try_from(PAGE_SIZE).expect("page size fits usize");
4017 let mut page = vec![0u8; page_size];
4018 for (i, byte) in page.iter_mut().enumerate() {
4019 let reduced = u8::try_from(i % 251).expect("modulo fits u8");
4020 *byte = reduced ^ seed;
4021 }
4022 page
4023 }
4024
4025 fn test_frame_payload_digest(
4026 page_number: u32,
4027 page_data: &[u8],
4028 db_size_if_commit: u32,
4029 ) -> [u8; 32] {
4030 let mut digest = ParallelWalFramePayloadDigestBuilder::new();
4031 digest.update(
4032 PageNumber::new(page_number).expect("test page number must be valid"),
4033 db_size_if_commit,
4034 page_data,
4035 );
4036 digest.finalize()
4037 }
4038
4039 fn sample_certificate(
4040 certificate_epoch: u64,
4041 commit_seq: u64,
4042 lane_record_counts: Vec<u32>,
4043 ) -> ParallelWalCommitCertificate {
4044 let lane_count = u16::try_from(lane_record_counts.len()).expect("test lane count fits u16");
4045 let mut certificate = ParallelWalCommitCertificate {
4046 format_version: fsqlite_wal::PARALLEL_WAL_COMMIT_CERTIFICATE_VERSION,
4047 residue: fsqlite_wal::ParallelWalOrderedResidue::CommitCertificateThenPublish,
4048 certificate_epoch,
4049 commit_seq_lo: fsqlite_types::CommitSeq::new(commit_seq),
4050 commit_seq_hi: fsqlite_types::CommitSeq::new(commit_seq),
4051 durable_segment_epoch: certificate_epoch,
4052 lane_count,
4053 lane_record_counts,
4054 db_size_pages: 1,
4055 page_set_size: 1,
4056 wal_frame_payload_digest: [0xA5; 32],
4057 certificate_crc32c: 0,
4058 fallback_active: false,
4059 };
4060 certificate.certificate_crc32c = certificate.computed_crc32c();
4061 certificate
4062 }
4063
4064 fn make_path_refreshing_backend(
4065 vfs: &MemoryVfs,
4066 cx: &Cx,
4067 ) -> PathRefreshingWalBackend<MemoryVfs> {
4068 let wal = WalFile::create(cx, open_wal_file(vfs, cx), PAGE_SIZE, 0, test_salts())
4069 .expect("create WAL");
4070 PathRefreshingWalBackend::new(
4071 vfs.clone(),
4072 std::path::Path::new("test.db"),
4073 std::path::Path::new("test.db-wal"),
4074 PAGE_SIZE,
4075 wal,
4076 true,
4077 #[cfg(all(feature = "native", any(unix, windows)))]
4078 None,
4079 )
4080 }
4081
4082 fn make_authorized_certificate_backend(
4083 vfs: &MemoryVfs,
4084 cx: &Cx,
4085 ) -> (
4086 PathRefreshingWalBackend<MemoryVfs>,
4087 ParallelWalCommitCertificate,
4088 ) {
4089 let mut backend = make_path_refreshing_backend(vfs, cx);
4090 let committed_page = sample_page(0x44);
4091 let mut certificate = sample_certificate(1, 1, vec![1]);
4092 certificate.wal_frame_payload_digest = test_frame_payload_digest(1, &committed_page, 1);
4093 certificate.certificate_crc32c = certificate.computed_crc32c();
4094 backend
4095 .persist_parallel_wal_commit_certificate(cx, &certificate, 1, 1, true)
4096 .expect("persist authorized certificate");
4097 backend
4098 .append_frame(cx, 1, &committed_page, 1)
4099 .expect("append matching commit marker");
4100 backend.sync(cx).expect("sync matching commit marker");
4101 (backend, certificate)
4102 }
4103
4104 struct AuthoritativeWalSnapshot {
4105 generation: WalGenerationIdentity,
4106 frame_count: usize,
4107 wal_bytes: Vec<u8>,
4108 certificate: ParallelWalCommitCertificate,
4109 committed_page: Vec<u8>,
4110 }
4111
4112 fn make_checkpoint_handoff_fault_backend(
4113 vfs: &CheckpointHandoffFaultVfs,
4114 cx: &Cx,
4115 ) -> (
4116 PathRefreshingWalBackend<CheckpointHandoffFaultVfs>,
4117 ParallelWalCommitCertificate,
4118 Vec<u8>,
4119 ) {
4120 let flags = VfsOpenFlags::READWRITE | VfsOpenFlags::CREATE | VfsOpenFlags::WAL;
4121 let (file, _) = vfs
4122 .open(cx, Some(Path::new("test.db-wal")), flags)
4123 .expect("open fault-injected WAL file");
4124 let wal = WalFile::create(cx, file, PAGE_SIZE, 0, test_salts())
4125 .expect("create fault-injected WAL");
4126 let mut backend = PathRefreshingWalBackend::new(
4127 vfs.clone(),
4128 Path::new("test.db"),
4129 Path::new("test.db-wal"),
4130 PAGE_SIZE,
4131 wal,
4132 true,
4133 #[cfg(all(feature = "native", any(unix, windows)))]
4134 None,
4135 );
4136 let committed_page = sample_page(0x47);
4137 let mut certificate = sample_certificate(1, 1, vec![1]);
4138 certificate.wal_frame_payload_digest = test_frame_payload_digest(1, &committed_page, 1);
4139 certificate.certificate_crc32c = certificate.computed_crc32c();
4140 backend
4141 .persist_parallel_wal_commit_certificate(cx, &certificate, 1, 1, true)
4142 .expect("persist authorized certificate");
4143 backend
4144 .append_frame(cx, 1, &committed_page, 1)
4145 .expect("append matching commit marker");
4146 backend.sync(cx).expect("sync matching commit marker");
4147 (backend, certificate, committed_page)
4148 }
4149
4150 fn read_fault_injected_wal(vfs: &CheckpointHandoffFaultVfs, cx: &Cx) -> Vec<u8> {
4151 let flags = VfsOpenFlags::READONLY | VfsOpenFlags::WAL;
4152 let (mut file, _) = vfs
4153 .open(cx, Some(Path::new("test.db-wal")), flags)
4154 .expect("open WAL snapshot");
4155 let len = usize::try_from(file.file_size(cx).expect("read WAL size"))
4156 .expect("WAL size fits usize");
4157 let mut bytes = vec![0_u8; len];
4158 assert_eq!(
4159 file.read(cx, &mut bytes, 0).expect("read WAL snapshot"),
4160 len
4161 );
4162 file.close(cx).expect("close WAL snapshot");
4163 bytes
4164 }
4165
4166 fn capture_authoritative_wal(
4167 backend: &PathRefreshingWalBackend<CheckpointHandoffFaultVfs>,
4168 vfs: &CheckpointHandoffFaultVfs,
4169 cx: &Cx,
4170 certificate: ParallelWalCommitCertificate,
4171 committed_page: Vec<u8>,
4172 ) -> AuthoritativeWalSnapshot {
4173 AuthoritativeWalSnapshot {
4174 generation: backend.inner.inner().generation_identity(),
4175 frame_count: backend.inner.frame_count(),
4176 wal_bytes: read_fault_injected_wal(vfs, cx),
4177 certificate,
4178 committed_page,
4179 }
4180 }
4181
4182 fn assert_authoritative_wal_unchanged(
4183 backend: &mut PathRefreshingWalBackend<CheckpointHandoffFaultVfs>,
4184 vfs: &CheckpointHandoffFaultVfs,
4185 cx: &Cx,
4186 before: &AuthoritativeWalSnapshot,
4187 ) {
4188 assert_eq!(
4189 backend.inner.inner().generation_identity(),
4190 before.generation,
4191 "checkpoint handoff failure must not reset the WAL generation"
4192 );
4193 assert_eq!(
4194 backend.inner.frame_count(),
4195 before.frame_count,
4196 "checkpoint handoff failure must not change the visible frame count"
4197 );
4198 assert_eq!(
4199 read_fault_injected_wal(vfs, cx),
4200 before.wal_bytes,
4201 "checkpoint handoff failure must leave the authoritative WAL byte-for-byte unchanged"
4202 );
4203 assert!(
4204 backend
4205 .inner
4206 .inner()
4207 .read_frame_header(cx, 0)
4208 .expect("read original commit frame")
4209 .is_commit(),
4210 "the original generation's commit marker must remain authoritative"
4211 );
4212 assert_eq!(
4213 backend
4214 .latest_authorized_parallel_wal_commit_certificate(cx)
4215 .expect("recover certificate from unchanged WAL generation"),
4216 Some(before.certificate.clone())
4217 );
4218 assert_eq!(
4219 backend
4220 .read_page(cx, 1)
4221 .expect("read committed page from unchanged WAL generation"),
4222 Some(before.committed_page.clone())
4223 );
4224 }
4225
4226 fn read_certificate_sidecar(vfs: &MemoryVfs, cx: &Cx) -> Vec<u8> {
4227 let path = std::path::Path::new("test.db-wal-cert");
4228 let (mut file, _) = vfs
4229 .open(cx, Some(path), VfsOpenFlags::READONLY | VfsOpenFlags::WAL)
4230 .expect("open certificate sidecar");
4231 let len = usize::try_from(file.file_size(cx).expect("read certificate sidecar size"))
4232 .expect("certificate sidecar size fits usize");
4233 let mut bytes = vec![0_u8; len];
4234 assert_eq!(
4235 file.read(cx, &mut bytes, 0)
4236 .expect("read certificate sidecar"),
4237 len
4238 );
4239 file.close(cx).expect("close certificate sidecar");
4240 bytes
4241 }
4242
4243 fn replace_certificate_sidecar(vfs: &MemoryVfs, cx: &Cx, bytes: &[u8]) {
4244 let path = std::path::Path::new("test.db-wal-cert");
4245 let flags = VfsOpenFlags::READWRITE | VfsOpenFlags::CREATE | VfsOpenFlags::WAL;
4246 let (mut file, _) = vfs
4247 .open(cx, Some(path), flags)
4248 .expect("open mutable certificate sidecar");
4249 file.truncate(cx, 0)
4250 .expect("truncate mutable certificate sidecar");
4251 file.write(cx, bytes, 0)
4252 .expect("replace certificate sidecar bytes");
4253 file.close(cx).expect("close mutable certificate sidecar");
4254 }
4255
4256 fn assert_wal_corrupt<T: std::fmt::Debug>(result: Result<T>, scenario: &str) {
4257 assert!(
4258 matches!(&result, Err(FrankenError::WalCorrupt { .. })),
4259 "{scenario} must fail closed with WalCorrupt, got {result:?}"
4260 );
4261 }
4262
4263 fn sqlite_page_one(encoded_page_size: u16) -> Vec<u8> {
4264 let mut page = sample_page(0x11);
4265 page[..16].copy_from_slice(b"SQLite format 3\0");
4266 page[16..18].copy_from_slice(&encoded_page_size.to_be_bytes());
4267 page
4268 }
4269
4270 fn write_main_db_pages(vfs: &MemoryVfs, cx: &Cx, pages: &[Vec<u8>]) {
4271 let flags = VfsOpenFlags::READWRITE | VfsOpenFlags::CREATE | VfsOpenFlags::MAIN_DB;
4272 let (mut file, _) = vfs
4273 .open(cx, Some(std::path::Path::new("test.db")), flags)
4274 .expect("open main database");
4275 file.truncate(cx, 0).expect("truncate main database");
4276 for (index, page) in pages.iter().enumerate() {
4277 let offset = u64::try_from(index)
4278 .expect("page index fits u64")
4279 .saturating_mul(u64::from(PAGE_SIZE));
4280 file.write(cx, page, offset).expect("write database page");
4281 }
4282 file.close(cx).expect("close main database");
4283 }
4284
4285 fn replacement_salts() -> WalSalts {
4286 WalSalts {
4287 salt1: 0x1234_5678,
4288 salt2: 0x9ABC_DEF0,
4289 }
4290 }
4291
4292 fn replace_path_visible_wal(vfs: &MemoryVfs, cx: &Cx) {
4293 let wal_path = std::path::Path::new("test.db-wal");
4294 vfs.delete(cx, wal_path, false)
4295 .expect("remove old path-visible WAL");
4296 let file = open_wal_file(vfs, cx);
4297 WalFile::create(cx, file, PAGE_SIZE, 1, replacement_salts())
4298 .expect("create replacement WAL")
4299 .close(cx)
4300 .expect("close replacement WAL");
4301 }
4302
4303 fn append_replacement_wal_page(
4304 vfs: &MemoryVfs,
4305 cx: &Cx,
4306 page_number: u32,
4307 page: &[u8],
4308 db_size_if_commit: u32,
4309 ) {
4310 let file = open_wal_file(vfs, cx);
4311 let wal = WalFile::open(cx, file).expect("open replacement WAL");
4312 let mut adapter = WalBackendAdapter::new(wal);
4313 adapter
4314 .append_frame(cx, page_number, page, db_size_if_commit)
4315 .expect("append replacement WAL page");
4316 adapter.sync(cx).expect("sync replacement WAL page");
4317 adapter
4318 .into_inner()
4319 .expect("sync drained the staged frames")
4320 .close(cx)
4321 .expect("close replacement WAL");
4322 }
4323
4324 fn make_generation_transition_backend(
4325 vfs: &MemoryVfs,
4326 cx: &Cx,
4327 ) -> (
4328 PathRefreshingWalBackend<MemoryVfs>,
4329 TransactionConflictSnapshot,
4330 Vec<u8>,
4331 ) {
4332 let page_one = sqlite_page_one(u16::try_from(PAGE_SIZE).expect("page size fits u16"));
4333 let page_two = sample_page(0x22);
4334 write_main_db_pages(vfs, cx, &[page_one.clone(), page_two.clone()]);
4335
4336 let file = open_wal_file(vfs, cx);
4337 let wal =
4338 WalFile::create(cx, file, PAGE_SIZE, 0, test_salts()).expect("create original WAL");
4339 let mut backend = PathRefreshingWalBackend::new(
4340 vfs.clone(),
4341 std::path::Path::new("test.db"),
4342 std::path::Path::new("test.db-wal"),
4343 PAGE_SIZE,
4344 wal,
4345 true,
4346 #[cfg(all(feature = "native", any(unix, windows)))]
4347 None,
4348 );
4349 backend
4350 .append_frame(cx, 1, &page_one, 0)
4351 .expect("append original page 1");
4352 backend
4353 .append_frame(cx, 2, &page_two, 2)
4354 .expect("append original commit");
4355 backend
4356 .begin_transaction(cx)
4357 .expect("pin original WAL generation");
4358 let pinned = backend.pinned_read_snapshot().expect("pinned WAL snapshot");
4359 let snapshot = TransactionConflictSnapshot {
4360 generation: pinned.generation,
4361 last_commit_frame: pinned.last_commit_frame,
4362 commit_count: pinned.commit_count,
4363 };
4364 replace_path_visible_wal(vfs, cx);
4365 (backend, snapshot, page_two)
4366 }
4367
4368 #[test]
4369 fn durable_certificate_sidecar_precedes_and_reconstructs_wal_commit() {
4370 let cx = test_cx();
4371 let vfs = MemoryVfs::new();
4372 let committed_page = sample_page(0x44);
4373 let file = open_wal_file(&vfs, &cx);
4374 let wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create WAL");
4375 let mut backend = PathRefreshingWalBackend::new(
4376 vfs.clone(),
4377 std::path::Path::new("test.db"),
4378 std::path::Path::new("test.db-wal"),
4379 PAGE_SIZE,
4380 wal,
4381 true,
4382 #[cfg(all(feature = "native", any(unix, windows)))]
4383 None,
4384 );
4385 let mut certificate = ParallelWalCommitCertificate {
4386 format_version: fsqlite_wal::PARALLEL_WAL_COMMIT_CERTIFICATE_VERSION,
4387 residue: fsqlite_wal::ParallelWalOrderedResidue::CommitCertificateThenPublish,
4388 certificate_epoch: 1,
4389 commit_seq_lo: fsqlite_types::CommitSeq::new(1),
4390 commit_seq_hi: fsqlite_types::CommitSeq::new(1),
4391 durable_segment_epoch: 1,
4392 lane_count: 1,
4393 lane_record_counts: vec![1],
4394 db_size_pages: 1,
4395 page_set_size: 1,
4396 wal_frame_payload_digest: test_frame_payload_digest(1, &committed_page, 1),
4397 certificate_crc32c: 0,
4398 fallback_active: false,
4399 };
4400 certificate.certificate_crc32c = certificate.computed_crc32c();
4401
4402 backend
4403 .persist_parallel_wal_commit_certificate(&cx, &certificate, 1, 1, true)
4404 .expect("persist certificate before WAL commit marker");
4405 assert_eq!(
4406 backend.inner.frame_count(),
4407 0,
4408 "certificate persistence must not itself expose a WAL commit marker"
4409 );
4410
4411 let certificate_path = std::path::Path::new("test.db-wal-cert");
4412 let (mut certificate_file, _) = vfs
4413 .open(
4414 &cx,
4415 Some(certificate_path),
4416 VfsOpenFlags::READONLY | VfsOpenFlags::WAL,
4417 )
4418 .expect("open certificate sidecar");
4419 let certificate_len = usize::try_from(
4420 certificate_file
4421 .file_size(&cx)
4422 .expect("certificate sidecar size"),
4423 )
4424 .expect("certificate sidecar size fits usize");
4425 let mut record_bytes = vec![0_u8; certificate_len];
4426 assert_eq!(
4427 certificate_file
4428 .read(&cx, &mut record_bytes, 0)
4429 .expect("read certificate sidecar"),
4430 certificate_len
4431 );
4432 certificate_file
4433 .close(&cx)
4434 .expect("close certificate sidecar");
4435 let reconstructed = ParallelWalDurableCertificateRecord::from_bytes(&record_bytes)
4436 .expect("reconstruct durable certificate record");
4437 assert_eq!(reconstructed.certificate, certificate);
4438 assert_eq!(reconstructed.wal_frame_start, 1);
4439 assert_eq!(reconstructed.wal_frame_end, 1);
4440 assert_eq!(
4441 reconstructed.wal_generation,
4442 backend.inner.inner().generation_identity()
4443 );
4444 assert!(
4445 !reconstructed.authorizes_wal_boundary(
4446 backend.inner.inner().generation_identity(),
4447 0,
4448 0,
4449 test_frame_payload_digest(1, &committed_page, 1),
4450 ),
4451 "orphan certificate must not authorize visibility before the matching commit marker"
4452 );
4453
4454 backend
4455 .append_frame(&cx, 1, &committed_page, 1)
4456 .expect("append matching WAL commit marker");
4457 backend.sync(&cx).expect("sync WAL commit marker");
4458 assert!(
4459 backend
4460 .inner
4461 .inner()
4462 .read_frame_header(&cx, 0)
4463 .expect("read matching WAL commit frame")
4464 .is_commit()
4465 );
4466 assert!(reconstructed.authorizes_wal_boundary(
4467 backend.inner.inner().generation_identity(),
4468 1,
4469 1,
4470 test_frame_payload_digest(1, &committed_page, 1),
4471 ));
4472
4473 let (mut certificate_file, _) = vfs
4474 .open(
4475 &cx,
4476 Some(certificate_path),
4477 VfsOpenFlags::READWRITE | VfsOpenFlags::WAL,
4478 )
4479 .expect("reopen certificate sidecar");
4480 let torn_offset = certificate_file
4481 .file_size(&cx)
4482 .expect("certificate sidecar size before torn tail");
4483 certificate_file
4484 .write(&cx, &[0xA5], torn_offset)
4485 .expect("append torn footer byte");
4486 certificate_file
4487 .close(&cx)
4488 .expect("close sidecar with torn tail");
4489 let recovered = backend
4490 .latest_authorized_parallel_wal_commit_certificate(&cx)
4491 .wait()
4492 .expect("torn certificate tail should recover the prior valid record")
4493 .expect("prior authorized certificate should remain discoverable");
4494 assert_eq!(recovered, certificate);
4495 }
4496
4497 #[test]
4498 fn content_mismatched_wal_interval_cannot_be_authorized_or_repaired() {
4499 let cx = test_cx();
4500 let vfs = MemoryVfs::new();
4501 let certified_page = sample_page(0x61);
4502 let actual_page = sample_page(0x62);
4503 let mut backend = make_path_refreshing_backend(&vfs, &cx);
4504 let mut certificate = sample_certificate(1, 1, vec![1]);
4505 certificate.wal_frame_payload_digest = test_frame_payload_digest(1, &certified_page, 1);
4506 certificate.certificate_crc32c = certificate.computed_crc32c();
4507
4508 backend
4509 .persist_parallel_wal_commit_certificate(&cx, &certificate, 1, 1, true)
4510 .expect("persist content-bound certificate");
4511 backend
4512 .append_frame(&cx, 1, &actual_page, 1)
4513 .expect("append differently valued commit frame");
4514 backend.sync(&cx).expect("sync mismatched commit frame");
4515
4516 let sidecar_before = read_certificate_sidecar(&vfs, &cx);
4517 assert!(
4518 backend
4519 .latest_authorized_parallel_wal_commit_certificate(&cx)
4520 .wait()
4521 .expect("content mismatch is a non-authorizing record")
4522 .is_none(),
4523 "matching generation and commit marker must not authorize different frame bytes"
4524 );
4525
4526 assert_wal_corrupt(
4527 backend
4528 .reconcile_parallel_wal_commit(&cx, &certificate, 1, 1, true)
4529 .wait(),
4530 "in-doubt content-bound reconciliation mismatch",
4531 );
4532 assert_eq!(
4533 read_certificate_sidecar(&vfs, &cx),
4534 sidecar_before,
4535 "digest mismatch must be diagnosed before sidecar repair"
4536 );
4537 assert_eq!(
4538 backend.inner.frame_count(),
4539 1,
4540 "digest mismatch must preserve the live WAL for diagnosis and retry"
4541 );
4542 }
4543
4544 #[test]
4545 fn absent_commit_marker_repairs_certificate_and_partial_wal_tail() {
4546 let cx = test_cx();
4547 let vfs = MemoryVfs::new();
4548 let mut backend = make_path_refreshing_backend(&vfs, &cx);
4549 let certificate = sample_certificate(1, 1, vec![1]);
4550 backend
4551 .persist_parallel_wal_commit_certificate(&cx, &certificate, 1, 1, true)
4552 .expect("persist orphan certificate");
4553
4554 let (mut tail_writer, _) = vfs
4555 .open(
4556 &cx,
4557 Some(std::path::Path::new("test.db-wal")),
4558 VfsOpenFlags::READWRITE | VfsOpenFlags::WAL,
4559 )
4560 .expect("open WAL for partial-tail injection");
4561 let committed_size = tail_writer.file_size(&cx).expect("read committed WAL size");
4562 tail_writer
4563 .write(&cx, &[0xA5; 7], committed_size)
4564 .expect("inject a partial physical frame");
4565 assert!(
4566 tail_writer.file_size(&cx).expect("read extended WAL size") > committed_size,
4567 "fault fixture must extend the physical WAL"
4568 );
4569 tail_writer.close(&cx).expect("close partial-tail injector");
4570
4571 assert_eq!(
4572 backend
4573 .reconcile_parallel_wal_commit(&cx, &certificate, 1, 1, true)
4574 .wait()
4575 .expect("missing commit marker must be exactly repairable"),
4576 ParallelWalCommitReconciliation::NotCommitted
4577 );
4578 assert!(
4579 read_certificate_sidecar(&vfs, &cx).is_empty(),
4580 "matching orphan certificate must be removed after NotCommitted proof"
4581 );
4582 let (mut repaired_wal, _) = vfs
4583 .open(
4584 &cx,
4585 Some(std::path::Path::new("test.db-wal")),
4586 VfsOpenFlags::READONLY | VfsOpenFlags::WAL,
4587 )
4588 .expect("open repaired WAL");
4589 assert_eq!(
4590 repaired_wal.file_size(&cx).expect("read repaired WAL size"),
4591 committed_size,
4592 "NotCommitted reconciliation must truncate the physical partial tail"
4593 );
4594 repaired_wal.close(&cx).expect("close repaired WAL");
4595 }
4596
4597 #[test]
4598 fn durable_certificate_recovery_accepts_every_truncated_record_prefix() {
4599 let cx = test_cx();
4600 let vfs = MemoryVfs::new();
4601 let (mut backend, authorized) = make_authorized_certificate_backend(&vfs, &cx);
4602 let authorized_bytes = read_certificate_sidecar(&vfs, &cx);
4603 let orphan = sample_certificate(2, 2, vec![1]);
4604 let orphan_bytes = ParallelWalDurableCertificateRecord::new(
4605 backend.inner.inner().generation_identity(),
4606 2,
4607 2,
4608 orphan,
4609 )
4610 .expect("construct orphan record")
4611 .to_bytes();
4612
4613 for prefix_len in 1..orphan_bytes.len() {
4614 let mut sidecar = authorized_bytes.clone();
4615 sidecar.extend_from_slice(&orphan_bytes[..prefix_len]);
4616 replace_certificate_sidecar(&vfs, &cx, &sidecar);
4617 let recovered_result = backend
4618 .latest_authorized_parallel_wal_commit_certificate(&cx)
4619 .wait();
4620 assert!(
4621 recovered_result.is_ok(),
4622 "truncated certificate prefix of {prefix_len} bytes must recover: {recovered_result:?}"
4623 );
4624 let recovered = recovered_result
4625 .expect("truncated certificate recovery was asserted successful")
4626 .expect("authorized record must remain discoverable");
4627 assert_eq!(recovered, authorized, "failed at prefix {prefix_len}");
4628 }
4629 }
4630
4631 #[test]
4632 fn durable_certificate_append_repairs_the_accepted_torn_suffix() {
4633 let cx = test_cx();
4634 let vfs = MemoryVfs::new();
4635 let (mut backend, authorized) = make_authorized_certificate_backend(&vfs, &cx);
4636 let authorized_bytes = read_certificate_sidecar(&vfs, &cx);
4637 let orphan = sample_certificate(2, 2, vec![1]);
4638 let orphan_bytes = ParallelWalDurableCertificateRecord::new(
4639 backend.inner.inner().generation_identity(),
4640 2,
4641 2,
4642 orphan.clone(),
4643 )
4644 .expect("construct orphan record")
4645 .to_bytes();
4646 for prefix_len in 1..orphan_bytes.len() {
4647 let mut torn_sidecar = authorized_bytes.clone();
4648 torn_sidecar.extend_from_slice(&orphan_bytes[..prefix_len]);
4649 replace_certificate_sidecar(&vfs, &cx, &torn_sidecar);
4650
4651 assert_eq!(
4652 backend
4653 .latest_authorized_parallel_wal_commit_certificate(&cx)
4654 .wait()
4655 .expect("one torn suffix should recover")
4656 .expect("authorized predecessor remains visible"),
4657 authorized,
4658 "read recovery failed for prefix {prefix_len}"
4659 );
4660
4661 backend
4662 .persist_parallel_wal_commit_certificate(&cx, &orphan, 2, 2, true)
4663 .expect("next append repairs the torn suffix first");
4664 let repaired_sidecar = read_certificate_sidecar(&vfs, &cx);
4665 assert_eq!(
4666 repaired_sidecar.len(),
4667 authorized_bytes.len() + orphan_bytes.len(),
4668 "replacement record did not start at the prior complete boundary for prefix {prefix_len}"
4669 );
4670 assert_eq!(
4671 backend
4672 .latest_authorized_parallel_wal_commit_certificate(&cx)
4673 .wait()
4674 .expect("orphan lookback crosses the repaired boundary")
4675 .expect("authorized predecessor remains discoverable"),
4676 authorized,
4677 "orphan lookback failed after repairing prefix {prefix_len}"
4678 );
4679 }
4680
4681 let mut corrupt_record = orphan_bytes;
4682 let envelope_crc_offset =
4683 corrupt_record.len() - ParallelWalDurableCertificateRecord::LENGTH_FOOTER_SIZE - 4;
4684 corrupt_record[envelope_crc_offset] ^= 0x80;
4685 let mut corrupt_sidecar = authorized_bytes;
4686 corrupt_sidecar.extend_from_slice(&corrupt_record);
4687 replace_certificate_sidecar(&vfs, &cx, &corrupt_sidecar);
4688 assert_wal_corrupt(
4689 backend
4690 .persist_parallel_wal_commit_certificate(&cx, &orphan, 2, 2, true)
4691 .wait(),
4692 "append-time complete record corruption",
4693 );
4694 }
4695
4696 #[test]
4697 fn durable_certificate_recovery_rejects_complete_corruption_and_garbage() {
4698 let cx = test_cx();
4699 let vfs = MemoryVfs::new();
4700 let (mut backend, _) = make_authorized_certificate_backend(&vfs, &cx);
4701 let authorized_bytes = read_certificate_sidecar(&vfs, &cx);
4702 let orphan = sample_certificate(2, 2, vec![1]);
4703 let orphan_bytes = ParallelWalDurableCertificateRecord::new(
4704 backend.inner.inner().generation_identity(),
4705 2,
4706 2,
4707 orphan,
4708 )
4709 .expect("construct orphan record")
4710 .to_bytes();
4711
4712 let mut bad_crc = orphan_bytes.clone();
4713 let envelope_crc_offset =
4714 bad_crc.len() - ParallelWalDurableCertificateRecord::LENGTH_FOOTER_SIZE - 4;
4715 bad_crc[envelope_crc_offset] ^= 0x80;
4716 let mut sidecar = authorized_bytes.clone();
4717 sidecar.extend_from_slice(&bad_crc);
4718 replace_certificate_sidecar(&vfs, &cx, &sidecar);
4719 assert_wal_corrupt(
4720 backend
4721 .latest_authorized_parallel_wal_commit_certificate(&cx)
4722 .wait(),
4723 "complete record with bad CRC",
4724 );
4725
4726 let mut bad_version = orphan_bytes.clone();
4727 bad_version[8] ^= 0x01;
4728 let mut sidecar = authorized_bytes.clone();
4729 sidecar.extend_from_slice(&bad_version);
4730 replace_certificate_sidecar(&vfs, &cx, &sidecar);
4731 assert_wal_corrupt(
4732 backend
4733 .latest_authorized_parallel_wal_commit_certificate(&cx)
4734 .wait(),
4735 "complete record with bad version",
4736 );
4737
4738 let mut bad_magic = orphan_bytes.clone();
4739 bad_magic[0] ^= 0x01;
4740 let mut sidecar = authorized_bytes.clone();
4741 sidecar.extend_from_slice(&bad_magic);
4742 replace_certificate_sidecar(&vfs, &cx, &sidecar);
4743 assert_wal_corrupt(
4744 backend
4745 .latest_authorized_parallel_wal_commit_certificate(&cx)
4746 .wait(),
4747 "complete record with bad magic",
4748 );
4749
4750 let mut bad_footer = orphan_bytes;
4751 let footer_offset =
4752 bad_footer.len() - ParallelWalDurableCertificateRecord::LENGTH_FOOTER_SIZE;
4753 bad_footer[footer_offset] ^= 0x80;
4754 let mut sidecar = authorized_bytes;
4755 sidecar.extend_from_slice(&bad_footer);
4756 replace_certificate_sidecar(&vfs, &cx, &sidecar);
4757 assert_wal_corrupt(
4758 backend
4759 .latest_authorized_parallel_wal_commit_certificate(&cx)
4760 .wait(),
4761 "complete record with bad footer",
4762 );
4763
4764 let garbage_vfs = MemoryVfs::new();
4765 let mut garbage_backend = make_path_refreshing_backend(&garbage_vfs, &cx);
4766 replace_certificate_sidecar(&garbage_vfs, &cx, &[0xA5; 128]);
4767 assert_wal_corrupt(
4768 garbage_backend
4769 .latest_authorized_parallel_wal_commit_certificate(&cx)
4770 .wait(),
4771 "nonempty garbage sidecar",
4772 );
4773
4774 let mut fake_magic = vec![0_u8; MIN_DURABLE_CERTIFICATE_RECORD_SIZE];
4775 fake_magic[..PARALLEL_WAL_DURABLE_CERTIFICATE_MAGIC.len()]
4776 .copy_from_slice(&PARALLEL_WAL_DURABLE_CERTIFICATE_MAGIC);
4777 fake_magic[8..10].copy_from_slice(
4778 &fsqlite_wal::PARALLEL_WAL_DURABLE_CERTIFICATE_RECORD_VERSION.to_le_bytes(),
4779 );
4780 let fake_record_len = u32::try_from(fake_magic.len()).expect("fake record length fits u32");
4781 fake_magic[10..14].copy_from_slice(&fake_record_len.to_le_bytes());
4782 let fake_footer_offset =
4783 fake_magic.len() - ParallelWalDurableCertificateRecord::LENGTH_FOOTER_SIZE;
4784 fake_magic[fake_footer_offset..].copy_from_slice(&fake_record_len.to_le_bytes());
4785 replace_certificate_sidecar(&garbage_vfs, &cx, &fake_magic);
4786 assert_wal_corrupt(
4787 garbage_backend
4788 .latest_authorized_parallel_wal_commit_certificate(&cx)
4789 .wait(),
4790 "fake magic and length without a valid envelope",
4791 );
4792 }
4793
4794 #[test]
4795 fn durable_certificate_maximum_size_is_shared_by_writer_and_reader() {
4796 let cx = test_cx();
4797 let vfs = MemoryVfs::new();
4798 let mut backend = make_path_refreshing_backend(&vfs, &cx);
4799 let certificate = sample_certificate(1, 1, vec![1; usize::from(u16::MAX)]);
4800 let record = ParallelWalDurableCertificateRecord::new(
4801 backend.inner.inner().generation_identity(),
4802 1,
4803 1,
4804 certificate.clone(),
4805 )
4806 .expect("construct maximum-size record");
4807 assert_eq!(
4808 record.to_bytes().len(),
4809 PARALLEL_WAL_MAX_DURABLE_CERTIFICATE_RECORD_SIZE
4810 );
4811 backend
4812 .persist_parallel_wal_commit_certificate(&cx, &certificate, 1, 1, true)
4813 .expect("writer accepts maximum-size record");
4814 assert!(
4815 backend
4816 .latest_authorized_parallel_wal_commit_certificate(&cx)
4817 .wait()
4818 .expect("reader accepts maximum-size record")
4819 .is_none(),
4820 "record remains unauthorized until its WAL commit marker exists"
4821 );
4822 }
4823
4824 #[test]
4825 fn durable_certificate_orphan_lookback_allows_exact_boundary_plus_torn_tail() {
4826 let cx = test_cx();
4827 let vfs = MemoryVfs::new();
4828 let (mut backend, authorized) = make_authorized_certificate_backend(&vfs, &cx);
4829 let mut sidecar = read_certificate_sidecar(&vfs, &cx);
4830 for orphan_index in 0..MAX_ORPHAN_CERTIFICATE_LOOKBACK {
4831 let epoch = u64::try_from(orphan_index).expect("orphan index fits u64") + 2;
4832 let orphan = sample_certificate(epoch, epoch, vec![1]);
4833 sidecar.extend_from_slice(
4834 &ParallelWalDurableCertificateRecord::new(
4835 backend.inner.inner().generation_identity(),
4836 2,
4837 2,
4838 orphan,
4839 )
4840 .expect("construct bounded orphan")
4841 .to_bytes(),
4842 );
4843 }
4844 sidecar.push(0xA5);
4845 replace_certificate_sidecar(&vfs, &cx, &sidecar);
4846 assert_eq!(
4847 backend
4848 .latest_authorized_parallel_wal_commit_certificate(&cx)
4849 .wait()
4850 .expect("64 orphans plus one torn suffix remain within bound")
4851 .expect("authorized predecessor is found"),
4852 authorized
4853 );
4854
4855 sidecar.pop();
4856 let overflow_epoch =
4857 u64::try_from(MAX_ORPHAN_CERTIFICATE_LOOKBACK).expect("lookback fits u64") + 2;
4858 let overflow = sample_certificate(overflow_epoch, overflow_epoch, vec![1]);
4859 sidecar.extend_from_slice(
4860 &ParallelWalDurableCertificateRecord::new(
4861 backend.inner.inner().generation_identity(),
4862 2,
4863 2,
4864 overflow,
4865 )
4866 .expect("construct overflow orphan")
4867 .to_bytes(),
4868 );
4869 replace_certificate_sidecar(&vfs, &cx, &sidecar);
4870 assert_wal_corrupt(
4871 backend
4872 .latest_authorized_parallel_wal_commit_certificate(&cx)
4873 .wait(),
4874 "65 unauthorized records",
4875 );
4876 }
4877
4878 #[test]
4879 fn certificate_and_handoff_fences_request_full_durability() {
4880 let cx = test_cx();
4881 let vfs = CheckpointHandoffFaultVfs::new();
4882 let (mut backend, certificate, _) = make_checkpoint_handoff_fault_backend(&vfs, &cx);
4883
4884 assert_eq!(
4885 vfs.take_sync_observations(),
4886 vec![CertificateSyncObservation::Durable(
4887 PathBuf::from(CERTIFICATE_PATH),
4888 SyncKind::FullDurable,
4889 )],
4890 "certificate append must use the strongest durability intent"
4891 );
4892
4893 assert_eq!(
4894 backend
4895 .reconcile_parallel_wal_commit(&cx, &certificate, 1, 1, true)
4896 .wait()
4897 .expect("reconcile committed certificate"),
4898 ParallelWalCommitReconciliation::Authorized
4899 );
4900 assert_eq!(
4901 vfs.take_sync_observations(),
4902 vec![CertificateSyncObservation::Durable(
4903 PathBuf::from(CERTIFICATE_PATH),
4904 SyncKind::FullDurable,
4905 )],
4906 "certificate reconciliation must preserve full durability intent"
4907 );
4908
4909 let record = backend
4910 .latest_authorized_durable_certificate_record(&cx)
4911 .wait()
4912 .expect("read authorized certificate record")
4913 .expect("authorized certificate record must exist");
4914 backend
4915 .persist_checkpoint_certificate_handoff(&cx, &record)
4916 .wait()
4917 .expect("persist checkpoint certificate handoff");
4918 assert_eq!(
4919 vfs.take_sync_observations(),
4920 vec![CertificateSyncObservation::Durable(
4921 PathBuf::from(CHECKPOINT_HANDOFF_PATH),
4922 SyncKind::FullDurable,
4923 )],
4924 "checkpoint handoff must use the strongest durability intent"
4925 );
4926 }
4927
4928 #[test]
4929 fn checkpoint_handoff_write_failure_preserves_authoritative_wal_generation() {
4930 let cx = test_cx();
4931 let vfs = CheckpointHandoffFaultVfs::new();
4932 let (mut backend, certificate, committed_page) =
4933 make_checkpoint_handoff_fault_backend(&vfs, &cx);
4934 let before = capture_authoritative_wal(&backend, &vfs, &cx, certificate, committed_page);
4935 vfs.fail_next_handoff_write();
4936
4937 let mut checkpoint_writer = MockCheckpointPageWriter;
4938 let error = backend
4939 .checkpoint(
4940 &cx,
4941 CheckpointMode::Truncate,
4942 &mut checkpoint_writer,
4943 0,
4944 None,
4945 )
4946 .expect_err("checkpoint must fail before reset when the handoff write fails");
4947 assert!(
4948 error
4949 .to_string()
4950 .contains("injected checkpoint handoff write failure"),
4951 "unexpected handoff write error: {error}"
4952 );
4953 assert_authoritative_wal_unchanged(&mut backend, &vfs, &cx, &before);
4954 }
4955
4956 #[test]
4957 fn checkpoint_handoff_durable_sync_failure_preserves_authoritative_wal_generation() {
4958 let cx = test_cx();
4959 let vfs = CheckpointHandoffFaultVfs::new();
4960 let (mut backend, certificate, committed_page) =
4961 make_checkpoint_handoff_fault_backend(&vfs, &cx);
4962 let before = capture_authoritative_wal(&backend, &vfs, &cx, certificate, committed_page);
4963 vfs.fail_next_handoff_sync();
4964
4965 let mut checkpoint_writer = MockCheckpointPageWriter;
4966 let error = backend
4967 .checkpoint(
4968 &cx,
4969 CheckpointMode::Truncate,
4970 &mut checkpoint_writer,
4971 0,
4972 None,
4973 )
4974 .expect_err("checkpoint must fail before reset when the handoff sync fails");
4975 assert!(
4976 error
4977 .to_string()
4978 .contains("injected checkpoint handoff durable-sync failure"),
4979 "unexpected handoff durable-sync error: {error}"
4980 );
4981 assert_authoritative_wal_unchanged(&mut backend, &vfs, &cx, &before);
4982 }
4983
4984 #[test]
4985 fn dropping_pending_checkpoint_handoff_write_preserves_authoritative_wal_generation() {
4986 let cx = test_cx();
4987 let vfs = CheckpointHandoffFaultVfs::new();
4988 let (mut backend, certificate, committed_page) =
4989 make_checkpoint_handoff_fault_backend(&vfs, &cx);
4990 let before = capture_authoritative_wal(&backend, &vfs, &cx, certificate, committed_page);
4991 vfs.pend_next_handoff_write();
4992
4993 let mut checkpoint_writer = MockCheckpointPageWriter;
4994 let reached_pending_handoff = {
4995 let mut checkpoint = backend.checkpoint(
4996 &cx,
4997 CheckpointMode::Truncate,
4998 &mut checkpoint_writer,
4999 0,
5000 None,
5001 );
5002 let mut task_cx = std::task::Context::from_waker(std::task::Waker::noop());
5003 matches!(
5004 std::future::Future::poll(checkpoint.as_mut(), &mut task_cx),
5005 std::task::Poll::Pending
5006 )
5007 };
5008 assert!(
5009 reached_pending_handoff,
5010 "checkpoint should remain pending inside the injected handoff write"
5011 );
5012 assert_authoritative_wal_unchanged(&mut backend, &vfs, &cx, &before);
5013 }
5014
5015 #[test]
5016 fn two_backend_instances_continue_authorized_certificate_clocks() {
5017 let cx = test_cx();
5018 let vfs = MemoryVfs::new();
5019 let wal = WalFile::create(&cx, open_wal_file(&vfs, &cx), PAGE_SIZE, 0, test_salts())
5020 .expect("create shared WAL");
5021 let mut first_backend = PathRefreshingWalBackend::new(
5022 vfs.clone(),
5023 std::path::Path::new("test.db"),
5024 std::path::Path::new("test.db-wal"),
5025 PAGE_SIZE,
5026 wal,
5027 true,
5028 #[cfg(all(feature = "native", any(unix, windows)))]
5029 None,
5030 );
5031 let request =
5032 |batch_id, wal_frame_payload_digest| fsqlite_wal::ParallelWalDurabilityRequest {
5033 trace_id: batch_id,
5034 scenario_id: "two-instance-continuity".to_owned(),
5035 certificate_epoch: 0,
5036 durable_segment_epoch: 0,
5037 batch_size: 1,
5038 batch_ids: vec![batch_id],
5039 lane_record_counts: vec![1],
5040 db_size_pages: 1,
5041 page_set_size: 1,
5042 control_mode: fsqlite_wal::ParallelWalOperatingMode::Auto,
5043 fallback_reason: None,
5044 checkpoint_active: false,
5045 wal_frame_payload_digest,
5046 };
5047
5048 let first_combiner = fsqlite_wal::ParallelWalDurabilityCombiner::default();
5049 let first_page = sample_page(0x51);
5050 let first_receipt = first_combiner
5051 .certify_and_publish(
5052 request(1, test_frame_payload_digest(1, &first_page, 1)),
5053 |certificate| {
5054 first_backend
5055 .persist_parallel_wal_commit_certificate(&cx, certificate, 1, 1, true)
5056 .wait()
5057 .and_then(|()| first_backend.append_frame(&cx, 1, &first_page, 1).wait())
5058 .and_then(|()| first_backend.sync(&cx))
5059 .map_err(|error| error.to_string())
5060 },
5061 )
5062 .expect("first backend publishes certificate");
5063
5064 let second_wal =
5065 WalFile::open(&cx, open_wal_file(&vfs, &cx)).expect("second backend opens shared WAL");
5066 let mut second_backend = PathRefreshingWalBackend::new(
5067 vfs.clone(),
5068 std::path::Path::new("test.db"),
5069 std::path::Path::new("test.db-wal"),
5070 PAGE_SIZE,
5071 second_wal,
5072 true,
5073 #[cfg(all(feature = "native", any(unix, windows)))]
5074 None,
5075 );
5076
5077 let orphan_combiner = fsqlite_wal::ParallelWalDurabilityCombiner::default();
5081 orphan_combiner
5082 .reconcile_authorized_seed(&first_receipt.certificate)
5083 .expect("seed orphan-producing process");
5084 let orphan_receipt = orphan_combiner
5085 .certify_and_publish(
5086 request(99, test_frame_payload_digest(1, &sample_page(0x52), 1)),
5087 |_| Ok(()),
5088 )
5089 .expect("construct deterministic orphan certificate");
5090 second_backend
5091 .persist_parallel_wal_commit_certificate(&cx, &orphan_receipt.certificate, 2, 2, true)
5092 .expect("persist well-formed orphan certificate tail");
5093 let authorized_seed = second_backend
5094 .latest_authorized_parallel_wal_commit_certificate(&cx)
5095 .expect("second backend performs bounded orphan lookback")
5096 .expect("preceding first certificate remains authorized");
5097 assert_eq!(authorized_seed, first_receipt.certificate);
5098
5099 let second_combiner = fsqlite_wal::ParallelWalDurabilityCombiner::default();
5100 second_combiner
5101 .reconcile_authorized_seed(&authorized_seed)
5102 .expect("seed second process-local combiner");
5103 let second_page = sample_page(0x52);
5104 let second_receipt = second_combiner
5105 .certify_and_publish(
5106 request(2, test_frame_payload_digest(1, &second_page, 1)),
5107 |certificate| {
5108 second_backend
5109 .persist_parallel_wal_commit_certificate(&cx, certificate, 2, 2, true)
5110 .wait()
5111 .and_then(|()| second_backend.append_frame(&cx, 1, &second_page, 1).wait())
5112 .and_then(|()| second_backend.sync(&cx))
5113 .map_err(|error| error.to_string())
5114 },
5115 )
5116 .expect("second backend publishes certificate");
5117
5118 assert_eq!(
5119 second_receipt.certificate.commit_seq_lo.get(),
5120 first_receipt.certificate.commit_seq_hi.get() + 1
5121 );
5122 assert_eq!(
5123 second_receipt.certificate.certificate_epoch,
5124 first_receipt.certificate.certificate_epoch + 1
5125 );
5126 assert_eq!(
5127 second_receipt.certificate, orphan_receipt.certificate,
5128 "continuation may reuse an orphan identity but must not overlap any authorized certificate"
5129 );
5130 let latest = second_backend
5131 .latest_authorized_parallel_wal_commit_certificate(&cx)
5132 .expect("read second bounded authorized tail")
5133 .expect("second certificate is authorized");
5134 assert_eq!(latest, second_receipt.certificate);
5135
5136 let generation_before_checkpoint = second_backend.inner.inner().generation_identity();
5137 let mut checkpoint_writer = MockCheckpointPageWriter;
5138 let checkpoint = second_backend
5139 .checkpoint(
5140 &cx,
5141 CheckpointMode::Truncate,
5142 &mut checkpoint_writer,
5143 0,
5144 None,
5145 )
5146 .expect("truncate checkpoint records certificate clock handoff");
5147 assert!(checkpoint.wal_was_reset);
5148 assert_ne!(
5149 second_backend.inner.inner().generation_identity(),
5150 generation_before_checkpoint
5151 );
5152 let checkpoint_seed = second_backend
5153 .latest_authorized_parallel_wal_commit_certificate(&cx)
5154 .expect("read checkpoint certificate clock handoff")
5155 .expect("reset generation retains the last consumed certificate clock");
5156 assert_eq!(checkpoint_seed, second_receipt.certificate);
5157
5158 let post_checkpoint_combiner = fsqlite_wal::ParallelWalDurabilityCombiner::default();
5159 post_checkpoint_combiner
5160 .reconcile_authorized_seed(&checkpoint_seed)
5161 .expect("seed fresh post-checkpoint combiner");
5162 let post_checkpoint_page = sample_page(0x53);
5163 let post_checkpoint_receipt = post_checkpoint_combiner
5164 .certify_and_publish(
5165 request(3, test_frame_payload_digest(1, &post_checkpoint_page, 1)),
5166 |certificate| {
5167 second_backend
5168 .persist_parallel_wal_commit_certificate(&cx, certificate, 1, 1, true)
5169 .wait()
5170 .and_then(|()| {
5171 second_backend
5172 .append_frame(&cx, 1, &post_checkpoint_page, 1)
5173 .wait()
5174 })
5175 .and_then(|()| second_backend.sync(&cx))
5176 .map_err(|error| error.to_string())
5177 },
5178 )
5179 .expect("publish first certificate in reset WAL generation");
5180 assert_eq!(
5181 post_checkpoint_receipt.certificate.commit_seq_lo.get(),
5182 second_receipt.certificate.commit_seq_hi.get() + 1
5183 );
5184 assert_eq!(
5185 post_checkpoint_receipt.certificate.certificate_epoch,
5186 second_receipt.certificate.certificate_epoch + 1
5187 );
5188 assert_eq!(
5189 second_backend
5190 .latest_authorized_parallel_wal_commit_certificate(&cx)
5191 .expect("read post-checkpoint current-generation certificate")
5192 .expect("post-checkpoint certificate is authorized"),
5193 post_checkpoint_receipt.certificate
5194 );
5195 }
5196
5197 fn open_wal_file(vfs: &MemoryVfs, cx: &Cx) -> <MemoryVfs as Vfs>::File {
5198 let flags = VfsOpenFlags::READWRITE | VfsOpenFlags::CREATE | VfsOpenFlags::WAL;
5199 let (file, _) = vfs
5200 .open(cx, Some(std::path::Path::new("test.db-wal")), flags)
5201 .expect("open WAL file");
5202 file
5203 }
5204
5205 fn make_adapter(vfs: &MemoryVfs, cx: &Cx) -> WalBackendAdapter<<MemoryVfs as Vfs>::File> {
5206 let file = open_wal_file(vfs, cx);
5207 let wal = WalFile::create(cx, file, PAGE_SIZE, 0, test_salts()).expect("create WAL");
5208 WalBackendAdapter::new(wal)
5209 }
5210
5211 fn make_fault_adapter(
5213 vfs: &CheckpointHandoffFaultVfs,
5214 cx: &Cx,
5215 ) -> WalBackendAdapter<<CheckpointHandoffFaultVfs as Vfs>::File> {
5216 let flags = VfsOpenFlags::READWRITE | VfsOpenFlags::CREATE | VfsOpenFlags::WAL;
5217 let (file, _) = vfs
5218 .open(cx, Some(std::path::Path::new("test.db-wal")), flags)
5219 .expect("open fault WAL file");
5220 let wal = WalFile::create(cx, file, PAGE_SIZE, 0, test_salts()).expect("create fault WAL");
5221 WalBackendAdapter::new(wal)
5222 }
5223
5224 #[test]
5227 fn test_adapter_append_and_frame_count() {
5228 let cx = test_cx();
5229 let vfs = MemoryVfs::new();
5230 let mut adapter = make_adapter(&vfs, &cx);
5231
5232 assert_eq!(adapter.frame_count(), 0);
5233
5234 let page = sample_page(0x42);
5235 adapter
5236 .append_frame(&cx, 1, &page, 0)
5237 .expect("append frame");
5238 assert_eq!(adapter.frame_count(), 1);
5239
5240 adapter
5241 .append_frame(&cx, 2, &sample_page(0x43), 2)
5242 .expect("append commit frame");
5243 assert_eq!(adapter.frame_count(), 2);
5244 }
5245
5246 #[test]
5247 fn test_adapter_read_page_found() {
5248 let cx = test_cx();
5249 let vfs = MemoryVfs::new();
5250 let mut adapter = make_adapter(&vfs, &cx);
5251
5252 let page1 = sample_page(0x10);
5253 let page2 = sample_page(0x20);
5254 adapter.append_frame(&cx, 1, &page1, 0).expect("append");
5255 adapter
5256 .append_frame(&cx, 2, &page2, 2)
5257 .expect("append commit");
5258
5259 let result = adapter.read_page(&cx, 1).expect("read page 1");
5260 assert_eq!(result, Some(page1));
5261
5262 let result = adapter.read_page(&cx, 2).expect("read page 2");
5263 assert_eq!(result, Some(page2));
5264 }
5265
5266 #[test]
5267 fn test_adapter_read_page_not_found() {
5268 let cx = test_cx();
5269 let vfs = MemoryVfs::new();
5270 let mut adapter = make_adapter(&vfs, &cx);
5271
5272 adapter
5273 .append_frame(&cx, 1, &sample_page(0x10), 1)
5274 .expect("append");
5275
5276 let result = adapter.read_page(&cx, 99).expect("read missing page");
5277 assert_eq!(result, None);
5278 }
5279
5280 #[test]
5281 fn test_adapter_read_page_returns_latest_version() {
5282 let cx = test_cx();
5283 let vfs = MemoryVfs::new();
5284 let mut adapter = make_adapter(&vfs, &cx);
5285
5286 let old_data = sample_page(0xAA);
5287 let new_data = sample_page(0xBB);
5288
5289 adapter
5291 .append_frame(&cx, 5, &old_data, 0)
5292 .expect("append old");
5293 adapter
5294 .append_frame(&cx, 5, &new_data, 1)
5295 .expect("append new (commit)");
5296
5297 let result = adapter.read_page(&cx, 5).expect("read page 5");
5298 assert_eq!(
5299 result,
5300 Some(new_data),
5301 "adapter should return the latest WAL version"
5302 );
5303 }
5304
5305 #[test]
5306 fn test_adapter_refreshes_cross_handle_visibility_and_append_position() {
5307 let cx = test_cx();
5308 let vfs = MemoryVfs::new();
5309
5310 let file1 = open_wal_file(&vfs, &cx);
5311 let wal1 = WalFile::create(&cx, file1, PAGE_SIZE, 0, test_salts()).expect("create WAL");
5312 let mut adapter1 = WalBackendAdapter::new(wal1);
5313
5314 let file2 = open_wal_file(&vfs, &cx);
5315 let wal2 = WalFile::open(&cx, file2).expect("open WAL");
5316 let mut adapter2 = WalBackendAdapter::new(wal2);
5317
5318 let page1 = sample_page(0x11);
5319 adapter1
5320 .append_frame(&cx, 1, &page1, 1)
5321 .expect("adapter1 append commit");
5322 adapter1.sync(&cx).expect("adapter1 sync");
5323 adapter2
5324 .begin_transaction(&cx)
5325 .expect("adapter2 begin transaction");
5326 assert_eq!(
5327 adapter2.read_page(&cx, 1).expect("adapter2 read page1"),
5328 Some(page1.clone()),
5329 "adapter2 should observe adapter1 commit at transaction begin"
5330 );
5331
5332 let page2 = sample_page(0x22);
5333 adapter2
5334 .append_frame(&cx, 2, &page2, 2)
5335 .expect("adapter2 append commit");
5336 adapter2.sync(&cx).expect("adapter2 sync");
5337 adapter1
5338 .begin_transaction(&cx)
5339 .expect("adapter1 begin transaction");
5340 assert_eq!(
5341 adapter1.read_page(&cx, 2).expect("adapter1 read page2"),
5342 Some(page2.clone()),
5343 "adapter1 should observe adapter2 commit at transaction begin"
5344 );
5345
5346 assert_eq!(
5348 adapter1.frame_count(),
5349 2,
5350 "shared WAL should contain both commit frames"
5351 );
5352 assert_eq!(
5353 adapter2.frame_count(),
5354 2,
5355 "shared WAL should contain both commit frames"
5356 );
5357 }
5358
5359 #[test]
5360 fn test_path_refresh_rejects_replacement_wal_page_size_mismatch() {
5361 let cx = test_cx();
5362 let vfs = MemoryVfs::new();
5363 let wal_path = std::path::Path::new("test.db-wal");
5364
5365 let file = open_wal_file(&vfs, &cx);
5366 let wal = WalFile::create(&cx, file, PAGE_SIZE, 0, test_salts()).expect("create WAL");
5367 let mut backend = PathRefreshingWalBackend::new(
5368 vfs.clone(),
5369 std::path::Path::new("test.db"),
5370 wal_path,
5371 PAGE_SIZE,
5372 wal,
5373 true,
5374 #[cfg(all(feature = "native", any(unix, windows)))]
5375 None,
5376 );
5377
5378 backend
5379 .append_frame(&cx, 1, &sample_page(0x31), 1)
5380 .expect("append through live backend");
5381 backend.sync(&cx).expect("sync live backend");
5382
5383 vfs.delete(&cx, wal_path, false)
5384 .expect("remove path-visible WAL");
5385 let flags = VfsOpenFlags::READWRITE | VfsOpenFlags::CREATE | VfsOpenFlags::WAL;
5386 let (replacement_file, _) = vfs
5387 .open(&cx, Some(wal_path), flags)
5388 .expect("open replacement WAL path");
5389 let replacement_page_size = PAGE_SIZE
5390 .checked_mul(2)
5391 .expect("test replacement page size fits u32");
5392 let replacement_wal = WalFile::create(
5393 &cx,
5394 replacement_file,
5395 replacement_page_size,
5396 0,
5397 test_salts(),
5398 )
5399 .expect("create mismatched replacement WAL");
5400 replacement_wal.close(&cx).expect("close replacement WAL");
5401
5402 let err = backend
5403 .begin_transaction(&cx)
5404 .expect_err("path refresh should reject mismatched WAL page size");
5405 assert!(
5406 matches!(
5407 err,
5408 FrankenError::WalCorrupt { ref detail }
5409 if detail.contains("does not match database page size")
5410 && detail.contains("during path refresh")
5411 ),
5412 "unexpected error: {err:?}"
5413 );
5414 }
5415
5416 #[test]
5417 fn test_generation_change_allows_identical_full_page_baseline() {
5418 let cx = test_cx();
5419 let vfs = MemoryVfs::new();
5420 let (mut backend, snapshot, page_two) = make_generation_transition_backend(&vfs, &cx);
5421 let baseline = TransactionConflictPageBaseline {
5422 page_number: 2,
5423 page_hash: *blake3::hash(&page_two).as_bytes(),
5424 };
5425
5426 let conflicts = backend
5427 .conflicting_pages_since_snapshot(&cx, snapshot, &[2], &[baseline])
5428 .expect("validate checkpoint-only generation transition");
5429 assert!(
5430 conflicts.is_empty(),
5431 "byte-identical checkpoint-only reset must not create a false conflict"
5432 );
5433 }
5434
5435 #[test]
5436 fn test_generation_change_rejects_changed_candidate_page() {
5437 let cx = test_cx();
5438 let vfs = MemoryVfs::new();
5439 let (mut backend, snapshot, page_two) = make_generation_transition_backend(&vfs, &cx);
5440 let changed_page_two = sample_page(0x33);
5441 write_main_db_pages(
5442 &vfs,
5443 &cx,
5444 &[
5445 sqlite_page_one(u16::try_from(PAGE_SIZE).expect("page size fits u16")),
5446 changed_page_two,
5447 ],
5448 );
5449 let baseline = TransactionConflictPageBaseline {
5450 page_number: 2,
5451 page_hash: *blake3::hash(&page_two).as_bytes(),
5452 };
5453
5454 let conflicts = backend
5455 .conflicting_pages_since_snapshot(&cx, snapshot, &[2], &[baseline])
5456 .expect("validate changed page across generation transition");
5457 assert_eq!(conflicts, vec![2]);
5458 }
5459
5460 #[test]
5461 fn test_generation_change_rejects_changed_candidate_from_replacement_wal() {
5462 let cx = test_cx();
5463 let vfs = MemoryVfs::new();
5464 let (mut backend, snapshot, page_two) = make_generation_transition_backend(&vfs, &cx);
5465 append_replacement_wal_page(&vfs, &cx, 2, &sample_page(0x44), 2);
5466 let baseline = TransactionConflictPageBaseline {
5467 page_number: 2,
5468 page_hash: *blake3::hash(&page_two).as_bytes(),
5469 };
5470
5471 let conflicts = backend
5472 .conflicting_pages_since_snapshot(&cx, snapshot, &[2], &[baseline])
5473 .expect("replacement WAL page must take precedence over identical main page");
5474 assert_eq!(conflicts, vec![2]);
5475 }
5476
5477 #[test]
5478 fn test_generation_change_rejects_missing_baseline() {
5479 let cx = test_cx();
5480 let vfs = MemoryVfs::new();
5481 let (mut backend, snapshot, _) = make_generation_transition_backend(&vfs, &cx);
5482
5483 let conflicts = backend
5484 .conflicting_pages_since_snapshot(&cx, snapshot, &[2], &[])
5485 .expect("missing baseline must fail closed");
5486 assert_eq!(conflicts, vec![2]);
5487 }
5488
5489 #[test]
5490 fn test_generation_change_rejects_conflicting_duplicate_baselines() {
5491 let cx = test_cx();
5492 let vfs = MemoryVfs::new();
5493 let (mut backend, snapshot, page_two) = make_generation_transition_backend(&vfs, &cx);
5494 let baselines = [
5495 TransactionConflictPageBaseline {
5496 page_number: 2,
5497 page_hash: *blake3::hash(&page_two).as_bytes(),
5498 },
5499 TransactionConflictPageBaseline {
5500 page_number: 2,
5501 page_hash: *blake3::hash(&sample_page(0x55)).as_bytes(),
5502 },
5503 ];
5504
5505 let conflicts = backend
5506 .conflicting_pages_since_snapshot(&cx, snapshot, &[2], &baselines)
5507 .expect("conflicting duplicate baselines must fail closed");
5508 assert_eq!(conflicts, vec![2]);
5509 }
5510
5511 #[test]
5512 fn test_generation_change_rejects_short_candidate_page() {
5513 let cx = test_cx();
5514 let vfs = MemoryVfs::new();
5515 let (mut backend, snapshot, page_two) = make_generation_transition_backend(&vfs, &cx);
5516 write_main_db_pages(
5517 &vfs,
5518 &cx,
5519 &[sqlite_page_one(
5520 u16::try_from(PAGE_SIZE).expect("page size fits u16"),
5521 )],
5522 );
5523 let baseline = TransactionConflictPageBaseline {
5524 page_number: 2,
5525 page_hash: *blake3::hash(&page_two).as_bytes(),
5526 };
5527
5528 let conflicts = backend
5529 .conflicting_pages_since_snapshot(&cx, snapshot, &[2], &[baseline])
5530 .expect("short page must fail closed");
5531 assert_eq!(conflicts, vec![2]);
5532 }
5533
5534 #[test]
5535 fn test_generation_change_rejects_database_page_size_change() {
5536 let cx = test_cx();
5537 let vfs = MemoryVfs::new();
5538 let (mut backend, snapshot, page_two) = make_generation_transition_backend(&vfs, &cx);
5539 write_main_db_pages(&vfs, &cx, &[sqlite_page_one(8192), page_two.clone()]);
5540 let baseline = TransactionConflictPageBaseline {
5541 page_number: 2,
5542 page_hash: *blake3::hash(&page_two).as_bytes(),
5543 };
5544
5545 let conflicts = backend
5546 .conflicting_pages_since_snapshot(&cx, snapshot, &[2], &[baseline])
5547 .expect("page-size change must fail closed");
5548 assert_eq!(conflicts, vec![2]);
5549 }
5550
5551 #[test]
5552 fn test_generation_change_decodes_64k_database_header_sentinel() {
5553 assert_eq!(
5554 sqlite_database_header_page_size(&sqlite_page_one(1)),
5555 Some(65_536)
5556 );
5557 }
5558
5559 #[test]
5560 fn test_adapter_batch_append_checksum_chain_matches_single_append() {
5561 let cx = test_cx();
5562 let vfs_single = MemoryVfs::new();
5563 let vfs_batch = MemoryVfs::new();
5564
5565 let mut adapter_single = make_adapter(&vfs_single, &cx);
5566 let mut adapter_batch = make_adapter(&vfs_batch, &cx);
5567
5568 let pages: Vec<Vec<u8>> = (0..4u8).map(sample_page).collect();
5569 let commit_sizes = [0_u32, 0, 0, 4];
5570
5571 for (index, page) in pages.iter().enumerate() {
5572 adapter_single
5573 .append_frame(
5574 &cx,
5575 u32::try_from(index + 1).expect("page number fits u32"),
5576 page,
5577 commit_sizes[index],
5578 )
5579 .expect("single append");
5580 }
5581
5582 let batch_frames: Vec<_> = pages
5583 .iter()
5584 .enumerate()
5585 .map(|(index, page)| WalFrameRef {
5586 page_number: u32::try_from(index + 1).expect("page number fits u32"),
5587 page_data: page,
5588 db_size_if_commit: commit_sizes[index],
5589 })
5590 .collect();
5591 adapter_batch
5592 .append_frames(&cx, &batch_frames)
5593 .expect("batch append");
5594
5595 assert_eq!(
5596 adapter_single.frame_count(),
5597 adapter_batch.frame_count(),
5598 "batch adapter append must preserve frame count"
5599 );
5600 assert_eq!(
5601 adapter_single.wal.running_checksum(),
5602 adapter_batch.wal.running_checksum(),
5603 "batch adapter append must preserve checksum chain"
5604 );
5605
5606 for frame_index in 0..pages.len() {
5607 let (single_header, single_data) = adapter_single
5608 .wal
5609 .read_frame(&cx, frame_index)
5610 .expect("read single frame");
5611 let (batch_header, batch_data) = adapter_batch
5612 .wal
5613 .read_frame(&cx, frame_index)
5614 .expect("read batch frame");
5615 assert_eq!(
5616 single_header, batch_header,
5617 "frame header {frame_index} must match"
5618 );
5619 assert_eq!(
5620 single_data, batch_data,
5621 "frame payload {frame_index} must match"
5622 );
5623 }
5624 }
5625
5626 #[test]
5627 fn test_adapter_prepared_batch_append_checksum_chain_matches_single_append() {
5628 let cx = test_cx();
5629 let vfs_single = MemoryVfs::new();
5630 let vfs_prepared = MemoryVfs::new();
5631
5632 let mut adapter_single = make_adapter(&vfs_single, &cx);
5633 let mut adapter_prepared = make_adapter(&vfs_prepared, &cx);
5634
5635 let pages: Vec<Vec<u8>> = (0..4u8).map(sample_page).collect();
5636 let commit_sizes = [0_u32, 0, 0, 4];
5637
5638 for (index, page) in pages.iter().enumerate() {
5639 adapter_single
5640 .append_frame(
5641 &cx,
5642 u32::try_from(index + 1).expect("page number fits u32"),
5643 page,
5644 commit_sizes[index],
5645 )
5646 .expect("single append");
5647 }
5648
5649 let batch_frames: Vec<_> = pages
5650 .iter()
5651 .enumerate()
5652 .map(|(index, page)| WalFrameRef {
5653 page_number: u32::try_from(index + 1).expect("page number fits u32"),
5654 page_data: page,
5655 db_size_if_commit: commit_sizes[index],
5656 })
5657 .collect();
5658 let mut prepared = adapter_prepared
5659 .prepare_append_frames(&batch_frames)
5660 .expect("prepare append")
5661 .expect("prepared batch");
5662 adapter_prepared
5663 .append_prepared_frames(&cx, &mut prepared)
5664 .expect("append prepared");
5665
5666 assert_eq!(
5667 adapter_single.frame_count(),
5668 adapter_prepared.frame_count(),
5669 "prepared adapter append must preserve frame count"
5670 );
5671 assert_eq!(
5672 adapter_single.wal.running_checksum(),
5673 adapter_prepared.wal.running_checksum(),
5674 "prepared adapter append must preserve checksum chain"
5675 );
5676
5677 for frame_index in 0..pages.len() {
5678 let (single_header, single_data) = adapter_single
5679 .wal
5680 .read_frame(&cx, frame_index)
5681 .expect("read single frame");
5682 let (prepared_header, prepared_data) = adapter_prepared
5683 .wal
5684 .read_frame(&cx, frame_index)
5685 .expect("read prepared frame");
5686 assert_eq!(
5687 single_header, prepared_header,
5688 "frame header {frame_index} must match"
5689 );
5690 assert_eq!(
5691 single_data, prepared_data,
5692 "frame payload {frame_index} must match"
5693 );
5694 }
5695 }
5696
5697 #[test]
5698 fn test_adapter_pre_finalize_reused_when_append_window_is_stable() {
5699 let cx = test_cx();
5700 let vfs_single = MemoryVfs::new();
5701 let vfs_prepared = MemoryVfs::new();
5702
5703 let mut adapter_single = make_adapter(&vfs_single, &cx);
5704 let mut adapter_prepared = make_adapter(&vfs_prepared, &cx);
5705
5706 let pages: Vec<Vec<u8>> = (0..3u8).map(sample_page).collect();
5707 let commit_sizes = [0_u32, 0, 3];
5708
5709 for (index, page) in pages.iter().enumerate() {
5710 adapter_single
5711 .append_frame(
5712 &cx,
5713 u32::try_from(index + 1).expect("page number fits u32"),
5714 page,
5715 commit_sizes[index],
5716 )
5717 .expect("single append");
5718 }
5719
5720 let batch_frames: Vec<_> = pages
5721 .iter()
5722 .enumerate()
5723 .map(|(index, page)| WalFrameRef {
5724 page_number: u32::try_from(index + 1).expect("page number fits u32"),
5725 page_data: page,
5726 db_size_if_commit: commit_sizes[index],
5727 })
5728 .collect();
5729 let mut prepared = adapter_prepared
5730 .prepare_append_frames(&batch_frames)
5731 .expect("prepare append")
5732 .expect("prepared batch");
5733 adapter_prepared
5734 .finalize_prepared_frames(&cx, &mut prepared)
5735 .expect("pre-finalize prepared batch");
5736 let finalized_for = prepared.finalized_for.expect("finalization state");
5737 let finalized_running_checksum = prepared
5738 .finalized_running_checksum
5739 .expect("finalized checksum");
5740
5741 adapter_prepared
5742 .append_prepared_frames(&cx, &mut prepared)
5743 .expect("append prepared");
5744
5745 assert_eq!(
5746 prepared.finalized_for,
5747 Some(finalized_for),
5748 "stable append window should reuse the pre-lock finalization state"
5749 );
5750 assert_eq!(
5751 prepared.finalized_running_checksum,
5752 Some(finalized_running_checksum),
5753 "stable append window should reuse the pre-lock finalized checksum"
5754 );
5755 assert_eq!(
5756 adapter_single.wal.running_checksum(),
5757 adapter_prepared.wal.running_checksum(),
5758 "stable reuse path must preserve checksum chain"
5759 );
5760 }
5761
5762 #[test]
5763 fn test_adapter_pre_finalize_reseeds_after_intervening_external_append() {
5764 let cx = test_cx();
5765 let baseline_vfs = MemoryVfs::new();
5766 let shared_vfs = MemoryVfs::new();
5767
5768 let mut baseline = make_adapter(&baseline_vfs, &cx);
5769 let mut prepared_writer = make_adapter(&shared_vfs, &cx);
5770 let intruder_file = open_wal_file(&shared_vfs, &cx);
5771 let intruder_wal = WalFile::open(&cx, intruder_file).expect("open shared WAL");
5772 let mut intruder = WalBackendAdapter::new(intruder_wal);
5773
5774 let pages: Vec<Vec<u8>> = (0..3u8).map(sample_page).collect();
5775 let commit_sizes = [0_u32, 0, 3];
5776 let intruder_page = sample_page(0xEE);
5777
5778 baseline
5779 .append_frame(&cx, 99, &intruder_page, 1)
5780 .expect("baseline intruder append");
5781 for (index, page) in pages.iter().enumerate() {
5782 baseline
5783 .append_frame(
5784 &cx,
5785 u32::try_from(index + 1).expect("page number fits u32"),
5786 page,
5787 commit_sizes[index],
5788 )
5789 .expect("baseline append");
5790 }
5791
5792 let batch_frames: Vec<_> = pages
5793 .iter()
5794 .enumerate()
5795 .map(|(index, page)| WalFrameRef {
5796 page_number: u32::try_from(index + 1).expect("page number fits u32"),
5797 page_data: page,
5798 db_size_if_commit: commit_sizes[index],
5799 })
5800 .collect();
5801 let mut prepared = prepared_writer
5802 .prepare_append_frames(&batch_frames)
5803 .expect("prepare append")
5804 .expect("prepared batch");
5805 prepared_writer
5806 .finalize_prepared_frames(&cx, &mut prepared)
5807 .expect("pre-finalize prepared batch");
5808 let stale_finalization_state = prepared.finalized_for;
5809
5810 intruder
5811 .append_frame(&cx, 99, &intruder_page, 1)
5812 .expect("intruder append");
5813 intruder.sync(&cx).expect("intruder sync");
5814
5815 prepared_writer
5816 .append_prepared_frames(&cx, &mut prepared)
5817 .expect("append prepared after external growth");
5818
5819 assert_ne!(
5820 prepared.finalized_for, stale_finalization_state,
5821 "intervening external growth should force prepared batch reseeding"
5822 );
5823 assert_eq!(
5824 baseline.wal.running_checksum(),
5825 prepared_writer.wal.running_checksum(),
5826 "reseeding path must preserve checksum chain"
5827 );
5828 assert_eq!(
5829 baseline.frame_count(),
5830 prepared_writer.frame_count(),
5831 "reseeding path must preserve frame count"
5832 );
5833 }
5834
5835 #[test]
5836 fn test_adapter_pins_read_snapshot_until_next_begin() {
5837 init_wal_publication_test_tracing();
5838 let cx = test_cx();
5839 let vfs = MemoryVfs::new();
5840
5841 let file_writer = open_wal_file(&vfs, &cx);
5842 let wal_writer =
5843 WalFile::create(&cx, file_writer, PAGE_SIZE, 0, test_salts()).expect("create WAL");
5844 let mut writer = WalBackendAdapter::new(wal_writer);
5845
5846 let file_reader = open_wal_file(&vfs, &cx);
5847 let wal_reader = WalFile::open(&cx, file_reader).expect("open WAL");
5848 let mut reader = WalBackendAdapter::new(wal_reader);
5849
5850 let v1 = sample_page(0x41);
5851 writer.append_frame(&cx, 3, &v1, 3).expect("append v1");
5852 writer.sync(&cx).expect("sync v1");
5853
5854 reader
5855 .begin_transaction(&cx)
5856 .expect("begin reader snapshot 1");
5857 let pinned_v1 = reader
5858 .pinned_read_snapshot()
5859 .expect("reader pins publication snapshot");
5860 assert_eq!(pinned_v1.last_commit_frame, Some(0));
5861 assert_eq!(pinned_v1.commit_count, 1);
5862 assert_eq!(pinned_v1.latest_frame_entries, 1);
5863 assert!(pinned_v1.lookup_contract_is_authoritative());
5864 assert_eq!(
5865 reader.read_page(&cx, 3).expect("reader sees v1"),
5866 Some(v1.clone())
5867 );
5868
5869 let v2 = sample_page(0x42);
5870 writer.append_frame(&cx, 3, &v2, 3).expect("append v2");
5871 writer.sync(&cx).expect("sync v2");
5872
5873 assert_eq!(
5875 reader
5876 .read_page(&cx, 3)
5877 .expect("reader remains on pinned snapshot"),
5878 Some(v1.clone())
5879 );
5880 assert_eq!(
5881 reader
5882 .pinned_read_snapshot()
5883 .expect("reader keeps the same pinned snapshot"),
5884 pinned_v1,
5885 "pinned publication metadata must stay stable until the next begin"
5886 );
5887
5888 reader
5890 .begin_transaction(&cx)
5891 .expect("begin reader snapshot 2");
5892 let pinned_v2 = reader
5893 .pinned_read_snapshot()
5894 .expect("reader repins publication snapshot");
5895 assert!(pinned_v2.publication_seq > pinned_v1.publication_seq);
5896 assert_eq!(pinned_v2.commit_count, 2);
5897 assert_eq!(pinned_v2.latest_frame_entries, 1);
5898 assert_eq!(reader.read_page(&cx, 3).expect("reader sees v2"), Some(v2));
5899 }
5900
5901 #[test]
5902 fn test_adapter_read_page_hides_uncommitted_frames() {
5903 let cx = test_cx();
5904 let vfs = MemoryVfs::new();
5905 let mut adapter = make_adapter(&vfs, &cx);
5906
5907 let committed = sample_page(0x31);
5908 let uncommitted = sample_page(0x32);
5909
5910 adapter
5911 .append_frame(&cx, 7, &committed, 7)
5912 .expect("append committed frame");
5913 adapter
5914 .append_frame(&cx, 7, &uncommitted, 0)
5915 .expect("append uncommitted frame");
5916
5917 let result = adapter.read_page(&cx, 7).expect("read committed page");
5918 assert_eq!(
5919 result,
5920 Some(committed),
5921 "reader must ignore uncommitted tail frames"
5922 );
5923 }
5924
5925 #[test]
5926 fn test_adapter_read_page_none_when_wal_has_no_commit_frame() {
5927 let cx = test_cx();
5928 let vfs = MemoryVfs::new();
5929 let mut adapter = make_adapter(&vfs, &cx);
5930
5931 adapter
5932 .append_frame(&cx, 3, &sample_page(0x44), 0)
5933 .expect("append uncommitted frame");
5934
5935 let result = adapter.read_page(&cx, 3).expect("read page");
5936 assert_eq!(result, None, "uncommitted WAL frames must stay invisible");
5937 }
5938
5939 #[test]
5940 fn test_adapter_read_page_empty_wal() {
5941 let cx = test_cx();
5942 let vfs = MemoryVfs::new();
5943 let mut adapter = make_adapter(&vfs, &cx);
5944
5945 let result = adapter.read_page(&cx, 1).expect("read from empty WAL");
5946 assert_eq!(result, None);
5947 }
5948
5949 #[test]
5950 fn test_adapter_sync() {
5951 let cx = test_cx();
5952 let vfs = MemoryVfs::new();
5953 let mut adapter = make_adapter(&vfs, &cx);
5954
5955 adapter
5956 .append_frame(&cx, 1, &sample_page(0), 1)
5957 .expect("append");
5958 adapter.sync(&cx).expect("sync should not fail");
5959 }
5960
5961 #[test]
5962 fn test_adapter_into_inner_fails_closed_until_sync() {
5963 let cx = test_cx();
5964 let staged_vfs = MemoryVfs::new();
5965 let mut staged = make_adapter(&staged_vfs, &cx);
5966
5967 staged
5968 .append_frame(&cx, 1, &sample_page(0), 1)
5969 .expect("append");
5970 assert!(
5971 matches!(staged.into_inner(), Err(FrankenError::Busy)),
5972 "an unsynced commit must prevent consuming the adapter"
5973 );
5974
5975 let synced_vfs = MemoryVfs::new();
5976 let mut synced = make_adapter(&synced_vfs, &cx);
5977 synced
5978 .append_frame(&cx, 1, &sample_page(0), 1)
5979 .expect("append");
5980 synced.sync(&cx).expect("sync staged commit");
5981
5982 assert_eq!(synced.inner().frame_count(), 1);
5983
5984 let wal = synced.into_inner().expect("sync drained the staged frames");
5985 assert_eq!(wal.frame_count(), 1);
5986 }
5987
5988 #[test]
5989 fn test_adapter_as_dyn_wal_backend() {
5990 let cx = test_cx();
5991 let vfs = MemoryVfs::new();
5992 let mut adapter = make_adapter(&vfs, &cx);
5993
5994 let backend: &mut dyn WalBackend = &mut adapter;
5996 backend
5997 .append_frame(&cx, 1, &sample_page(0x77), 1)
5998 .expect("append via dyn");
5999 assert_eq!(backend.frame_count(), 1);
6000
6001 let page = backend.read_page(&cx, 1).expect("read via dyn");
6002 assert_eq!(page, Some(sample_page(0x77)));
6003 }
6004
6005 #[test]
6006 fn test_publication_snapshots_are_visible_through_wal_backend_trait() {
6007 init_wal_publication_test_tracing();
6008 let cx = test_cx();
6009 let vfs = MemoryVfs::new();
6010
6011 let file_writer = open_wal_file(&vfs, &cx);
6012 let wal_writer =
6013 WalFile::create(&cx, file_writer, PAGE_SIZE, 0, test_salts()).expect("create WAL");
6014 let mut writer = WalBackendAdapter::new(wal_writer);
6015
6016 writer
6017 .append_frame(&cx, 4, &sample_page(0x84), 4)
6018 .expect("append committed frame");
6019 writer.sync(&cx).expect("sync committed frame");
6020
6021 let file_reader = open_wal_file(&vfs, &cx);
6022 let wal_reader = WalFile::open(&cx, file_reader).expect("open WAL");
6023 let mut reader = WalBackendAdapter::new(wal_reader);
6024 let backend: &mut dyn WalBackend = &mut reader;
6025
6026 let published_before = backend
6027 .published_snapshot()
6028 .expect("trait should expose the adapter publication summary");
6029 assert_eq!(published_before.last_commit_frame, None);
6030 assert_eq!(published_before.commit_count, 0);
6031
6032 let refreshed = backend
6033 .refresh_published_snapshot(&cx)
6034 .expect("refresh through trait should succeed")
6035 .expect("adapter should republish an existing committed prefix");
6036 assert_eq!(refreshed.last_commit_frame, Some(0));
6037 assert_eq!(refreshed.commit_count, 1);
6038 assert_eq!(refreshed.latest_frame_entries, 1);
6039
6040 backend
6041 .begin_transaction(&cx)
6042 .expect("begin_transaction through trait should pin snapshot");
6043 let pinned = backend
6044 .pinned_read_snapshot()
6045 .expect("trait should expose the pinned read snapshot");
6046 assert_eq!(pinned, refreshed);
6047 }
6048
6049 #[test]
6052 fn test_page_index_returns_correct_data() {
6053 let cx = test_cx();
6055 let vfs = MemoryVfs::new();
6056 let mut adapter = make_adapter(&vfs, &cx);
6057
6058 let page1 = sample_page(0x01);
6059 let page2 = sample_page(0x02);
6060 let page3 = sample_page(0x03);
6061
6062 adapter.append_frame(&cx, 1, &page1, 0).expect("append");
6063 adapter.append_frame(&cx, 2, &page2, 0).expect("append");
6064 adapter
6065 .append_frame(&cx, 3, &page3, 3)
6066 .expect("append commit");
6067
6068 assert_eq!(adapter.read_page(&cx, 1).expect("read"), Some(page1));
6070 assert_eq!(adapter.read_page(&cx, 2).expect("read"), Some(page2));
6071 assert_eq!(adapter.read_page(&cx, 3).expect("read"), Some(page3));
6072
6073 assert_eq!(adapter.read_page(&cx, 99).expect("read"), None);
6075 }
6076
6077 #[test]
6078 fn test_page_index_returns_latest_version() {
6079 let cx = test_cx();
6081 let vfs = MemoryVfs::new();
6082 let mut adapter = make_adapter(&vfs, &cx);
6083
6084 let old_data = sample_page(0xAA);
6085 let new_data = sample_page(0xBB);
6086
6087 adapter
6088 .append_frame(&cx, 5, &old_data, 0)
6089 .expect("append old");
6090 adapter
6091 .append_frame(&cx, 5, &new_data, 1)
6092 .expect("append new (commit)");
6093
6094 assert_eq!(
6095 adapter.read_page(&cx, 5).expect("read"),
6096 Some(new_data),
6097 "page index must return the latest frame for a page"
6098 );
6099 }
6100
6101 #[test]
6102 fn test_page_index_invalidated_on_wal_reset() {
6103 let cx = test_cx();
6106 let vfs = MemoryVfs::new();
6107 let mut adapter = make_adapter(&vfs, &cx);
6108
6109 let old_data = sample_page(0x11);
6110 adapter
6111 .append_frame(&cx, 1, &old_data, 1)
6112 .expect("append commit");
6113
6114 assert_eq!(adapter.read_page(&cx, 1).expect("read old"), Some(old_data));
6116
6117 let new_salts = WalSalts {
6119 salt1: 0xAAAA_BBBB,
6120 salt2: 0xCCCC_DDDD,
6121 };
6122 adapter
6123 .inner_mut()
6124 .expect("no staged batch blocks inner access")
6125 .reset(&cx, 1, new_salts, false)
6126 .expect("WAL reset");
6127
6128 let new_data = sample_page(0x22);
6130 adapter
6131 .append_frame(&cx, 1, &new_data, 1)
6132 .expect("append new generation commit");
6133
6134 let result = adapter.read_page(&cx, 1).expect("read after reset");
6136 assert_eq!(
6137 result,
6138 Some(new_data),
6139 "after WAL reset, page index must return new-generation data, not stale cached data"
6140 );
6141
6142 let old_only = sample_page(0x33);
6144 assert_eq!(
6146 adapter.read_page(&cx, 99).expect("read non-existent"),
6147 None,
6148 "pages from old WAL generation must not appear after reset"
6149 );
6150 drop(old_only);
6152 }
6153
6154 #[test]
6155 fn test_page_index_invalidated_on_same_salt_generation_change() {
6156 init_wal_publication_test_tracing();
6157 let cx = test_cx();
6160 let vfs = MemoryVfs::new();
6161 let mut adapter = make_adapter(&vfs, &cx);
6162
6163 let reused_salts = adapter.inner().header().salts;
6164 let old_data = sample_page(0x11);
6165 adapter
6166 .append_frame(&cx, 1, &old_data, 1)
6167 .expect("append commit");
6168 assert_eq!(adapter.read_page(&cx, 1).expect("read old"), Some(old_data));
6169
6170 adapter
6171 .inner_mut()
6172 .expect("no staged batch blocks inner access")
6173 .reset(&cx, 1, reused_salts, false)
6174 .expect("reset with same salts");
6175 let new_data = sample_page(0x22);
6176 adapter
6177 .append_frame(&cx, 2, &new_data, 2)
6178 .expect("append new generation commit");
6179 let refreshed = adapter
6180 .refresh_published_snapshot(&cx)
6181 .expect("refresh published snapshot after same-salt reset");
6182 assert_eq!(refreshed.generation.checkpoint_seq, 1);
6183 assert_eq!(refreshed.generation.salts, reused_salts);
6184 assert_eq!(refreshed.last_commit_frame, Some(0));
6185 assert_eq!(refreshed.commit_count, 1);
6186 assert_eq!(refreshed.latest_frame_entries, 1);
6187
6188 assert_eq!(
6189 adapter.read_page(&cx, 1).expect("old page should be gone"),
6190 None,
6191 "cached index entries from the previous generation must be invalidated"
6192 );
6193 assert_eq!(
6194 adapter.read_page(&cx, 2).expect("read new page"),
6195 Some(new_data),
6196 "adapter must resolve pages from the new generation even when salts are reused"
6197 );
6198 }
6199
6200 #[test]
6201 fn test_refresh_published_snapshot_materializes_existing_committed_prefix() {
6202 init_wal_publication_test_tracing();
6203 let cx = test_cx();
6204 let vfs = MemoryVfs::new();
6205
6206 let file_writer = open_wal_file(&vfs, &cx);
6207 let wal_writer =
6208 WalFile::create(&cx, file_writer, PAGE_SIZE, 0, test_salts()).expect("create WAL");
6209 let mut writer = WalBackendAdapter::new(wal_writer);
6210
6211 let p1 = sample_page(0x71);
6212 let p2 = sample_page(0x72);
6213 writer.append_frame(&cx, 1, &p1, 0).expect("append p1");
6214 writer
6215 .append_frame(&cx, 2, &p2, 2)
6216 .expect("append p2 commit");
6217 writer.sync(&cx).expect("sync writer");
6218
6219 let file_reader = open_wal_file(&vfs, &cx);
6220 let wal_reader = WalFile::open(&cx, file_reader).expect("open reader WAL");
6221 let mut reader = WalBackendAdapter::new(wal_reader);
6222
6223 let before = reader.published_snapshot();
6224 assert_eq!(before.last_commit_frame, None);
6225 assert_eq!(before.commit_count, 0);
6226 assert_eq!(before.latest_frame_entries, 0);
6227
6228 let refreshed = reader
6229 .refresh_published_snapshot(&cx)
6230 .expect("refresh published snapshot");
6231 assert_eq!(refreshed.last_commit_frame, Some(1));
6232 assert_eq!(refreshed.commit_count, 1);
6233 assert_eq!(refreshed.latest_frame_entries, 2);
6234 assert!(refreshed.lookup_contract_is_authoritative());
6235 assert_eq!(reader.read_page(&cx, 1).expect("read p1"), Some(p1));
6236 assert_eq!(reader.read_page(&cx, 2).expect("read p2"), Some(p2));
6237 }
6238
6239 #[test]
6240 fn test_page_index_incremental_extend_after_durable_sync() {
6241 let cx = test_cx();
6244 let vfs = MemoryVfs::new();
6245 let mut adapter = make_adapter(&vfs, &cx);
6246
6247 let page1 = sample_page(0x10);
6248 adapter
6249 .append_frame(&cx, 1, &page1, 1)
6250 .expect("append commit 1");
6251 adapter.sync(&cx).expect("durably publish commit 1");
6252
6253 assert_eq!(
6255 adapter.read_page(&cx, 1).expect("read"),
6256 Some(page1.clone())
6257 );
6258
6259 let page2 = sample_page(0x20);
6261 let page1_v2 = sample_page(0x30);
6262 adapter
6263 .append_frame(&cx, 2, &page2, 0)
6264 .expect("append page 2");
6265 adapter
6266 .append_frame(&cx, 1, &page1_v2, 3)
6267 .expect("append page 1 v2 (commit)");
6268 adapter.sync(&cx).expect("durably publish commit 2");
6269
6270 assert_eq!(
6272 adapter.read_page(&cx, 1).expect("read page 1 v2"),
6273 Some(page1_v2),
6274 "incremental index extend should pick up the updated page"
6275 );
6276 assert_eq!(adapter.read_page(&cx, 2).expect("read page 2"), Some(page2));
6277 }
6278
6279 fn commit_batch_pages() -> (Vec<u8>, Vec<u8>) {
6281 (sample_page(0x71), sample_page(0x72))
6282 }
6283
6284 fn assert_publication_unchanged(adapter: &WalBackendAdapter<impl VfsFile>, context: &str) {
6286 assert_eq!(
6287 adapter.published_snapshot.last_commit_frame, None,
6288 "{context}: publication must not advance before a successful sync"
6289 );
6290 assert_eq!(
6291 adapter.published_snapshot.commit_count, 0,
6292 "{context}: commit count must not advance before a successful sync"
6293 );
6294 assert!(
6295 adapter.published_snapshot.page_index.is_empty(),
6296 "{context}: no page may be visible before a successful sync"
6297 );
6298 }
6299
6300 #[test]
6301 fn test_append_frame_without_sync_leaves_publication_unchanged() {
6302 let cx = test_cx();
6303 let vfs = MemoryVfs::new();
6304 let mut adapter = make_adapter(&vfs, &cx);
6305
6306 let (p1, p2) = commit_batch_pages();
6307 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
6308 adapter.append_frame(&cx, 2, &p2, 2).expect("append commit");
6309
6310 assert_publication_unchanged(&adapter, "append_frame");
6311 assert_eq!(
6312 adapter.pending_publication_commit,
6313 Some(1),
6314 "append_frame must stage the commit horizon for a later sync"
6315 );
6316 }
6317
6318 #[test]
6319 fn test_append_frames_without_sync_leaves_publication_unchanged() {
6320 let cx = test_cx();
6321 let vfs = MemoryVfs::new();
6322 let mut adapter = make_adapter(&vfs, &cx);
6323
6324 let (p1, p2) = commit_batch_pages();
6325 let frames = [
6326 WalFrameRef {
6327 page_number: 1,
6328 page_data: &p1,
6329 db_size_if_commit: 0,
6330 },
6331 WalFrameRef {
6332 page_number: 2,
6333 page_data: &p2,
6334 db_size_if_commit: 2,
6335 },
6336 ];
6337 adapter
6338 .append_frames(&cx, &frames)
6339 .expect("append frames batch");
6340
6341 assert_publication_unchanged(&adapter, "append_frames");
6342 assert_eq!(
6343 adapter.pending_publication_commit,
6344 Some(1),
6345 "append_frames must stage the commit horizon for a later sync"
6346 );
6347 }
6348
6349 #[test]
6350 fn test_append_frames_tracked_without_sync_leaves_publication_unchanged() {
6351 let cx = test_cx();
6352 let vfs = MemoryVfs::new();
6353 let mut adapter = make_adapter(&vfs, &cx);
6354
6355 let (p1, p2) = commit_batch_pages();
6356 let frames = [
6357 WalFrameRef {
6358 page_number: 1,
6359 page_data: &p1,
6360 db_size_if_commit: 0,
6361 },
6362 WalFrameRef {
6363 page_number: 2,
6364 page_data: &p2,
6365 db_size_if_commit: 2,
6366 },
6367 ];
6368 adapter
6369 .append_frames_tracked(&cx, &frames, VfsWriteCompletion::new())
6370 .expect("append tracked frames batch");
6371
6372 assert_publication_unchanged(&adapter, "append_frames_tracked");
6373 assert_eq!(
6374 adapter.pending_publication_commit,
6375 Some(1),
6376 "append_frames_tracked must stage the commit horizon for a later sync"
6377 );
6378 }
6379
6380 #[test]
6381 fn test_append_prepared_frames_without_sync_leaves_publication_unchanged() {
6382 let cx = test_cx();
6383 let vfs = MemoryVfs::new();
6384 let mut adapter = make_adapter(&vfs, &cx);
6385
6386 let (p1, p2) = commit_batch_pages();
6387 let frames = [
6388 WalFrameRef {
6389 page_number: 1,
6390 page_data: &p1,
6391 db_size_if_commit: 0,
6392 },
6393 WalFrameRef {
6394 page_number: 2,
6395 page_data: &p2,
6396 db_size_if_commit: 2,
6397 },
6398 ];
6399 let mut prepared = adapter
6400 .prepare_append_frames(&frames)
6401 .expect("prepare append")
6402 .expect("prepared batch");
6403 adapter
6404 .append_prepared_frames(&cx, &mut prepared)
6405 .expect("append prepared");
6406
6407 assert_publication_unchanged(&adapter, "append_prepared_frames");
6408 assert_eq!(
6409 adapter.pending_publication_commit,
6410 Some(1),
6411 "append_prepared_frames must stage the commit horizon for a later sync"
6412 );
6413 }
6414
6415 #[test]
6416 fn test_append_prepared_frames_tracked_without_sync_leaves_publication_unchanged() {
6417 let cx = test_cx();
6418 let vfs = MemoryVfs::new();
6419 let mut adapter = make_adapter(&vfs, &cx);
6420
6421 let (p1, p2) = commit_batch_pages();
6422 let frames = [
6423 WalFrameRef {
6424 page_number: 1,
6425 page_data: &p1,
6426 db_size_if_commit: 0,
6427 },
6428 WalFrameRef {
6429 page_number: 2,
6430 page_data: &p2,
6431 db_size_if_commit: 2,
6432 },
6433 ];
6434 let mut prepared = adapter
6435 .prepare_append_frames(&frames)
6436 .expect("prepare append")
6437 .expect("prepared batch");
6438 adapter
6439 .append_prepared_frames_tracked(&cx, &mut prepared, VfsWriteCompletion::new())
6440 .expect("append prepared tracked");
6441
6442 assert_publication_unchanged(&adapter, "append_prepared_frames_tracked");
6443 assert_eq!(
6444 adapter.pending_publication_commit,
6445 Some(1),
6446 "append_prepared_frames_tracked must stage the commit horizon for a later sync"
6447 );
6448 }
6449
6450 #[test]
6451 fn test_successful_sync_publishes_staged_commit_horizon() {
6452 let cx = test_cx();
6453 let vfs = MemoryVfs::new();
6454 let mut adapter = make_adapter(&vfs, &cx);
6455
6456 let (p1, p2) = commit_batch_pages();
6457 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
6458 adapter.append_frame(&cx, 2, &p2, 2).expect("append commit");
6459 assert_publication_unchanged(&adapter, "before sync");
6460
6461 adapter.sync(&cx).expect("sync must succeed");
6462
6463 assert_eq!(
6464 adapter.published_snapshot.last_commit_frame,
6465 Some(1),
6466 "a successful sync must publish the staged commit horizon"
6467 );
6468 assert_eq!(
6469 adapter.published_snapshot.commit_count, 1,
6470 "a successful sync must publish the staged commit count"
6471 );
6472 assert_eq!(
6473 adapter.published_snapshot.page_index.len(),
6474 2,
6475 "a successful sync must publish every staged page"
6476 );
6477 assert_eq!(
6478 adapter.pending_publication_commit, None,
6479 "a published batch must no longer be staged"
6480 );
6481 assert!(
6482 adapter.pending_publication_frames.is_empty(),
6483 "a published batch must drain its staged frames"
6484 );
6485 }
6486
6487 #[test]
6488 fn test_failed_sync_advances_no_publication_and_retry_publishes() {
6489 let cx = test_cx();
6490 let vfs = CheckpointHandoffFaultVfs::new();
6491 let mut adapter = make_fault_adapter(&vfs, &cx);
6492
6493 let (p1, p2) = commit_batch_pages();
6494 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
6495 adapter.append_frame(&cx, 2, &p2, 2).expect("append commit");
6496
6497 vfs.fail_next_wal_sync();
6498 let failure = adapter
6499 .sync(&cx)
6500 .expect_err("injected WAL sync failure must surface");
6501 assert!(
6502 failure.to_string().contains("injected WAL sync failure"),
6503 "sync must report the injected durability failure, got: {failure}"
6504 );
6505
6506 assert_publication_unchanged(&adapter, "after failed sync");
6507 assert_eq!(
6508 adapter.pending_publication_commit,
6509 Some(1),
6510 "a failed sync must preserve the staged horizon for retry"
6511 );
6512 assert!(
6513 !adapter.pending_publication_frames.is_empty(),
6514 "a failed sync must preserve staged frames for retry"
6515 );
6516
6517 adapter.sync(&cx).expect("retry sync must succeed");
6519
6520 assert_eq!(
6521 adapter.published_snapshot.last_commit_frame,
6522 Some(1),
6523 "retrying sync must publish the preserved commit horizon"
6524 );
6525 assert_eq!(
6526 adapter.published_snapshot.commit_count, 1,
6527 "retrying sync must publish the preserved commit count"
6528 );
6529 assert_eq!(
6530 adapter.published_snapshot.page_index.len(),
6531 2,
6532 "retrying sync must publish every preserved page"
6533 );
6534 assert_eq!(
6535 adapter.pending_publication_commit, None,
6536 "a retried publication must clear the staged horizon"
6537 );
6538 }
6539
6540 #[test]
6541 fn test_failed_sync_then_append_cannot_drop_or_publish_pending() {
6542 let cx = test_cx();
6543 let vfs = CheckpointHandoffFaultVfs::new();
6544 let mut adapter = make_fault_adapter(&vfs, &cx);
6545
6546 let (p1, p2) = commit_batch_pages();
6547 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
6548 adapter.append_frame(&cx, 2, &p2, 2).expect("append commit");
6549
6550 vfs.fail_next_wal_sync();
6551 adapter
6552 .sync(&cx)
6553 .expect_err("injected WAL sync failure must surface");
6554
6555 let staged_after_failure = adapter.pending_publication_commit;
6556 let staged_frames_after_failure = adapter.pending_publication_frames.len();
6557 assert_eq!(
6558 staged_after_failure,
6559 Some(1),
6560 "failed sync must preserve the staged horizon"
6561 );
6562
6563 let p3 = sample_page(0x73);
6566 adapter
6567 .append_frame(&cx, 3, &p3, 3)
6568 .expect("append after failed sync");
6569
6570 assert_publication_unchanged(&adapter, "append after failed sync");
6571 assert!(
6572 adapter.pending_publication_frames.len() > staged_frames_after_failure,
6573 "append after a failed sync must extend, never discard, the staged batch"
6574 );
6575 assert_eq!(
6576 adapter.pending_publication_commit,
6577 Some(2),
6578 "append after a failed sync must carry the staged horizon forward"
6579 );
6580
6581 adapter.sync(&cx).expect("sync after failed attempt");
6583 assert_eq!(
6584 adapter.published_snapshot.last_commit_frame,
6585 Some(2),
6586 "recovered sync must publish the full preserved horizon"
6587 );
6588 assert_eq!(
6589 adapter.pending_publication_commit, None,
6590 "recovered sync must clear the staged horizon"
6591 );
6592 }
6593
6594 #[test]
6595 fn test_failed_sync_then_begin_transaction_then_append_fails_closed() {
6596 let cx = test_cx();
6597 let vfs = CheckpointHandoffFaultVfs::new();
6598 let mut adapter = make_fault_adapter(&vfs, &cx);
6599
6600 let (p1, p2) = commit_batch_pages();
6601 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
6602 adapter.append_frame(&cx, 2, &p2, 2).expect("append commit");
6603
6604 vfs.fail_next_wal_sync();
6605 adapter
6606 .sync(&cx)
6607 .expect_err("injected WAL sync failure must surface");
6608 assert_eq!(
6609 adapter.pending_publication_commit,
6610 Some(1),
6611 "failed sync must preserve the staged horizon"
6612 );
6613
6614 let begin_error = adapter
6618 .begin_transaction(&cx)
6619 .expect_err("begin_transaction must fail closed while frames are staged");
6620 assert!(
6621 matches!(begin_error, FrankenError::Busy),
6622 "staged-state rejection must be retryable Busy, not corruption: {begin_error:?}"
6623 );
6624 assert_publication_unchanged(&adapter, "begin_transaction refused after failed sync");
6625 assert_eq!(
6626 adapter.pending_publication_commit,
6627 Some(1),
6628 "a refused begin_transaction must not drop the staged horizon"
6629 );
6630 assert!(
6631 adapter.pinned_read_snapshot().is_none(),
6632 "a refused begin_transaction must not pin a read snapshot"
6633 );
6634
6635 adapter.refresh_before_append = true;
6638 let p3 = sample_page(0x74);
6639 let append_error = adapter
6640 .append_frame(&cx, 3, &p3, 3)
6641 .expect_err("append must fail closed while frames are staged");
6642 assert!(
6643 matches!(append_error, FrankenError::Busy),
6644 "append rejection must be retryable Busy: {append_error:?}"
6645 );
6646 assert_publication_unchanged(&adapter, "append refused after failed sync");
6647 assert_eq!(
6648 adapter.pending_publication_commit,
6649 Some(1),
6650 "a refused append must leave the staged horizon intact"
6651 );
6652 assert!(
6653 !adapter.pending_publication_frames.is_empty(),
6654 "a refused append must leave the staged frames intact"
6655 );
6656 adapter.refresh_before_append = false;
6657
6658 adapter.sync(&cx).expect("sync after failed attempt");
6660 assert_eq!(
6661 adapter.published_snapshot.last_commit_frame,
6662 Some(1),
6663 "recovered sync must publish the preserved horizon"
6664 );
6665 }
6666
6667 #[test]
6668 fn test_failed_sync_then_checkpoint_fails_closed_and_preserves_state() {
6669 let cx = test_cx();
6670 let vfs = CheckpointHandoffFaultVfs::new();
6671 let mut adapter = make_fault_adapter(&vfs, &cx);
6672
6673 let (p1, p2) = commit_batch_pages();
6674 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
6675 adapter.append_frame(&cx, 2, &p2, 2).expect("append commit");
6676
6677 vfs.fail_next_wal_sync();
6678 adapter
6679 .sync(&cx)
6680 .expect_err("injected WAL sync failure must surface");
6681
6682 let frames_before = adapter.frame_count();
6683 let staged_before = adapter.pending_publication_commit;
6684 let staged_frame_count_before = adapter.pending_publication_frames.len();
6685
6686 let mut writer = MockCheckpointPageWriter;
6690 let checkpoint_error = adapter
6691 .checkpoint(&cx, CheckpointMode::Passive, &mut writer, 0, None)
6692 .expect_err("checkpoint must fail closed while frames are staged");
6693 assert!(
6694 matches!(checkpoint_error, FrankenError::CheckpointFailed { .. }),
6695 "checkpoint rejection must be CheckpointFailed, not corruption: {checkpoint_error:?}"
6696 );
6697
6698 assert_eq!(
6699 adapter.frame_count(),
6700 frames_before,
6701 "a refused checkpoint must not mutate WAL bytes"
6702 );
6703 assert_publication_unchanged(&adapter, "checkpoint refused");
6704 assert_eq!(
6705 adapter.pending_publication_commit, staged_before,
6706 "a refused checkpoint must preserve the staged horizon"
6707 );
6708 assert_eq!(
6709 adapter.pending_publication_frames.len(),
6710 staged_frame_count_before,
6711 "a refused checkpoint must preserve the staged frames"
6712 );
6713
6714 adapter.sync(&cx).expect("retry sync must succeed");
6716 assert_eq!(
6717 adapter.published_snapshot.last_commit_frame,
6718 Some(1),
6719 "retry sync must publish the preserved horizon"
6720 );
6721 assert_eq!(
6722 adapter.pending_publication_commit, None,
6723 "a published batch must no longer be staged"
6724 );
6725 }
6726
6727 #[test]
6728 fn test_midtransaction_sync_preserves_uncommitted_frames_and_allows_continuation() {
6729 let cx = test_cx();
6730 let vfs = MemoryVfs::new();
6731 let mut adapter = make_adapter(&vfs, &cx);
6732
6733 let (p1, p2) = commit_batch_pages();
6734
6735 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
6738 assert_eq!(
6739 adapter.pending_publication_commit, None,
6740 "a non-commit append stages no commit horizon"
6741 );
6742 adapter
6743 .sync(&cx)
6744 .expect("mid-transaction sync must succeed");
6745
6746 assert_publication_unchanged(&adapter, "sync of uncommitted frames");
6747 assert!(
6748 !adapter.pending_publication_frames.is_empty(),
6749 "a mid-transaction sync must preserve durable-but-uncommitted frames"
6750 );
6751
6752 adapter
6754 .append_frame(&cx, 2, &p2, 2)
6755 .expect("commit append after mid-transaction sync must be allowed");
6756 assert_eq!(
6757 adapter.pending_publication_commit,
6758 Some(1),
6759 "the commit append must stage the horizon for the whole batch"
6760 );
6761 assert_publication_unchanged(&adapter, "commit staged but not yet synced");
6762
6763 adapter.sync(&cx).expect("commit sync must succeed");
6764
6765 assert_eq!(
6766 adapter.published_snapshot.last_commit_frame,
6767 Some(1),
6768 "the commit sync must publish the whole batch"
6769 );
6770 assert_eq!(
6771 adapter.published_snapshot.commit_count, 1,
6772 "the batch must publish exactly one commit"
6773 );
6774 assert_eq!(
6775 adapter.published_snapshot.page_index.len(),
6776 2,
6777 "both pages must be published exactly once"
6778 );
6779 assert_eq!(
6780 adapter.published_snapshot.page_index.get(&1),
6781 Some(&0),
6782 "page 1 must map to its frame from before the mid-transaction sync"
6783 );
6784 assert_eq!(
6785 adapter.published_snapshot.page_index.get(&2),
6786 Some(&1),
6787 "page 2 must map to the commit frame"
6788 );
6789 assert!(
6790 !adapter.has_pending_publication(),
6791 "a published batch must leave nothing staged"
6792 );
6793 }
6794
6795 #[test]
6796 fn test_inner_mut_fails_closed_while_batch_is_staged() {
6797 let cx = test_cx();
6798 let vfs = CheckpointHandoffFaultVfs::new();
6799 let mut adapter = make_fault_adapter(&vfs, &cx);
6800
6801 let (p1, p2) = commit_batch_pages();
6802 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
6803 adapter.append_frame(&cx, 2, &p2, 2).expect("append commit");
6804
6805 assert!(
6806 adapter.has_pending_publication(),
6807 "an appended-but-unsynced batch must report as pending"
6808 );
6809 assert!(
6812 matches!(adapter.inner_mut(), Err(FrankenError::Busy)),
6813 "inner_mut must fail closed with retryable Busy while frames are staged"
6814 );
6815 assert_eq!(
6816 adapter.pending_publication_commit,
6817 Some(1),
6818 "a refused inner_mut must preserve the staged horizon"
6819 );
6820
6821 adapter.sync(&cx).expect("sync staged batch");
6823 assert!(
6824 !adapter.has_pending_publication(),
6825 "a published batch must clear the pending flag"
6826 );
6827 adapter
6828 .inner_mut()
6829 .expect("inner_mut must succeed once the batch is drained");
6830 }
6831
6832 #[test]
6833 fn test_unpinned_refresh_does_not_expose_staged_horizon_before_sync() {
6834 let cx = test_cx();
6835 let vfs = MemoryVfs::new();
6836 let mut adapter = make_adapter(&vfs, &cx);
6837
6838 let (p1, p2) = commit_batch_pages();
6839 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
6840 adapter.append_frame(&cx, 2, &p2, 2).expect("append commit");
6841
6842 adapter
6845 .refresh_published_snapshot(&cx)
6846 .expect("refresh published snapshot");
6847 assert_publication_unchanged(&adapter, "refresh with staged frames");
6848 assert_eq!(
6849 adapter.pending_publication_commit,
6850 Some(1),
6851 "refresh must leave the staged horizon intact"
6852 );
6853
6854 adapter.sync(&cx).expect("sync staged batch");
6855 assert_eq!(
6856 adapter.published_snapshot.last_commit_frame,
6857 Some(1),
6858 "sync must publish once the staged batch is durable"
6859 );
6860 }
6861
6862 #[test]
6863 fn test_authorized_deferred_commit_publishes_without_claiming_fsync() {
6864 let cx = test_cx();
6865 let vfs = MemoryVfs::new();
6866 let mut adapter = make_adapter(&vfs, &cx);
6867
6868 let (p1, p2) = commit_batch_pages();
6869 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
6870 adapter.append_frame(&cx, 2, &p2, 2).expect("append commit");
6871 let fsynced_before = adapter.wal.last_fsynced_frame_count();
6872
6873 adapter
6874 .publish_authorized_deferred_commit(&cx)
6875 .expect("parallel-WAL authorization must publish the deferred commit");
6876
6877 assert_eq!(
6878 adapter.published_snapshot.last_commit_frame,
6879 Some(1),
6880 "the authorized commit marker must become visible"
6881 );
6882 assert_eq!(
6883 adapter.published_snapshot.commit_count, 1,
6884 "the authorized batch must publish exactly one commit"
6885 );
6886 assert!(
6887 !adapter.has_pending_publication(),
6888 "authorization must drain the staged publication horizon"
6889 );
6890 assert_eq!(
6891 adapter.wal.last_fsynced_frame_count(),
6892 fsynced_before,
6893 "deferred authorization must not claim or force an fsync"
6894 );
6895 adapter
6896 .begin_transaction(&cx)
6897 .expect("the next transaction must not see a stale Busy");
6898 }
6899
6900 #[test]
6901 fn test_commit_append_publishes_visibility_snapshot() {
6902 init_wal_publication_test_tracing();
6903 let cx = test_cx();
6904 let vfs = MemoryVfs::new();
6905 let mut adapter = make_adapter(&vfs, &cx);
6906
6907 let p1 = sample_page(0x41);
6908 let p2 = sample_page(0x42);
6909 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
6910 adapter.append_frame(&cx, 2, &p2, 2).expect("append commit");
6911 adapter.sync(&cx).expect("sync commit batch");
6914
6915 assert_eq!(
6916 adapter.published_snapshot.last_commit_frame,
6917 Some(1),
6918 "synced commit should publish the visible commit horizon"
6919 );
6920 assert_eq!(
6921 adapter.published_snapshot.commit_count, 1,
6922 "synced commit should track the visible WAL commit count"
6923 );
6924 assert_eq!(
6925 adapter.published_snapshot.page_index.len(),
6926 2,
6927 "published snapshot should track both committed pages"
6928 );
6929 assert_eq!(
6930 adapter.published_snapshot.page_index.get(&2),
6931 Some(&1),
6932 "published snapshot must map each page to its latest committed frame"
6933 );
6934 }
6935
6936 #[test]
6937 fn test_prepared_append_publishes_visibility_snapshot() {
6938 init_wal_publication_test_tracing();
6939 let cx = test_cx();
6940 let vfs = MemoryVfs::new();
6941 let mut adapter = make_adapter(&vfs, &cx);
6942
6943 let p1 = sample_page(0x51);
6944 let p2 = sample_page(0x52);
6945 let frames = [
6946 WalFrameRef {
6947 page_number: 1,
6948 page_data: &p1,
6949 db_size_if_commit: 0,
6950 },
6951 WalFrameRef {
6952 page_number: 2,
6953 page_data: &p2,
6954 db_size_if_commit: 2,
6955 },
6956 ];
6957 let mut prepared = adapter
6958 .prepare_append_frames(&frames)
6959 .expect("prepare append")
6960 .expect("prepared batch");
6961 adapter
6962 .append_prepared_frames(&cx, &mut prepared)
6963 .expect("append prepared");
6964 adapter.sync(&cx).expect("sync prepared commit batch");
6966
6967 assert_eq!(
6968 adapter.published_snapshot.last_commit_frame,
6969 Some(1),
6970 "synced prepared commit should publish the visible commit horizon"
6971 );
6972 assert_eq!(
6973 adapter.published_snapshot.commit_count, 1,
6974 "synced prepared commit should track the visible WAL commit count"
6975 );
6976 assert_eq!(
6977 adapter.published_snapshot.page_index.len(),
6978 2,
6979 "synced prepared commit should publish all committed pages"
6980 );
6981 assert_eq!(
6982 adapter.published_snapshot.page_index.get(&2),
6983 Some(&1),
6984 "prepared commit append must map each page to its latest committed frame"
6985 );
6986 }
6987
6988 #[test]
6989 fn test_commit_publication_refreshes_external_prefix_before_local_commit() {
6990 let cx = test_cx();
6991 let vfs = MemoryVfs::new();
6992
6993 let file_writer = open_wal_file(&vfs, &cx);
6994 let wal_writer =
6995 WalFile::create(&cx, file_writer, PAGE_SIZE, 0, test_salts()).expect("create WAL");
6996 let mut writer = WalBackendAdapter::new(wal_writer);
6997
6998 let file_follower = open_wal_file(&vfs, &cx);
6999 let wal_follower = WalFile::open(&cx, file_follower).expect("open WAL");
7000 let mut follower = WalBackendAdapter::new(wal_follower);
7001
7002 let p1 = sample_page(0x61);
7003 writer
7004 .append_frame(&cx, 1, &p1, 1)
7005 .expect("writer commit 1");
7006 writer.sync(&cx).expect("sync writer commit 1");
7007
7008 let p2 = sample_page(0x62);
7009 writer
7010 .append_frame(&cx, 2, &p2, 2)
7011 .expect("writer commit 2");
7012 writer.sync(&cx).expect("sync writer commit 2");
7013
7014 let p3 = sample_page(0x63);
7015 follower
7016 .append_frame(&cx, 3, &p3, 3)
7017 .expect("follower local commit");
7018
7019 assert_eq!(
7020 follower.published_snapshot.last_commit_frame,
7021 Some(2),
7022 "local commit should publish on top of refreshed external WAL state"
7023 );
7024 assert_eq!(
7025 follower.published_snapshot.commit_count, 3,
7026 "local commit publication should include refreshed external commits"
7027 );
7028 assert_eq!(
7029 follower.published_snapshot.page_index.get(&1),
7030 Some(&0),
7031 "refresh-before-append should preserve earlier committed pages"
7032 );
7033 assert_eq!(
7034 follower.published_snapshot.page_index.get(&2),
7035 Some(&1),
7036 "refresh-before-append should publish externally committed pages"
7037 );
7038 assert_eq!(
7039 follower.published_snapshot.page_index.get(&3),
7040 Some(&2),
7041 "local commit should extend the published WAL visibility map"
7042 );
7043 assert_eq!(follower.read_page(&cx, 1).expect("read p1"), Some(p1));
7044 assert_eq!(follower.read_page(&cx, 2).expect("read p2"), Some(p2));
7045 assert_eq!(follower.read_page(&cx, 3).expect("read p3"), Some(p3));
7046 }
7047
7048 #[test]
7049 fn test_truncate_checkpoint_republishes_empty_generation_snapshot() {
7050 init_wal_publication_test_tracing();
7051 let cx = test_cx();
7052 let vfs = MemoryVfs::new();
7053 let mut adapter = make_adapter(&vfs, &cx);
7054 let mut writer = MockCheckpointPageWriter;
7055
7056 adapter
7057 .append_frame(&cx, 1, &sample_page(0x61), 1)
7058 .expect("append committed frame");
7059 adapter.sync(&cx).expect("sync committed frame");
7063 let before = adapter.published_snapshot();
7064 assert_eq!(before.last_commit_frame, Some(0));
7065 assert_eq!(before.commit_count, 1);
7066 assert_eq!(before.latest_frame_entries, 1);
7067
7068 let result = adapter
7069 .checkpoint(&cx, CheckpointMode::Truncate, &mut writer, 0, None)
7070 .expect("truncate checkpoint");
7071 assert!(result.completed);
7072 assert!(result.wal_was_reset);
7073
7074 let after = adapter.published_snapshot();
7075 assert_ne!(
7076 before.generation, after.generation,
7077 "truncate checkpoint should publish a new WAL generation"
7078 );
7079 assert_eq!(after.last_commit_frame, None);
7080 assert_eq!(after.commit_count, 0);
7081 assert_eq!(after.latest_frame_entries, 0);
7082 assert!(after.lookup_contract_is_authoritative());
7083 }
7084
7085 #[test]
7088 fn test_partial_index_falls_back_to_linear_scan() {
7089 init_wal_publication_test_tracing();
7090 let cx = test_cx();
7093 let vfs = MemoryVfs::new();
7094 let mut adapter = make_adapter(&vfs, &cx);
7095
7096 adapter.set_page_index_cap(2);
7099
7100 let p1 = sample_page(0x01);
7103 let p2 = sample_page(0x02);
7104 let p3 = sample_page(0x03);
7105 let p4 = sample_page(0x04);
7106 let p5 = sample_page(0x05);
7107
7108 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
7109 adapter.append_frame(&cx, 2, &p2, 0).expect("append p2");
7110 adapter.append_frame(&cx, 3, &p3, 0).expect("append p3");
7111 adapter.append_frame(&cx, 4, &p4, 0).expect("append p4");
7112 adapter
7113 .append_frame(&cx, 5, &p5, 5)
7114 .expect("append p5 (commit)");
7115
7116 assert_eq!(
7118 adapter.read_page(&cx, 1).expect("read p1"),
7119 Some(p1),
7120 "indexed page should be found via HashMap"
7121 );
7122 assert_eq!(
7123 adapter.read_page(&cx, 2).expect("read p2"),
7124 Some(p2),
7125 "indexed page should be found via HashMap"
7126 );
7127
7128 assert_eq!(
7131 adapter.read_page(&cx, 3).expect("read p3"),
7132 Some(p3),
7133 "non-indexed page must be found via linear scan fallback"
7134 );
7135 assert_eq!(
7136 adapter.read_page(&cx, 4).expect("read p4"),
7137 Some(p4),
7138 "non-indexed page must be found via linear scan fallback"
7139 );
7140 assert_eq!(
7141 adapter.read_page(&cx, 5).expect("read p5"),
7142 Some(p5),
7143 "non-indexed page must be found via linear scan fallback"
7144 );
7145
7146 assert_eq!(
7148 adapter.read_page(&cx, 99).expect("read non-existent"),
7149 None,
7150 "non-existent page must return None even with partial index"
7151 );
7152
7153 assert!(
7155 adapter.published_snapshot.index_is_partial,
7156 "index_is_partial should be true when cap is exceeded"
7157 );
7158 }
7159
7160 #[test]
7161 fn test_partial_index_returns_latest_version_via_fallback() {
7162 let cx = test_cx();
7166 let vfs = MemoryVfs::new();
7167 let mut adapter = make_adapter(&vfs, &cx);
7168
7169 adapter.set_page_index_cap(1);
7171
7172 let old_p2 = sample_page(0xAA);
7173 let new_p2 = sample_page(0xBB);
7174
7175 adapter
7177 .append_frame(&cx, 1, &sample_page(0x01), 0)
7178 .expect("append p1");
7179 adapter
7181 .append_frame(&cx, 2, &old_p2, 0)
7182 .expect("append p2 old");
7183 adapter
7186 .append_frame(&cx, 2, &new_p2, 3)
7187 .expect("append p2 new (commit)");
7188
7189 assert_eq!(
7191 adapter.read_page(&cx, 2).expect("read p2"),
7192 Some(new_p2),
7193 "backwards scan must return the most recent frame for the page"
7194 );
7195 }
7196
7197 #[test]
7198 fn test_lookup_contract_distinguishes_authoritative_and_fallback_paths() {
7199 init_wal_publication_test_tracing();
7200 let cx = test_cx();
7201 let vfs = MemoryVfs::new();
7202 let mut adapter = make_adapter(&vfs, &cx);
7203 adapter.set_page_index_cap(1);
7204
7205 let p1 = sample_page(0x01);
7206 let p2 = sample_page(0x02);
7207 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
7208 adapter
7209 .append_frame(&cx, 2, &p2, 2)
7210 .expect("append p2 commit");
7211
7212 let last_commit = adapter
7213 .inner_mut()
7214 .expect("no staged batch blocks inner access")
7215 .last_commit_frame(&cx)
7216 .expect("last commit")
7217 .expect("commit exists");
7218 adapter
7219 .publish_visible_snapshot(&cx, Some(last_commit), "lookup_contract_test")
7220 .expect("build published snapshot");
7221 let snapshot = adapter.published_snapshot.clone();
7222
7223 assert_eq!(
7224 adapter
7225 .resolve_visible_frame(&cx, &snapshot, 1)
7226 .expect("resolve indexed page"),
7227 WalPageLookupResolution::AuthoritativeHit { frame_index: 0 }
7228 );
7229 assert_eq!(
7230 adapter
7231 .resolve_visible_frame(&cx, &snapshot, 2)
7232 .expect("resolve fallback page"),
7233 WalPageLookupResolution::PartialIndexFallbackHit { frame_index: 1 }
7234 );
7235 assert_eq!(
7236 adapter
7237 .resolve_visible_frame(&cx, &snapshot, 99)
7238 .expect("resolve missing page"),
7239 WalPageLookupResolution::PartialIndexFallbackMiss
7240 );
7241 }
7242
7243 #[test]
7244 fn test_lookup_contract_is_authoritative_by_default() {
7245 let cx = test_cx();
7246 let vfs = MemoryVfs::new();
7247 let mut adapter = make_adapter(&vfs, &cx);
7248
7249 let p1 = sample_page(0x11);
7250 let p2 = sample_page(0x22);
7251 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
7252 adapter
7253 .append_frame(&cx, 2, &p2, 2)
7254 .expect("append p2 commit");
7255
7256 let last_commit = adapter
7257 .inner_mut()
7258 .expect("no staged batch blocks inner access")
7259 .last_commit_frame(&cx)
7260 .expect("last commit")
7261 .expect("commit exists");
7262 adapter
7263 .publish_visible_snapshot(&cx, Some(last_commit), "lookup_contract_default")
7264 .expect("build published snapshot");
7265 let snapshot = adapter.published_snapshot.clone();
7266
7267 assert!(
7268 !snapshot.index_is_partial,
7269 "default index should be authoritative"
7270 );
7271 assert_eq!(
7272 adapter
7273 .resolve_visible_frame(&cx, &snapshot, 1)
7274 .expect("resolve page 1"),
7275 WalPageLookupResolution::AuthoritativeHit { frame_index: 0 }
7276 );
7277 assert_eq!(
7278 adapter
7279 .resolve_visible_frame(&cx, &snapshot, 2)
7280 .expect("resolve page 2"),
7281 WalPageLookupResolution::AuthoritativeHit { frame_index: 1 }
7282 );
7283 assert_eq!(
7284 adapter
7285 .resolve_visible_frame(&cx, &snapshot, 99)
7286 .expect("resolve missing page"),
7287 WalPageLookupResolution::AuthoritativeMiss
7288 );
7289 }
7290
7291 #[test]
7292 fn test_committed_txns_since_page_uses_visible_frame_horizon() {
7293 let cx = test_cx();
7294 let vfs = MemoryVfs::new();
7295 let mut adapter = make_adapter(&vfs, &cx);
7296
7297 let p1 = sample_page(0x31);
7298 let p2 = sample_page(0x32);
7299 let p3 = sample_page(0x33);
7300
7301 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
7302 adapter.append_frame(&cx, 2, &p2, 2).expect("commit tx1");
7303 adapter.append_frame(&cx, 3, &p3, 0).expect("append p3");
7304 adapter.append_frame(&cx, 2, &p2, 3).expect("commit tx2");
7305
7306 assert_eq!(
7307 adapter
7308 .committed_txns_since_page(&cx, 1)
7309 .expect("count txns since page 1"),
7310 1
7311 );
7312 assert_eq!(
7313 adapter
7314 .committed_txns_since_page(&cx, 2)
7315 .expect("count txns since page 2"),
7316 0
7317 );
7318 assert_eq!(
7319 adapter
7320 .committed_txns_since_page(&cx, 99)
7321 .expect("count txns since missing page"),
7322 2
7323 );
7324 assert_eq!(
7325 adapter
7326 .committed_txn_count(&cx)
7327 .expect("count visible transactions"),
7328 2
7329 );
7330 }
7331
7332 #[test]
7333 fn test_conflicting_pages_since_snapshot_detects_later_wal_writes() {
7334 let cx = test_cx();
7335 let vfs = MemoryVfs::new();
7336 let mut adapter = make_adapter(&vfs, &cx);
7337
7338 let p1 = sample_page(0x41);
7339 let p2_before = sample_page(0x42);
7340 let p2_after = sample_page(0x43);
7341 let p3 = sample_page(0x44);
7342
7343 adapter.append_frame(&cx, 1, &p1, 0).expect("append p1");
7344 adapter
7345 .append_frame(&cx, 2, &p2_before, 2)
7346 .expect("commit tx1");
7347 adapter
7348 .begin_transaction(&cx)
7349 .expect("pin transaction snapshot");
7350 let pinned = adapter
7351 .pinned_read_snapshot()
7352 .expect("transaction should expose pinned WAL snapshot");
7353 let conflict_snapshot = TransactionConflictSnapshot {
7354 generation: pinned.generation,
7355 last_commit_frame: pinned.last_commit_frame,
7356 commit_count: pinned.commit_count,
7357 };
7358
7359 adapter
7360 .append_frame(&cx, 3, &p3, 0)
7361 .expect("append unrelated later page");
7362 adapter
7363 .append_frame(&cx, 2, &p2_after, 3)
7364 .expect("commit later page 2 update");
7365
7366 let conflicts = adapter
7367 .conflicting_pages_since_snapshot(&cx, conflict_snapshot, &[2, 99], &[])
7368 .expect("conflict check should scan later committed frames");
7369 assert_eq!(conflicts, vec![2]);
7370
7371 let unrelated = adapter
7372 .conflicting_pages_since_snapshot(&cx, conflict_snapshot, &[99], &[])
7373 .expect("unrelated page should stay conflict-free");
7374 assert!(unrelated.is_empty());
7375 }
7376
7377 #[test]
7380 fn test_checkpoint_adapter_write_page() {
7381 let cx = test_cx();
7382 let mut writer = MockCheckpointPageWriter;
7383 let mut adapter = CheckpointTargetAdapterRef {
7384 writer: &mut writer,
7385 };
7386
7387 let page_no = PageNumber::new(1).expect("valid page number");
7388 adapter
7389 .write_page(&cx, page_no, &[0u8; 4096])
7390 .expect("write_page");
7391 }
7392
7393 #[test]
7394 fn test_checkpoint_adapter_truncate_db() {
7395 let cx = test_cx();
7396 let mut writer = MockCheckpointPageWriter;
7397 let mut adapter = CheckpointTargetAdapterRef {
7398 writer: &mut writer,
7399 };
7400
7401 adapter.truncate_db(&cx, 10).expect("truncate_db");
7402 }
7403
7404 #[test]
7405 fn test_checkpoint_adapter_sync_db() {
7406 let cx = test_cx();
7407 let mut writer = MockCheckpointPageWriter;
7408 let mut adapter = CheckpointTargetAdapterRef {
7409 writer: &mut writer,
7410 };
7411
7412 adapter.sync_db(&cx).expect("sync_db");
7413 }
7414
7415 #[test]
7416 fn test_checkpoint_adapter_as_dyn_target() {
7417 let cx = test_cx();
7418 let mut writer = MockCheckpointPageWriter;
7419 let mut adapter = CheckpointTargetAdapterRef {
7420 writer: &mut writer,
7421 };
7422
7423 let target: &mut dyn CheckpointTarget = &mut adapter;
7425 let page_no = PageNumber::new(3).expect("valid page number");
7426 target
7427 .write_page(&cx, page_no, &[0u8; 4096])
7428 .expect("write via dyn");
7429 target.truncate_db(&cx, 5).expect("truncate via dyn");
7430 target.sync_db(&cx).expect("sync via dyn");
7431 }
7432}