Skip to main content

fsqlite_core/
wal_adapter.rs

1//! Adapters bridging the WAL and pager crates at runtime.
2//!
3//! These adapters break the circular dependency between `fsqlite-pager` and
4//! `fsqlite-wal`:
5//!
6//! - [`WalBackendAdapter`] wraps `WalFile` to satisfy the pager's
7//!   [`WalBackend`] trait (pager -> WAL direction).
8//! - `CheckpointTargetAdapterRef` wraps `CheckpointPageWriter` to satisfy the
9//!   WAL executor's [`CheckpointTarget`] trait (WAL -> pager direction).
10
11use 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
96// ---------------------------------------------------------------------------
97// WalBackendAdapter: WalFile -> WalBackend
98// ---------------------------------------------------------------------------
99
100/// Completes a tracked backend write as an error if it is discarded before
101/// ownership reaches the VFS source that performs the physical mutation.
102struct 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
124/// Adapter wrapping [`WalFile`] to implement the pager's [`WalBackend`] trait.
125///
126/// The pager calls `dyn WalBackend` during WAL-mode commits and page reads.
127/// This adapter delegates those calls to the concrete `WalFile<F>` from
128/// `fsqlite-wal`.
129/// Default steady-state page-index cap.
130///
131/// Normal runtime operation keeps the published WAL page index authoritative
132/// for the full visible generation. Tests can still lower this cap explicitly
133/// to exercise the bounded fallback path.
134const 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/// How a visible page lookup was resolved for the current WAL generation.
150///
151/// The steady-state contract is that `Authoritative*` outcomes come from a
152/// complete per-generation index. `PartialIndexFallback*` outcomes are an
153/// explicit slow-path exception used only when a lowered cap makes the
154/// in-memory index incomplete.
155#[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/// Immutable visibility snapshot published for one WAL generation.
195///
196/// Readers pin one of these snapshots at transaction start so page lookups stay
197/// bound to a stable committed horizon even if later commits advance the active
198/// publication plane.
199#[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    /// Guard so commit-time append refresh runs only once per commit batch.
247    refresh_before_append: bool,
248    /// Active commit-published visibility plane for the current WAL generation.
249    published_snapshot: WalPublishedSnapshot,
250    /// Monotonic publication sequence assigned to the next published snapshot.
251    next_publication_seq: u64,
252    /// Transaction-bounded read snapshot pinned at `begin_transaction()`.
253    read_snapshot: Option<WalPublishedSnapshot>,
254    /// Frames appended after the last published commit horizon.
255    pending_publication_frames: Vec<PendingPublicationFrame>,
256    /// Highest commit frame staged by the append path but not yet published.
257    ///
258    /// Appends only stage this horizon; publication is deferred until
259    /// [`WalBackend::sync`] durably persists the frames. Preserved verbatim when
260    /// a sync fails so the next successful sync republishes the same batch.
261    pending_publication_commit: Option<usize>,
262    /// WAL generation observed when the pending frames were staged.
263    ///
264    /// Publication is refused if the generation moves before the sync lands,
265    /// because a checkpoint or restart invalidates the staged frame indices.
266    pending_publication_generation: Option<WalGenerationIdentity>,
267    /// Optional FEC commit hook for encoding repair symbols on commit.
268    #[cfg(all(not(target_arch = "wasm32"), feature = "native"))]
269    fec_hook: Option<FecCommitHook>,
270    /// Accumulated FEC commit results (for later sidecar persistence).
271    #[cfg(all(not(target_arch = "wasm32"), feature = "native"))]
272    fec_pending: Vec<FecCommitResult>,
273    /// Maximum number of unique pages the index will track. Defaults to a
274    /// full authoritative index in steady state. Tests can lower the cap to
275    /// exercise the partial-index fallback path explicitly.
276    page_index_cap: usize,
277}
278
279impl<F: VfsFile> WalBackendAdapter<F> {
280    /// Wrap an existing [`WalFile`] in the adapter (FEC disabled).
281    #[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    /// Wrap an existing [`WalFile`] with an FEC commit hook.
302    #[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    /// Whether staged, unpublished frames remain.
322    ///
323    /// Staged frames may already be durable — an intermediate sync makes them so
324    /// without committing them — but they are not yet part of the published
325    /// visibility plane. Callers that would discard, consume, or replace this
326    /// adapter must check this first: dropping the staged metadata loses the
327    /// batch, and a freshly wrapped adapter would republish those frames
328    /// straight from the WAL with no knowledge of their publication state
329    /// (GH #187).
330    #[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    /// Consume the adapter and return the inner [`WalFile`].
336    ///
337    /// Fails closed while staged, unpublished frames remain: consuming the
338    /// adapter discards the staged publication metadata, and the `WalFile` can
339    /// be rewrapped by an adapter that would then publish those frames without
340    /// knowing whether they were ever published or fsynced (GH #187). Drain the
341    /// batch with a successful commit sync first.
342    /// Returns [`FrankenError::Busy`]: this is a retryable ordering condition,
343    /// not database corruption.
344    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    /// Borrow the inner [`WalFile`].
352    #[must_use]
353    pub fn inner(&self) -> &WalFile<F> {
354        &self.wal
355    }
356
357    /// Mutably borrow the inner [`WalFile`] for explicit external mutation.
358    ///
359    /// Invalidates the publication plane, since the caller may mutate WAL state
360    /// arbitrarily. That invalidation discards any staged batch, so this fails
361    /// closed while one exists rather than silently dropping the commit horizon
362    /// (GH #187): after the discard, a later publish would see no pending state
363    /// and could expose frames that were never fsynced. Drain the batch with a
364    /// successful sync first.
365    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    /// Capture the currently published WAL visibility summary for this handle.
374    ///
375    /// This is a cheap snapshot of the publication plane the adapter has
376    /// already materialized. Call [`Self::refresh_published_snapshot`] first if
377    /// the caller needs to bind to the latest on-disk committed prefix.
378    #[must_use]
379    pub fn published_snapshot(&self) -> WalPublicationSnapshot {
380        wal_publication_snapshot_from_published(&self.published_snapshot)
381    }
382
383    /// Capture the currently pinned read snapshot, if this handle has one.
384    #[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    /// Refresh this handle from disk and republish the latest committed WAL
392    /// visibility summary without pinning a read transaction.
393    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    /// Discard published and pinned snapshots after external WAL mutation.
401    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    /// Publish an immutable visibility snapshot for the current committed WAL prefix.
411    ///
412    /// The commit path advances this plane directly, and readers pin a clone of
413    /// the published snapshot instead of mutating shared lookup state under an
414    /// active transaction.
415    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    /// Resolve the most recent visible frame for `page_number`.
546    ///
547    /// The normal contract is `Authoritative*`: the published page index fully
548    /// covers the visible WAL generation, so a miss means the page is absent.
549    /// `PartialIndexFallback*` is a bounded slow-path used only when the capped
550    /// index is known to be incomplete.
551    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    /// Scan frame headers from `start..=end` (inclusive), populate the page index,
578    /// and count commit frames in the same pass.
579    ///
580    /// Since we scan forward, later frames naturally overwrite earlier entries
581    /// for the same page number, ensuring "newest frame wins" semantics.
582    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            // Only insert if we haven't hit the capacity cap, or if this page
598            // is already tracked (update is free).
599            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                // A page was dropped because the index is full -- mark it as
605                // partial so that `read_page` knows a HashMap miss cannot be
606                // trusted and must fall back to a linear scan.
607                *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    /// Backwards linear scan of committed frames to find a page that was not
617    /// captured by the capped page index.
618    ///
619    /// Scans from `last_commit_frame` down to frame 0 and returns the index
620    /// of the first (i.e., most recent) frame containing `page_number`, or
621    /// `None` if the page is not in the WAL at all.
622    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    /// Take any pending FEC commit results for sidecar persistence.
638    #[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    /// Whether FEC encoding is active.
644    #[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    /// Discard buffered FEC pages (e.g. on transaction rollback).
653    #[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    /// Override the page index capacity (for testing only).
661    #[cfg(test)]
662    fn set_page_index_cap(&mut self, cap: usize) {
663        self.page_index_cap = cap;
664        // Invalidate so the next read rebuilds with the new cap.
665        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        // While a local batch is staged, the WAL's own commit horizon includes
763        // frames this handle appended but has not yet fsynced. Refresh and
764        // unpinned read paths must not expose them, so clamp to the durable
765        // prefix. With nothing staged the horizon is used unchanged, preserving
766        // publication of commits made durable elsewhere.
767        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        // Fail closed. A non-empty staged batch means a durability barrier has
787        // not completed: either no sync has run, or one failed. Discarding it
788        // here would silently drop the horizon, and republishing straight from
789        // the WAL would expose frames that were never fsynced. Any path that
790        // sets `refresh_before_append` while a batch is staged — a failed sync
791        // followed by `begin_transaction`, or by `checkpoint` — funnels through
792        // here, so guarding this single choke point covers all of them.
793        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    /// Drop every staged frame and the horizon that would have published it.
803    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    /// Stage a commit horizon for publication without advancing visibility.
810    ///
811    /// Appends never publish directly: the frames may sit in the host page cache
812    /// with no durable backing, so exposing them to readers would surface a
813    /// commit that a crash could still erase. The horizon is retained until
814    /// [`WalBackend::sync`] persists the batch.
815    fn stage_pending_commit_publication(&mut self, last_commit_frame: usize) -> Result<()> {
816        let generation = self.wal.generation_identity();
817        // Fail closed rather than silently overwriting: staged frame indices are
818        // only meaningful within one generation, so a mixed-generation batch
819        // must never be merged into a single publishable horizon.
820        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    /// Confirm the staged horizon is still publishable against the live WAL.
837    ///
838    /// Refuses when the generation moved (a checkpoint or restart reindexes the
839    /// WAL, invalidating staged frame indices), when the WAL is shorter than the
840    /// staged horizon, or when the WAL does not yet report the staged commit.
841    /// Callers must leave the pending state untouched on refusal so a later sync
842    /// can retry the same batch.
843    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        // Delegate to the WAL's own durability tracker rather than duplicating
896        // it: only it knows how far a successful fsync actually reached.
897        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    /// Publish a logically authorized `synchronous=NORMAL` commit without
910    /// claiming that an fsync occurred.
911    ///
912    /// The pager calls this only after the parallel-WAL certificate and both
913    /// tracked write completions are terminal. A failed-sync path never reaches
914    /// this hook, so its pending horizon remains fail-closed for a later retry.
915    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    /// Publish the staged commit horizon after a successful durability barrier.
927    ///
928    /// Fully synchronous: the staged frames already carry every page/frame pair
929    /// the snapshot needs, so no WAL scan — and therefore no async I/O — is
930    /// required. On refusal or failure the pending state is preserved verbatim
931    /// so the next successful sync retries the identical batch.
932    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    /// Install a published snapshot from staged frames alone.
964    ///
965    /// Deliberately synchronous. Every page/frame pair needed for the delta is
966    /// already staged by `record_appended_frames`, so this never scans the WAL
967    /// and never performs I/O; that keeps it callable from the synchronous
968    /// [`WalBackend::sync`] path without a runtime or `block_on`.
969    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            // No staged frame advances the horizon, which can only happen when
1032            // the published plane already covers `last_commit_frame` (an
1033            // extendable plane always contains the staged commit frame itself).
1034            // Rebuilding here would need a WAL scan, and this path must stay
1035            // synchronous, so leave visibility untouched. External refresh paths
1036            // retain the async rebuild for the cases that genuinely need it.
1037            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
1080/// Convert pager checkpoint mode to WAL checkpoint mode.
1081fn 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            // Reject at the earliest illegal transition: before `wal.refresh`,
1094            // before pinning `read_snapshot`, and before re-arming
1095            // `refresh_before_append`. Beginning a transaction on top of staged,
1096            // unpublished frames would otherwise leave a half-transition whose
1097            // only symptom is a later append failure.
1098            if self.has_pending_publication() {
1099                return Err(FrankenError::Busy);
1100            }
1101            // Establish a transaction-bounded snapshot once, instead of doing an
1102            // expensive refresh for every page read.
1103            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                // Refresh and synchronize the published base snapshot once before
1141                // the commit batch starts, then publish local frame deltas directly
1142                // from the append path.
1143                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            // Feed the frame to the FEC hook.  On commit, it encodes repair
1155            // symbols and stores them for later sidecar persistence.
1156            #[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                        // FEC encoding failure is non-fatal -- log and continue.
1171                        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                // Stage only: the frames are not durable until `sync`, so
1178                // publishing here would expose a commit a crash could erase.
1179                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                // Stage only: publication is deferred to the durability barrier.
1250                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                // Stage only: publication is deferred to the durability barrier.
1329                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        // Optimistically finalize against the adapter's current WAL state.
1386        // The append path still validates against both local and on-disk state
1387        // and will refresh/reseed if another writer advanced the append window.
1388        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                // Stage only: publication is deferred to the durability barrier.
1463                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                // Stage only: publication is deferred to the durability barrier.
1550                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            // Read the frame data at the resolved position.
1593            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            // Runtime integrity check: verify the frame actually contains our page.
1600            // This guards against index corruption or stale entries.
1601            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            // Strip the 24-byte frame header in place rather than
1612            // allocating a second page-sized Vec. Mirrors the fix in
1613            // `read_page_pinned` (`d9c410bb`): `frame_buf[HEADER..].to_vec()`
1614            // allocates a fresh 4 KiB buffer, memcpys the page payload into
1615            // it, then drops the original 4 KiB+24 B scratch — an alloc/free
1616            // round-trip on the hot WAL read path. Using `copy_within` +
1617            // `truncate` reuses the already-populated buffer: one memmove
1618            // (over the same bytes `to_vec` would have copied) and no new
1619            // allocation. `read_page` is the `&mut self` fallback path taken
1620            // when the caller does not hold a pinned snapshot — still hot
1621            // under mixed OLTP and write-path conflict resolution.
1622            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    // bd-db300.3.8.7: shared-lock read path for pinned snapshots.
1643    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            // Strip the 24-byte frame header in place instead of allocating
1683            // a fresh page-sized Vec. The pre-existing pattern did
1684            // `frame_buf[HEADER..].to_vec()` — on a 4 KiB page that
1685            // allocated a second 4 KiB buffer plus a 4 KiB memcpy and then
1686            // dropped the original 4 KiB+24 B frame_buf. On an MT pinned-
1687            // read workload every page served from the WAL paid that per-
1688            // read alloc/free round-trip; `_int_malloc` and `cfree` already
1689            // showed up in recent 2-thread profiles. Here we keep the
1690            // already-populated `frame_buf`, memmove the page bytes over
1691            // the header, truncate to `page_size`, and return it — one
1692            // allocation per read instead of two.
1693            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        // Durability first. Only once the frames are on stable storage may the
1851        // staged commit horizon become visible to readers.
1852        //
1853        // `refresh_before_append` is deliberately NOT set on the failure paths.
1854        // Setting it would let the next append run
1855        // `synchronize_publication_before_append`, which discards the preserved
1856        // batch and republishes straight from the WAL — reinstating exactly the
1857        // publish-before-fsync hazard this guard exists to prevent. Leaving it
1858        // clear keeps the staged batch intact for a later retry.
1859        self.wal.sync(cx, SyncFlags::NORMAL)?;
1860        self.publish_pending_after_sync(cx)?;
1861        // Re-arm the pre-append resynchronization only when nothing is staged.
1862        //
1863        // Syncing mid-transaction makes the appended frames durable but does not
1864        // commit them: with no commit marker yet, `publish_pending_after_sync`
1865        // correctly publishes nothing and the frames stay staged for the commit
1866        // still to come. Re-arming here would send the next append through
1867        // `synchronize_publication_before_append`, whose fail-closed guard would
1868        // then reject every further append — including the commit marker — and
1869        // strand the transaction permanently.
1870        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            // Fail closed BEFORE `wal.refresh` or any writer mutation. Checkpoint
1890            // backfills and may reset the WAL, and its inner paths can call
1891            // `invalidate_publication`, which discards the staged batch. Running
1892            // any of that against frames that were never fsynced would both lose
1893            // the staged horizon and risk backfilling non-durable frames, so the
1894            // batch must be drained by a successful sync first.
1895            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            // Refresh so planner state reflects the latest on-disk WAL shape.
1903            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            // Build checkpoint state for the planner.
1908            let state = CheckpointState {
1909                total_frames,
1910                backfilled_frames,
1911                oldest_reader_frame,
1912            };
1913
1914            // Wrap the CheckpointPageWriter in a CheckpointTargetAdapter.
1915            let mut target = CheckpointTargetAdapterRef { writer };
1916
1917            // Execute the checkpoint.
1918            let result =
1919                execute_checkpoint(cx, &mut self.wal, to_wal_mode(mode), state, &mut target)
1920                    .await?;
1921
1922            // Checkpoint-aware FEC lifecycle: once frames are backfilled to the
1923            // database file, their FEC symbols are no longer needed.  Clear
1924            // pending FEC results for the checkpointed range.
1925            #[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            // If the WAL was fully reset, also discard any buffered FEC pages
1939            // and invalidate the page index (salts changed).
1940            #[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        // A failed append can leave fewer bytes than the magic itself. Once a
2009        // strict record boundary anchors the suffix, those bytes are
2010        // unambiguously one incomplete append (including legacy one-byte
2011        // fault injections that predate the magic prefix).
2012        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    // A complete envelope is corruption, not a torn suffix. Strict decoding
2057    // gives a precise CRC/footer/version diagnostic. A valid complete record
2058    // here would mean more than one suffix record escaped the footer walk,
2059    // which is equally outside the one-torn-append recovery contract.
2060    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
2093/// WAL backend that can recover when the path-visible `-wal` sidecar is
2094/// removed or replaced while this process still owns an old file descriptor.
2095///
2096/// Real SQLite can checkpoint and unlink/reset `db-wal` when it does not know
2097/// about a live FrankenSQLite handle. `WalFile::refresh` is intentionally
2098/// descriptor-local, so it cannot notice that path-level mutation. This wrapper
2099/// performs a path probe before mutable WAL operations and swaps in a freshly
2100/// opened/created `WalFile` when the path-visible sidecar no longer matches the
2101/// open handle.
2102pub 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    /// Swap in a replacement WAL, discarding the previous adapter.
2151    ///
2152    /// Fails closed while the outgoing adapter still holds a staged batch: the
2153    /// replacement would consume away the pending metadata, and the freshly
2154    /// wrapped adapter would republish those frames from the WAL without knowing
2155    /// they were never fsynced (GH #187). A successful sync must drain the batch
2156    /// before a path-visible replacement can proceed.
2157    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        // Random salts (GH #201): the replacement WAL must reject frames
2174        // from the file it replaces.
2175        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    /// Revalidate conflict candidates across a WAL-generation transition.
2232    ///
2233    /// A stock SQLite reader may checkpoint and replace an otherwise
2234    /// unchanged WAL while a FrankenSQLite transaction is open. Treating the
2235    /// generation change itself as a write conflict produces a false
2236    /// `BusySnapshot`. Conversely, blindly accepting the new generation can
2237    /// overwrite a real external commit that was checkpointed into the main
2238    /// database. The only safe admission proof is therefore page-specific:
2239    /// every candidate must have a transaction-snapshot baseline, and its
2240    /// latest committed full-page image (new WAL first, main DB otherwise)
2241    /// must hash identically.
2242    ///
2243    /// Any missing/ambiguous baseline, unreadable or short main page, invalid
2244    /// database header, page-size change, WAL read error, or close failure
2245    /// fails closed by returning every candidate as conflicting.
2246    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        // Validate the current main-database header before trusting offsets.
2290        // SQLite encodes a 64 KiB page as the u16 value 1.
2291        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(&current_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    /// Return a safe append boundary, repairing exactly one validated torn
2484    /// suffix in place.
2485    ///
2486    /// The caller must hold the database's external writer or maintenance
2487    /// gate for the whole scan/truncate/append sequence. Keeping this helper
2488    /// private prevents a scan/reopen race from becoming part of the API.
2489    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        // The EOF footer was not a complete valid record. Locate at most one
2535        // complete anchor plus one maximum-sized suffix, using exact footer
2536        // boundaries rather than a free-form magic scan.
2537        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            // A VFS may report a failed write after changing a prefix of the
2690            // destination. Restore the append boundary under a masked child
2691            // context so a cooperative cancellation cannot leave a torn tail
2692            // when the failed future itself is allowed to finish.
2693            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        // Match the WAL's configured synchronous policy exactly. Even when
2708        // `sync` is false, this ordered sidecar write precedes the WAL marker;
2709        // neither write then claims power-loss-stable persistence.
2710        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                // The original write may have created the sidecar but been
2776                // dropped before its directory entry was fenced.
2777                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        // This fence is written before the checkpoint is allowed to reset the
2827        // WAL generation. Once the in-place replacement starts, finish it
2828        // under a cancellation mask; if any stage fails, the checkpoint
2829        // returns before reset and the old WAL remains authoritative.
2830        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            // Healthy operation is O(1): the final four bytes identify the
2981            // exact newest record, so only its footer and bytes are read.
2982            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                    // A matching magic or self-declared length makes this a
3019                    // fully-present envelope candidate. Strict decoding is
3020                    // mandatory even when its magic/version/CRC/footer is
3021                    // corrupt; complete corruption is never a torn suffix.
3022                    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                // Invalid EOF footer means the last append may have torn.
3034                // Search only footer-derived candidates within one maximum
3035                // suffix, retaining enough preceding bytes for one maximum
3036                // anchor record. Magic is only a cheap validation after a
3037                // candidate footer establishes an exact boundary; it is never
3038                // used as a free-form scan key.
3039                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                // Append order is generation order. Once the newest tail
3131                // belongs to a prior reset generation, no earlier sidecar
3132                // record can authorize the current WAL; checkpoint clock
3133                // continuation comes from the fixed handoff anchor instead.
3134                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            // Only after the live WAL shape is classified as the exact
3479            // pre-interval prefix may reconciliation repair torn sidecar bytes
3480            // or remove the matching orphan certificate. Unexpected WAL state
3481            // preserves all durable evidence for diagnosis and retry.
3482            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                // Fence the certificate clock before the checkpoint is
3594                // allowed to reset the WAL generation. Replacing the handoff
3595                // is intentionally non-authoritative while the old WAL and
3596                // sidecar remain reconstructible: a crash, cancellation, or
3597                // write failure here aborts the checkpoint without destroying
3598                // the previous generation's source of truth.
3599                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
3611/// Adapter wrapping a `&mut dyn CheckpointPageWriter` to implement `CheckpointTarget`.
3612///
3613/// This is used internally by `WalBackendAdapter::checkpoint` to bridge the
3614/// pager's writer to the WAL executor's target trait.
3615struct 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// ---------------------------------------------------------------------------
3639// Tests
3640// ---------------------------------------------------------------------------
3641
3642#[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 the next sync on a non-handoff (i.e. WAL) file.
3676        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        /// Arm a one-shot sync failure on the WAL file itself.
3716        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    /// Deliberate no-op (frankensqlite#299).
3967    ///
3968    /// This helper previously installed a process-global `TRACE` subscriber via
3969    /// `tracing_subscriber::fmt()...with_test_writer().try_init()`. `try_init()`
3970    /// is process-wide and first-caller-wins, so the first of the 9 callers
3971    /// changed tracing enablement — and libtest output capture — for every
3972    /// unrelated test running afterwards in this binary, making a later failure
3973    /// replay the whole captured global trace stream.
3974    ///
3975    /// `fsqlite-core` already fixed the identical pattern in b262b6a6 for its
3976    /// other helpers; this one was missed. No caller here asserts on emitted
3977    /// trace events, so the body is simply removed and the call sites are kept
3978    /// so the diff stays test-only.
3979    ///
3980    /// See `wal_publication_tracing_helper_installs_no_global_subscriber`.
3981    fn init_wal_publication_test_tracing() {}
3982
3983    /// frankensqlite#299 regression: the WAL publication tracing helper must not
3984    /// install, or otherwise disturb, a process-global subscriber.
3985    ///
3986    /// Only the equality assertion is made, deliberately. Unlike the pager
3987    /// crate, this test binary contains another global-subscriber installation
3988    /// site outside this file, so an absolute `!has_been_set()` assertion would
3989    /// be order-dependent and could fail for reasons unrelated to this helper.
3990    /// Comparing dispatcher state across the call is untaintable and proves the
3991    /// exact property under test: that this helper is inert.
3992    #[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        // Simulate a crash after certificate durability but before its WAL
5078        // commit marker. Bounded tail lookup must step over this well-formed
5079        // orphan and recover the preceding authorized seed.
5080        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    /// Adapter backed by the fault VFS so WAL `sync` failures can be injected.
5212    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    // -- WalBackendAdapter tests --
5225
5226    #[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        // Write page 5 twice -- the adapter should return the latest.
5290        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        // Ensure the second writer appended to frame 1 (not frame 0 overwrite).
5347        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        // Same transaction snapshot must stay stable (no mid-transaction drift).
5874        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        // A new transaction snapshot should pick up the latest commit.
5889        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        // Verify it can be used as a trait object.
5995        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    // -- Page index O(1) lookup tests --
6050
6051    #[test]
6052    fn test_page_index_returns_correct_data() {
6053        // Write several pages, verify O(1) index returns the right data.
6054        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        // All three pages should be readable via the index.
6069        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        // Non-existent page returns None.
6074        assert_eq!(adapter.read_page(&cx, 99).expect("read"), None);
6075    }
6076
6077    #[test]
6078    fn test_page_index_returns_latest_version() {
6079        // Write the same page twice; the index should point to the newer frame.
6080        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        // Simulate a WAL reset with new salts. The index must be rebuilt so
6104        // stale entries from the old generation are not returned.
6105        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        // Read page 1 to populate the index.
6115        assert_eq!(adapter.read_page(&cx, 1).expect("read old"), Some(old_data));
6116
6117        // Reset WAL with new salts (simulates checkpoint reset).
6118        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        // Write new data for the same page number in the new generation.
6129        let new_data = sample_page(0x22);
6130        adapter
6131            .append_frame(&cx, 1, &new_data, 1)
6132            .expect("append new generation commit");
6133
6134        // The index must have been invalidated; we should get the new data.
6135        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        // A page that existed only in the old generation should be gone.
6143        let old_only = sample_page(0x33);
6144        // (We never wrote page 99 in the new generation.)
6145        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        // Suppress unused variable warning.
6151        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        // Generation identity must include checkpoint_seq. Reusing salts across
6158        // reset must still invalidate the cached page index and avoid ABA bugs.
6159        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        // Verify that the index extends incrementally once each commit crosses
6242        // the durable-sync publication barrier.
6243        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        // First read builds the index.
6254        assert_eq!(
6255            adapter.read_page(&cx, 1).expect("read"),
6256            Some(page1.clone())
6257        );
6258
6259        // Append more committed frames.
6260        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        // Reading should trigger incremental extend, not full rebuild.
6271        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    /// Frames for a two-page commit batch, the second frame carrying the commit.
6280    fn commit_batch_pages() -> (Vec<u8>, Vec<u8>) {
6281        (sample_page(0x71), sample_page(0x72))
6282    }
6283
6284    /// Assert no commit horizon has been published yet.
6285    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        // Retry: the same staged batch publishes once durability succeeds.
6518        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        // A further append must not run the pre-append resynchronization, which
6564        // would discard the preserved batch and republish the unsynced horizon.
6565        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        // Durability finally succeeds: the whole preserved batch publishes.
6582        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        // `begin_transaction` must reject at the earliest illegal transition,
6615        // before refreshing, pinning a read snapshot, or re-arming the
6616        // pre-append guard — and it must be a retryable Busy, not corruption.
6617        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        // Defense in depth: the pre-append choke guard still refuses for any
6636        // other path that re-arms `refresh_before_append`.
6637        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        // The batch is still recoverable: a successful sync publishes it.
6659        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        // Checkpoint must refuse before touching the WAL: it backfills, may
6687        // reset, and can invalidate the publication plane, all of which would
6688        // destroy the staged batch.
6689        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        // Retry: durability succeeds and the preserved batch publishes.
6715        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        // Append a non-commit frame, then sync. The frame becomes durable but is
6736        // not committed, so nothing may be published.
6737        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        // Continuation must remain possible: the commit marker still lands.
6753        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        // `expect_err` would require `WalFile: Debug`, which the fault-VFS file
6810        // type does not implement, so assert on the pattern directly.
6811        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        // Once drained, the escape hatch opens again.
6822        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        // An explicit refresh must not publish frames this handle has staged but
6843        // not yet made durable.
6844        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        // Publication is deferred to the durability barrier (#187); the commit
6912        // horizon only becomes visible once `sync` persists the frames.
6913        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        // Publication is deferred to the durability barrier (#187).
6965        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        // Publication is deferred to the durability barrier (#187), and
7060        // checkpoint now fails closed while a batch is staged, so the batch must
7061        // be drained before checkpointing.
7062        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    // -- Partial index fallback tests --
7086
7087    #[test]
7088    fn test_partial_index_falls_back_to_linear_scan() {
7089        init_wal_publication_test_tracing();
7090        // Verify that when the page index cap is hit, pages that weren't
7091        // indexed are still found via the backwards linear scan fallback.
7092        let cx = test_cx();
7093        let vfs = MemoryVfs::new();
7094        let mut adapter = make_adapter(&vfs, &cx);
7095
7096        // Set a very small cap so we can trigger the partial-index path
7097        // with just a handful of frames.
7098        adapter.set_page_index_cap(2);
7099
7100        // Write 5 distinct pages.  With a cap of 2, only the first 2 unique
7101        // pages will be indexed; pages 3-5 will be dropped from the index.
7102        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        // Pages 1 and 2 should be in the index (fast path).
7117        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        // Pages 3-5 were NOT indexed, but must still be found via the
7129        // backwards linear scan fallback.
7130        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        // A page that was never written should still return None.
7147        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        // Verify the index was indeed marked partial.
7154        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        // When the same page appears multiple times and overflows the index,
7163        // the backwards scan must return the LATEST (highest frame index)
7164        // version, not the first one it encounters in a forward scan.
7165        let cx = test_cx();
7166        let vfs = MemoryVfs::new();
7167        let mut adapter = make_adapter(&vfs, &cx);
7168
7169        // Cap at 1 so only page 1 fits in the index.
7170        adapter.set_page_index_cap(1);
7171
7172        let old_p2 = sample_page(0xAA);
7173        let new_p2 = sample_page(0xBB);
7174
7175        // Frame 0: page 1 (indexed)
7176        adapter
7177            .append_frame(&cx, 1, &sample_page(0x01), 0)
7178            .expect("append p1");
7179        // Frame 1: page 2 old version (NOT indexed -- cap exceeded)
7180        adapter
7181            .append_frame(&cx, 2, &old_p2, 0)
7182            .expect("append p2 old");
7183        // Frame 2: page 2 new version (NOT indexed -- cap exceeded, and
7184        // page 2 is not already in the index so it won't be updated)
7185        adapter
7186            .append_frame(&cx, 2, &new_p2, 3)
7187            .expect("append p2 new (commit)");
7188
7189        // The backwards scan from frame 2 should find the newest version first.
7190        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    // -- CheckpointTargetAdapterRef tests --
7378
7379    #[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        // Verify it can be used as a trait object.
7424        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}