1use std::collections::HashMap;
16use std::future::Future;
17use std::pin::Pin;
18
19use crate::pager::SimpleTransaction;
20use fsqlite_error::{FrankenError, Result};
21use fsqlite_types::cx::Cx;
22use fsqlite_types::{CommitSeq, PageData, PageNumber, PageSize};
23#[cfg(all(feature = "native", target_os = "linux"))]
24use fsqlite_vfs::IoUringVfs;
25#[cfg(all(feature = "native", unix))]
26use fsqlite_vfs::UnixVfs;
27#[cfg(all(feature = "native", target_os = "windows"))]
28use fsqlite_vfs::WindowsVfs;
29use fsqlite_vfs::{MemoryVfs, VfsWriteCompletion};
30use fsqlite_wal::{
31 ParallelWalCommitCertificate, TransactionConflictPageBaseline, TransactionConflictSnapshot,
32 WalGenerationIdentity, checksum::WalChecksumTransform,
33};
34
35pub(crate) mod sealed {
42 pub trait Sealed {}
44}
45
46#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
60pub enum JournalMode {
61 #[default]
64 Delete,
65 Wal,
69}
70
71#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
84pub enum CheckpointMode {
85 #[default]
88 Passive,
89 Full,
92 Restart,
94 Truncate,
96}
97
98#[derive(Debug, Clone, PartialEq, Eq)]
100pub struct CheckpointResult {
101 pub total_frames: u32,
103 pub frames_backfilled: u32,
105 pub completed: bool,
107 pub wal_was_reset: bool,
109 pub requested_mode: CheckpointMode,
111 pub effective_mode: CheckpointMode,
114}
115
116#[derive(Debug, Clone, Copy, PartialEq, Eq)]
121pub struct WalPublicationSnapshot {
122 pub publication_seq: u64,
124 pub generation: WalGenerationIdentity,
126 pub last_commit_frame: Option<usize>,
128 pub commit_count: u64,
130 pub latest_frame_entries: usize,
132 pub index_is_partial: bool,
134}
135
136impl WalPublicationSnapshot {
137 #[must_use]
138 pub const fn lookup_contract_is_authoritative(self) -> bool {
139 !self.index_is_partial
140 }
141}
142
143#[derive(Debug, Clone, Copy, PartialEq, Eq)]
151pub struct WalLogicalReadSnapshot {
152 pub generation: WalGenerationIdentity,
154 pub last_commit_frame: Option<usize>,
156 pub visible_commit_seq: CommitSeq,
158}
159
160#[derive(Debug, Clone, Copy, PartialEq, Eq)]
162pub enum ParallelWalCommitReconciliation {
163 Authorized,
166 NotCommitted,
168}
169
170pub type WalFuture<'a, T> = Pin<Box<dyn Future<Output = Result<T>> + Send + 'a>>;
179
180struct WalTrackedCompletionGuard(VfsWriteCompletion);
187
188impl WalTrackedCompletionGuard {
189 fn complete_success(&self) {
190 self.0.complete_success();
191 }
192
193 fn complete_error(&self) {
194 self.0.complete_error();
195 }
196}
197
198impl Drop for WalTrackedCompletionGuard {
199 fn drop(&mut self) {
200 self.0.complete_error();
201 }
202}
203
204pub trait WalBackend: Send + Sync {
205 fn begin_transaction<'a>(&'a mut self, _cx: &'a Cx) -> WalFuture<'a, ()> {
210 Box::pin(async { Ok(()) })
211 }
212
213 #[must_use]
218 fn published_snapshot(&self) -> Option<WalPublicationSnapshot> {
219 None
220 }
221
222 #[must_use]
227 fn pinned_read_snapshot(&self) -> Option<WalPublicationSnapshot> {
228 None
229 }
230
231 fn pinned_logical_read_snapshot<'a>(
238 &'a self,
239 _cx: &'a Cx,
240 ) -> WalFuture<'a, Option<WalLogicalReadSnapshot>> {
241 Box::pin(async { Ok(None) })
242 }
243
244 fn refresh_published_snapshot<'a>(
250 &'a mut self,
251 _cx: &'a Cx,
252 ) -> WalFuture<'a, Option<WalPublicationSnapshot>> {
253 Box::pin(async { Ok(self.published_snapshot()) })
254 }
255
256 fn publish_authorized_deferred_commit<'a>(&'a mut self, _cx: &'a Cx) -> WalFuture<'a, ()> {
264 Box::pin(async { Ok(()) })
265 }
266
267 fn append_frame<'a>(
274 &'a mut self,
275 cx: &'a Cx,
276 page_number: u32,
277 page_data: &'a [u8],
278 db_size_if_commit: u32,
279 ) -> WalFuture<'a, ()>;
280
281 fn append_frames<'a>(
286 &'a mut self,
287 cx: &'a Cx,
288 frames: &'a [WalFrameRef<'a>],
289 ) -> WalFuture<'a, ()> {
290 Box::pin(async move {
291 for frame in frames {
292 self.append_frame(
293 cx,
294 frame.page_number,
295 frame.page_data,
296 frame.db_size_if_commit,
297 )
298 .await?;
299 }
300 Ok(())
301 })
302 }
303
304 fn append_frames_tracked<'a>(
313 &'a mut self,
314 cx: &'a Cx,
315 frames: &'a [WalFrameRef<'a>],
316 completion: VfsWriteCompletion,
317 ) -> WalFuture<'a, ()> {
318 let completion = WalTrackedCompletionGuard(completion);
319 Box::pin(async move {
320 let result = self.append_frames(cx, frames).await;
321 if result.is_ok() {
322 completion.complete_success();
323 } else {
324 completion.complete_error();
325 }
326 result
327 })
328 }
329
330 fn prepare_append_frames(
336 &self,
337 _frames: &[WalFrameRef<'_>],
338 ) -> Result<Option<PreparedWalFrameBatch>> {
339 Ok(None)
340 }
341
342 fn finalize_prepared_frames(
349 &self,
350 _cx: &Cx,
351 _prepared: &mut PreparedWalFrameBatch,
352 ) -> Result<()> {
353 Ok(())
354 }
355
356 fn append_prepared_frames<'a>(
362 &'a mut self,
363 cx: &'a Cx,
364 prepared: &'a mut PreparedWalFrameBatch,
365 ) -> WalFuture<'a, ()> {
366 Box::pin(async move {
367 for index in 0..prepared.frame_count() {
368 let meta = prepared.frame_metas[index];
369 self.append_frame(
370 cx,
371 meta.page_number,
372 prepared.page_data(index),
373 meta.db_size_if_commit,
374 )
375 .await?;
376 }
377 Ok(())
378 })
379 }
380
381 fn append_prepared_frames_tracked<'a>(
383 &'a mut self,
384 cx: &'a Cx,
385 prepared: &'a mut PreparedWalFrameBatch,
386 completion: VfsWriteCompletion,
387 ) -> WalFuture<'a, ()> {
388 let completion = WalTrackedCompletionGuard(completion);
389 Box::pin(async move {
390 let result = self.append_prepared_frames(cx, prepared).await;
391 if result.is_ok() {
392 completion.complete_success();
393 } else {
394 completion.complete_error();
395 }
396 result
397 })
398 }
399
400 fn persist_parallel_wal_commit_certificate<'a>(
415 &'a mut self,
416 _cx: &'a Cx,
417 _certificate: &'a ParallelWalCommitCertificate,
418 _wal_frame_start: u64,
419 _wal_frame_end: u64,
420 _sync: bool,
421 ) -> WalFuture<'a, ()> {
422 Box::pin(async { Err(FrankenError::Unsupported) })
423 }
424
425 fn persist_parallel_wal_commit_certificate_tracked<'a>(
428 &'a mut self,
429 cx: &'a Cx,
430 certificate: &'a ParallelWalCommitCertificate,
431 wal_frame_start: u64,
432 wal_frame_end: u64,
433 sync: bool,
434 completion: VfsWriteCompletion,
435 ) -> WalFuture<'a, ()> {
436 let completion = WalTrackedCompletionGuard(completion);
437 Box::pin(async move {
438 let result = self
439 .persist_parallel_wal_commit_certificate(
440 cx,
441 certificate,
442 wal_frame_start,
443 wal_frame_end,
444 sync,
445 )
446 .await;
447 if result.is_ok() {
448 completion.complete_success();
449 } else {
450 completion.complete_error();
451 }
452 result
453 })
454 }
455
456 fn reconcile_parallel_wal_commit<'a>(
467 &'a mut self,
468 _cx: &'a Cx,
469 _certificate: &'a ParallelWalCommitCertificate,
470 _wal_frame_start: u64,
471 _wal_frame_end: u64,
472 _sync: bool,
473 ) -> WalFuture<'a, ParallelWalCommitReconciliation> {
474 Box::pin(async { Err(FrankenError::Unsupported) })
475 }
476
477 fn latest_authorized_parallel_wal_commit_certificate<'a>(
489 &'a mut self,
490 _cx: &'a Cx,
491 ) -> WalFuture<'a, Option<ParallelWalCommitCertificate>> {
492 Box::pin(async { Ok(None) })
493 }
494
495 fn read_page<'a>(&'a mut self, cx: &'a Cx, page_number: u32) -> WalFuture<'a, Option<Vec<u8>>>;
502
503 fn read_page_at_appended_tail<'a>(
520 &'a mut self,
521 cx: &'a Cx,
522 page_number: u32,
523 ) -> WalFuture<'a, Option<Vec<u8>>> {
524 self.read_page(cx, page_number)
525 }
526
527 fn read_page_pinned<'a>(
540 &'a self,
541 _cx: &'a Cx,
542 _page_number: u32,
543 ) -> WalFuture<'a, Option<Vec<u8>>> {
544 Box::pin(async {
545 Err(FrankenError::internal(
548 "read_page_pinned not supported by this WalBackend; use read_page",
549 ))
550 })
551 }
552
553 fn supports_pinned_reads(&self) -> bool {
557 false
558 }
559
560 fn committed_txns_since_page<'a>(
567 &'a mut self,
568 _cx: &'a Cx,
569 _page_number: u32,
570 ) -> WalFuture<'a, u64> {
571 Box::pin(async { Ok(0) })
572 }
573
574 fn conflicting_pages_since_snapshot<'a>(
583 &'a mut self,
584 _cx: &'a Cx,
585 _snapshot: TransactionConflictSnapshot,
586 _page_numbers: &'a [u32],
587 _page_baselines: &'a [TransactionConflictPageBaseline],
588 ) -> WalFuture<'a, Vec<u32>> {
589 Box::pin(async { Ok(Vec::new()) })
590 }
591
592 fn committed_txn_count<'a>(&'a mut self, _cx: &'a Cx) -> WalFuture<'a, u64> {
599 Box::pin(async { Ok(0) })
600 }
601
602 fn sync(&mut self, cx: &Cx) -> Result<()>;
604
605 fn frame_count(&self) -> usize;
607
608 fn checkpoint<'a>(
625 &'a mut self,
626 cx: &'a Cx,
627 mode: CheckpointMode,
628 writer: &'a mut dyn CheckpointPageWriter,
629 backfilled_frames: u32,
630 oldest_reader_frame: Option<u32>,
631 ) -> WalFuture<'a, CheckpointResult>;
632}
633
634#[derive(Debug, Clone, Copy)]
636pub struct WalFrameRef<'a> {
637 pub page_number: u32,
639 pub page_data: &'a [u8],
641 pub db_size_if_commit: u32,
643}
644
645#[derive(Debug, Clone, Copy, PartialEq, Eq)]
647pub struct PreparedWalFrameMeta {
648 pub page_number: u32,
650 pub db_size_if_commit: u32,
652}
653
654pub type PreparedWalChecksumTransform = WalChecksumTransform;
659
660#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
662pub struct PreparedWalChecksumSeed {
663 pub s1: u32,
665 pub s2: u32,
667}
668
669#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
674pub struct PreparedWalFinalizationState {
675 pub checkpoint_seq: u32,
677 pub salt1: u32,
679 pub salt2: u32,
681 pub start_frame_index: usize,
683 pub seed: PreparedWalChecksumSeed,
685}
686
687#[derive(Debug, Clone, PartialEq, Eq)]
689pub struct PreparedWalFrameBatch {
690 pub frame_size: usize,
692 pub page_data_offset: usize,
694 pub big_endian_checksum: bool,
696 pub frame_metas: Vec<PreparedWalFrameMeta>,
698 pub checksum_transforms: Vec<PreparedWalChecksumTransform>,
700 pub frame_bytes: Vec<u8>,
702 pub last_commit_frame_offset: Option<usize>,
704 pub finalized_for: Option<PreparedWalFinalizationState>,
706 pub finalized_running_checksum: Option<PreparedWalChecksumSeed>,
708}
709
710impl PreparedWalFrameBatch {
711 #[must_use]
713 pub fn frame_count(&self) -> usize {
714 self.frame_metas.len()
715 }
716
717 #[must_use]
719 pub fn page_size(&self) -> usize {
720 self.frame_size.saturating_sub(self.page_data_offset)
721 }
722
723 #[must_use]
725 pub fn frame_refs(&self) -> Vec<WalFrameRef<'_>> {
726 self.frame_metas
727 .iter()
728 .enumerate()
729 .map(|(index, meta)| {
730 let frame_start = index * self.frame_size;
731 let page_start = frame_start + self.page_data_offset;
732 let page_end = frame_start + self.frame_size;
733 WalFrameRef {
734 page_number: meta.page_number,
735 page_data: &self.frame_bytes[page_start..page_end],
736 db_size_if_commit: meta.db_size_if_commit,
737 }
738 })
739 .collect()
740 }
741
742 #[must_use]
744 pub fn page_data(&self, index: usize) -> &[u8] {
745 let frame_start = index * self.frame_size;
746 let page_start = frame_start + self.page_data_offset;
747 let page_end = frame_start + self.frame_size;
748 &self.frame_bytes[page_start..page_end]
749 }
750
751 #[must_use]
753 pub fn frame_slice(&self, index: usize) -> &[u8] {
754 let frame_start = index * self.frame_size;
755 let frame_end = frame_start + self.frame_size;
756 &self.frame_bytes[frame_start..frame_end]
757 }
758
759 pub fn set_db_size_if_commit(&mut self, index: usize, db_size_if_commit: u32) {
761 self.frame_metas[index].db_size_if_commit = db_size_if_commit;
762 let frame_start = index * self.frame_size;
763 let db_size_offset = frame_start + 4;
764 self.frame_bytes[db_size_offset..db_size_offset + 4]
765 .copy_from_slice(&db_size_if_commit.to_be_bytes());
766 self.finalized_for = None;
767 self.finalized_running_checksum = None;
768 }
769
770 pub fn recompute_checksum_transforms(&mut self) -> Result<()> {
772 let page_size = self.page_size();
773 self.checksum_transforms = (0..self.frame_count())
774 .map(|index| {
775 WalChecksumTransform::for_wal_frame(
776 self.frame_slice(index),
777 page_size,
778 self.big_endian_checksum,
779 )
780 })
781 .collect::<Result<Vec<_>>>()?;
782 self.finalized_for = None;
783 self.finalized_running_checksum = None;
784 Ok(())
785 }
786}
787
788#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
797pub enum TransactionMode {
798 #[default]
801 Deferred,
802 Immediate,
806 Exclusive,
809 Concurrent,
816 ReadOnly,
819}
820
821pub trait MvccPager: sealed::Sealed + Send + Sync {
842 type Txn: TransactionHandle;
844
845 fn begin<'a>(
851 &'a self,
852 cx: &'a Cx,
853 mode: TransactionMode,
854 ) -> impl Future<Output = Result<Self::Txn>> + 'a;
855
856 fn journal_mode(&self) -> JournalMode;
858
859 fn is_readonly(&self) -> bool;
861
862 fn set_journal_mode<'a>(
870 &'a self,
871 cx: &'a Cx,
872 mode: JournalMode,
873 ) -> impl Future<Output = Result<JournalMode>> + 'a;
874
875 fn set_wal_backend(&self, backend: Box<dyn WalBackend>) -> Result<()>;
880}
881
882#[derive(Debug, Clone, Copy, PartialEq, Eq)]
894pub enum PagerCommitState {
895 NotCommitted,
897 InDoubt,
899 DurableNeedsPublication,
901 Committed,
903}
904
905impl PagerCommitState {
906 #[must_use]
908 pub const fn retains_commit_obligation(self) -> bool {
909 !matches!(self, Self::NotCommitted)
910 }
911}
912
913pub trait TransactionHandle: sealed::Sealed + Send {
928 fn get_page<'a>(
934 &'a self,
935 cx: &'a Cx,
936 page_no: PageNumber,
937 ) -> impl Future<Output = Result<PageData>> + 'a;
938
939 fn prefetch_page_hint(&self, _cx: &Cx, _page_no: PageNumber) {}
944
945 fn write_page<'a>(
950 &'a mut self,
951 cx: &'a Cx,
952 page_no: PageNumber,
953 data: &'a [u8],
954 ) -> impl Future<Output = Result<()>> + 'a;
955
956 fn write_page_data<'a>(
961 &'a mut self,
962 cx: &'a Cx,
963 page_no: PageNumber,
964 data: PageData,
965 ) -> impl Future<Output = Result<()>> + 'a {
966 async move { self.write_page(cx, page_no, data.as_bytes()).await }
967 }
968
969 fn try_take_staged_page_data(&mut self, _page_no: PageNumber) -> Option<PageData> {
976 None
977 }
978
979 fn try_mutate_staged_page_data(
985 &mut self,
986 _page_no: PageNumber,
987 _f: &mut dyn FnMut(&mut PageData),
988 ) -> bool {
989 false
990 }
991
992 fn restore_staged_page_data<'a>(
998 &'a mut self,
999 cx: &'a Cx,
1000 page_no: PageNumber,
1001 data: PageData,
1002 ) -> impl Future<Output = Result<()>> + 'a {
1003 async move { self.write_page_data(cx, page_no, data).await }
1004 }
1005
1006 fn allocate_page<'a>(&'a mut self, cx: &'a Cx)
1010 -> impl Future<Output = Result<PageNumber>> + 'a;
1011
1012 fn free_page<'a>(
1014 &'a mut self,
1015 cx: &'a Cx,
1016 page_no: PageNumber,
1017 ) -> impl Future<Output = Result<()>> + 'a;
1018
1019 fn commit<'a>(&'a mut self, cx: &'a Cx) -> impl Future<Output = Result<()>> + 'a;
1025
1026 fn pager_commit_state(&self) -> PagerCommitState {
1032 PagerCommitState::NotCommitted
1033 }
1034
1035 fn commit_and_retain<'a>(&'a mut self, cx: &'a Cx) -> impl Future<Output = Result<bool>> + 'a {
1051 async move {
1052 self.commit(cx).await?;
1053 Ok(false)
1054 }
1055 }
1056
1057 fn is_writer(&self) -> bool;
1063
1064 fn has_pending_writes(&self) -> bool;
1069
1070 fn published_visible_commit_seq_hint(&self) -> Option<fsqlite_types::CommitSeq> {
1076 None
1077 }
1078
1079 fn pending_commit_pages(&self) -> Result<Vec<PageNumber>> {
1083 Ok(Vec::new())
1084 }
1085
1086 fn pending_conflict_pages(&self) -> Result<Vec<PageNumber>> {
1094 self.pending_commit_pages()
1095 }
1096
1097 fn pending_conflict_pages_conservative(&self) -> Vec<PageNumber> {
1112 self.write_set_page_numbers()
1113 }
1114
1115 fn write_set_page_numbers(&self) -> Vec<PageNumber> {
1118 Vec::new()
1119 }
1120
1121 fn page_one_in_pending_commit_surface(&self) -> Result<bool> {
1124 Ok(self.pending_commit_pages()?.contains(&PageNumber::ONE))
1125 }
1126
1127 fn page_size(&self) -> PageSize {
1132 PageSize::default()
1133 }
1134
1135 fn allocate_page_requires_page_one_conflict_tracking(&self) -> Result<bool> {
1144 Ok(true)
1145 }
1146
1147 fn free_page_requires_page_one_conflict_tracking(&self, _page_no: PageNumber) -> Result<bool> {
1156 Ok(true)
1157 }
1158
1159 fn write_page_requires_page_one_conflict_tracking(&self, _page_no: PageNumber) -> Result<bool> {
1169 Ok(true)
1170 }
1171
1172 fn rollback<'a>(&'a mut self, cx: &'a Cx) -> impl Future<Output = Result<()>> + 'a;
1178
1179 fn record_write_witness(&mut self, _cx: &Cx, _key: fsqlite_types::WitnessKey) {}
1184
1185 fn savepoint(&mut self, cx: &Cx, name: &str) -> Result<()>;
1191
1192 fn release_savepoint(&mut self, cx: &Cx, name: &str) -> Result<()>;
1198
1199 fn rollback_to_savepoint(&mut self, cx: &Cx, name: &str) -> Result<()>;
1206}
1207
1208pub trait CheckpointPageWriter: sealed::Sealed + Send {
1222 fn write_page<'a>(
1224 &'a mut self,
1225 cx: &'a Cx,
1226 page_no: PageNumber,
1227 data: &'a [u8],
1228 ) -> WalFuture<'a, ()>;
1229
1230 fn truncate<'a>(&'a mut self, cx: &'a Cx, n_pages: u32) -> WalFuture<'a, ()>;
1232
1233 fn sync<'a>(&'a mut self, cx: &'a Cx) -> WalFuture<'a, ()>;
1235}
1236
1237#[derive(Debug, Default, Clone, Copy)]
1243pub struct MockMvccPager;
1244
1245impl sealed::Sealed for MockMvccPager {}
1246
1247impl MvccPager for MockMvccPager {
1248 type Txn = MockTransaction;
1249
1250 fn begin<'a>(
1251 &'a self,
1252 _cx: &'a Cx,
1253 _mode: TransactionMode,
1254 ) -> impl Future<Output = Result<Self::Txn>> + 'a {
1255 async {
1256 Ok(MockTransaction {
1257 committed: false,
1258 next_page: 2,
1259 savepoint_names: Vec::new(),
1260 })
1261 }
1262 }
1263
1264 fn journal_mode(&self) -> JournalMode {
1265 JournalMode::Delete
1266 }
1267
1268 fn is_readonly(&self) -> bool {
1269 false
1270 }
1271
1272 fn set_journal_mode<'a>(
1273 &'a self,
1274 _cx: &'a Cx,
1275 mode: JournalMode,
1276 ) -> impl Future<Output = Result<JournalMode>> + 'a {
1277 async move { Ok(mode) }
1278 }
1279
1280 fn set_wal_backend(&self, _backend: Box<dyn WalBackend>) -> Result<()> {
1281 Ok(())
1282 }
1283}
1284
1285#[derive(Debug, Clone)]
1287pub struct MockTransaction {
1288 committed: bool,
1289 next_page: u32,
1290 savepoint_names: Vec<String>,
1291}
1292
1293impl sealed::Sealed for MockTransaction {}
1294
1295impl TransactionHandle for MockTransaction {
1296 fn get_page<'a>(
1297 &'a self,
1298 _cx: &'a Cx,
1299 page_no: PageNumber,
1300 ) -> impl Future<Output = Result<PageData>> + 'a {
1301 async move {
1302 let size = fsqlite_types::PageSize::default();
1303 let mut data = PageData::zeroed(size);
1304 data.as_bytes_mut()[..4].copy_from_slice(&page_no.get().to_le_bytes());
1305 Ok(data)
1306 }
1307 }
1308
1309 fn write_page<'a>(
1310 &'a mut self,
1311 _cx: &'a Cx,
1312 _page_no: PageNumber,
1313 _data: &'a [u8],
1314 ) -> impl Future<Output = Result<()>> + 'a {
1315 async { Ok(()) }
1316 }
1317
1318 fn allocate_page<'a>(
1319 &'a mut self,
1320 _cx: &'a Cx,
1321 ) -> impl Future<Output = Result<PageNumber>> + 'a {
1322 async move {
1323 let page = PageNumber::new(self.next_page)
1324 .expect("mock allocator must always produce non-zero page numbers");
1325 self.next_page += 1;
1326 Ok(page)
1327 }
1328 }
1329
1330 fn free_page<'a>(
1331 &'a mut self,
1332 _cx: &'a Cx,
1333 _page_no: PageNumber,
1334 ) -> impl Future<Output = Result<()>> + 'a {
1335 async { Ok(()) }
1336 }
1337
1338 fn commit<'a>(&'a mut self, _cx: &'a Cx) -> impl Future<Output = Result<()>> + 'a {
1339 async move {
1340 self.committed = true;
1341 Ok(())
1342 }
1343 }
1344
1345 fn pager_commit_state(&self) -> PagerCommitState {
1346 if self.committed {
1347 PagerCommitState::Committed
1348 } else {
1349 PagerCommitState::NotCommitted
1350 }
1351 }
1352
1353 fn is_writer(&self) -> bool {
1354 false
1355 }
1356
1357 fn has_pending_writes(&self) -> bool {
1358 false
1359 }
1360
1361 fn pending_commit_pages(&self) -> Result<Vec<PageNumber>> {
1362 Ok(Vec::new())
1363 }
1364
1365 fn rollback<'a>(&'a mut self, _cx: &'a Cx) -> impl Future<Output = Result<()>> + 'a {
1366 async { Ok(()) }
1367 }
1368
1369 fn record_write_witness(&mut self, _cx: &Cx, _key: fsqlite_types::WitnessKey) {}
1370
1371 fn savepoint(&mut self, _cx: &Cx, name: &str) -> Result<()> {
1372 self.savepoint_names.push(name.to_owned());
1373 Ok(())
1374 }
1375
1376 fn release_savepoint(&mut self, _cx: &Cx, name: &str) -> Result<()> {
1377 if let Some(pos) = self.savepoint_names.iter().rposition(|n| n == name) {
1378 self.savepoint_names.truncate(pos);
1379 Ok(())
1380 } else {
1381 Err(fsqlite_error::FrankenError::internal(format!(
1382 "no savepoint named '{name}'"
1383 )))
1384 }
1385 }
1386
1387 fn rollback_to_savepoint(&mut self, _cx: &Cx, name: &str) -> Result<()> {
1388 if let Some(pos) = self.savepoint_names.iter().rposition(|n| n == name) {
1389 self.savepoint_names.truncate(pos + 1);
1390 Ok(())
1391 } else {
1392 Err(fsqlite_error::FrankenError::internal(format!(
1393 "no savepoint named '{name}'"
1394 )))
1395 }
1396 }
1397}
1398
1399#[derive(Debug, Default, Clone, Copy)]
1402pub struct MemoryMockMvccPager;
1403
1404impl sealed::Sealed for MemoryMockMvccPager {}
1405
1406impl MvccPager for MemoryMockMvccPager {
1407 type Txn = MemoryMockTransaction;
1408
1409 fn begin<'a>(
1410 &'a self,
1411 _cx: &'a Cx,
1412 _mode: TransactionMode,
1413 ) -> impl Future<Output = Result<Self::Txn>> + 'a {
1414 async {
1415 Ok(MemoryMockTransaction {
1416 committed: false,
1417 next_page: 2,
1418 pages: HashMap::new(),
1419 savepoints: Vec::new(),
1420 })
1421 }
1422 }
1423
1424 fn journal_mode(&self) -> JournalMode {
1425 JournalMode::Delete
1426 }
1427
1428 fn is_readonly(&self) -> bool {
1429 false
1430 }
1431
1432 fn set_journal_mode<'a>(
1433 &'a self,
1434 _cx: &'a Cx,
1435 mode: JournalMode,
1436 ) -> impl Future<Output = Result<JournalMode>> + 'a {
1437 async move { Ok(mode) }
1438 }
1439
1440 fn set_wal_backend(&self, _backend: Box<dyn WalBackend>) -> Result<()> {
1441 Ok(())
1442 }
1443}
1444
1445#[derive(Debug, Clone)]
1446struct MemoryMockSavepoint {
1447 name: String,
1448 next_page: u32,
1449 pages: HashMap<PageNumber, PageData>,
1450}
1451
1452#[derive(Debug, Clone)]
1455pub struct MemoryMockTransaction {
1456 committed: bool,
1457 next_page: u32,
1458 pages: HashMap<PageNumber, PageData>,
1459 savepoints: Vec<MemoryMockSavepoint>,
1460}
1461
1462impl sealed::Sealed for MemoryMockTransaction {}
1463
1464impl TransactionHandle for MemoryMockTransaction {
1465 fn get_page<'a>(
1466 &'a self,
1467 _cx: &'a Cx,
1468 page_no: PageNumber,
1469 ) -> impl Future<Output = Result<PageData>> + 'a {
1470 async move {
1471 Ok(self
1472 .pages
1473 .get(&page_no)
1474 .cloned()
1475 .unwrap_or_else(|| PageData::zeroed(fsqlite_types::PageSize::default())))
1476 }
1477 }
1478
1479 fn write_page<'a>(
1480 &'a mut self,
1481 _cx: &'a Cx,
1482 page_no: PageNumber,
1483 data: &'a [u8],
1484 ) -> impl Future<Output = Result<()>> + 'a {
1485 async move {
1486 self.committed = false;
1487 let page_size = fsqlite_types::PageSize::default().as_usize();
1488 let mut page = vec![0_u8; page_size];
1489 let copy_len = data.len().min(page_size);
1490 page[..copy_len].copy_from_slice(&data[..copy_len]);
1491 self.pages.insert(page_no, PageData::from_vec(page));
1492 Ok(())
1493 }
1494 }
1495
1496 fn write_page_data<'a>(
1497 &'a mut self,
1498 _cx: &'a Cx,
1499 page_no: PageNumber,
1500 data: PageData,
1501 ) -> impl Future<Output = Result<()>> + 'a {
1502 async move {
1503 self.committed = false;
1504 let page_size = fsqlite_types::PageSize::default().as_usize();
1505 let mut page = vec![0_u8; page_size];
1506 let copy_len = data.len().min(page_size);
1507 page[..copy_len].copy_from_slice(&data.as_bytes()[..copy_len]);
1508 self.pages.insert(page_no, PageData::from_vec(page));
1509 Ok(())
1510 }
1511 }
1512
1513 fn allocate_page<'a>(
1514 &'a mut self,
1515 _cx: &'a Cx,
1516 ) -> impl Future<Output = Result<PageNumber>> + 'a {
1517 async move {
1518 self.committed = false;
1519 let page = PageNumber::new(self.next_page)
1520 .expect("mock allocator must always produce non-zero page numbers");
1521 self.next_page += 1;
1522 self.pages
1523 .entry(page)
1524 .or_insert_with(|| PageData::zeroed(fsqlite_types::PageSize::default()));
1525 Ok(page)
1526 }
1527 }
1528
1529 fn free_page<'a>(
1530 &'a mut self,
1531 _cx: &'a Cx,
1532 page_no: PageNumber,
1533 ) -> impl Future<Output = Result<()>> + 'a {
1534 async move {
1535 self.committed = false;
1536 self.pages.remove(&page_no);
1537 Ok(())
1538 }
1539 }
1540
1541 fn commit<'a>(&'a mut self, _cx: &'a Cx) -> impl Future<Output = Result<()>> + 'a {
1542 async move {
1543 self.committed = true;
1544 Ok(())
1545 }
1546 }
1547
1548 fn pager_commit_state(&self) -> PagerCommitState {
1549 if self.committed {
1550 PagerCommitState::Committed
1551 } else {
1552 PagerCommitState::NotCommitted
1553 }
1554 }
1555
1556 fn is_writer(&self) -> bool {
1557 !self.pages.is_empty()
1558 }
1559
1560 fn has_pending_writes(&self) -> bool {
1561 !self.committed && !self.pages.is_empty()
1562 }
1563
1564 fn pending_commit_pages(&self) -> Result<Vec<PageNumber>> {
1565 let mut pages = self.pages.keys().copied().collect::<Vec<_>>();
1566 pages.sort_unstable();
1567 Ok(pages)
1568 }
1569
1570 fn rollback<'a>(&'a mut self, _cx: &'a Cx) -> impl Future<Output = Result<()>> + 'a {
1571 async move {
1572 self.committed = false;
1573 self.next_page = 2;
1574 self.pages.clear();
1575 self.savepoints.clear();
1576 Ok(())
1577 }
1578 }
1579
1580 fn record_write_witness(&mut self, _cx: &Cx, _key: fsqlite_types::WitnessKey) {}
1581
1582 fn savepoint(&mut self, _cx: &Cx, name: &str) -> Result<()> {
1583 self.savepoints.push(MemoryMockSavepoint {
1584 name: name.to_owned(),
1585 next_page: self.next_page,
1586 pages: self.pages.clone(),
1587 });
1588 Ok(())
1589 }
1590
1591 fn release_savepoint(&mut self, _cx: &Cx, name: &str) -> Result<()> {
1592 if let Some(pos) = self.savepoints.iter().rposition(|sp| sp.name == name) {
1593 self.savepoints.truncate(pos);
1594 Ok(())
1595 } else {
1596 Err(fsqlite_error::FrankenError::internal(format!(
1597 "no savepoint named '{name}'"
1598 )))
1599 }
1600 }
1601
1602 fn rollback_to_savepoint(&mut self, _cx: &Cx, name: &str) -> Result<()> {
1603 if let Some(pos) = self.savepoints.iter().rposition(|sp| sp.name == name) {
1604 let snapshot = self.savepoints[pos].clone();
1605 self.next_page = snapshot.next_page;
1606 self.pages = snapshot.pages;
1607 self.savepoints.truncate(pos + 1);
1608 Ok(())
1609 } else {
1610 Err(fsqlite_error::FrankenError::internal(format!(
1611 "no savepoint named '{name}'"
1612 )))
1613 }
1614 }
1615}
1616
1617#[cfg_attr(
1620 target_arch = "wasm32",
1621 expect(
1622 clippy::large_enum_variant,
1623 reason = "native transaction variants are absent on wasm, making the intentional inline memory transaction an apparent size outlier"
1624 )
1625)]
1626pub enum TransactionKind {
1627 Memory(SimpleTransaction<MemoryVfs>),
1629 #[cfg(all(feature = "native", target_os = "linux"))]
1631 IoUring(SimpleTransaction<IoUringVfs>),
1632 #[cfg(all(feature = "native", unix))]
1634 Unix(SimpleTransaction<UnixVfs>),
1635 #[cfg(all(feature = "native", target_os = "windows"))]
1637 Windows(SimpleTransaction<WindowsVfs>),
1638 Mock(MockTransaction),
1640 MemoryMock(MemoryMockTransaction),
1642 Drained,
1647}
1648
1649impl std::fmt::Debug for TransactionKind {
1650 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1651 match self {
1652 Self::Memory(_) => f.write_str("TransactionKind::Memory"),
1653 #[cfg(all(feature = "native", target_os = "linux"))]
1654 Self::IoUring(_) => f.write_str("TransactionKind::IoUring"),
1655 #[cfg(all(feature = "native", unix))]
1656 Self::Unix(_) => f.write_str("TransactionKind::Unix"),
1657 #[cfg(all(feature = "native", target_os = "windows"))]
1658 Self::Windows(_) => f.write_str("TransactionKind::Windows"),
1659 Self::Mock(_) => f.write_str("TransactionKind::Mock"),
1660 Self::MemoryMock(_) => f.write_str("TransactionKind::MemoryMock"),
1661 Self::Drained => f.write_str("TransactionKind::Drained"),
1662 }
1663 }
1664}
1665
1666impl TransactionKind {
1667 #[must_use]
1674 pub fn live_freelist_pages(&self) -> Vec<PageNumber> {
1675 match self {
1676 Self::Memory(txn) => txn.live_freelist_pages(),
1677 #[cfg(all(feature = "native", target_os = "linux"))]
1678 Self::IoUring(txn) => txn.live_freelist_pages(),
1679 #[cfg(all(feature = "native", unix))]
1680 Self::Unix(txn) => txn.live_freelist_pages(),
1681 #[cfg(all(feature = "native", target_os = "windows"))]
1682 Self::Windows(txn) => txn.live_freelist_pages(),
1683 Self::Mock(_) | Self::MemoryMock(_) | Self::Drained => Vec::new(),
1684 }
1685 }
1686
1687 #[must_use]
1693 pub fn live_db_size(&self) -> u32 {
1694 match self {
1695 Self::Memory(txn) => txn.live_db_size(),
1696 #[cfg(all(feature = "native", target_os = "linux"))]
1697 Self::IoUring(txn) => txn.live_db_size(),
1698 #[cfg(all(feature = "native", unix))]
1699 Self::Unix(txn) => txn.live_db_size(),
1700 #[cfg(all(feature = "native", target_os = "windows"))]
1701 Self::Windows(txn) => txn.live_db_size(),
1702 Self::Mock(_) | Self::MemoryMock(_) | Self::Drained => 0,
1703 }
1704 }
1705
1706 #[must_use]
1710 pub fn snapshot_db_size(&self) -> u32 {
1711 match self {
1712 Self::Memory(txn) => txn.snapshot_db_size(),
1713 #[cfg(all(feature = "native", target_os = "linux"))]
1714 Self::IoUring(txn) => txn.snapshot_db_size(),
1715 #[cfg(all(feature = "native", unix))]
1716 Self::Unix(txn) => txn.snapshot_db_size(),
1717 #[cfg(all(feature = "native", target_os = "windows"))]
1718 Self::Windows(txn) => txn.snapshot_db_size(),
1719 Self::Mock(_) | Self::MemoryMock(_) | Self::Drained => 0,
1720 }
1721 }
1722
1723 #[must_use]
1727 pub fn visible_db_size_bound(&self) -> u32 {
1728 match self {
1729 Self::Memory(txn) => txn.visible_db_size_bound(),
1730 #[cfg(all(feature = "native", target_os = "linux"))]
1731 Self::IoUring(txn) => txn.visible_db_size_bound(),
1732 #[cfg(all(feature = "native", unix))]
1733 Self::Unix(txn) => txn.visible_db_size_bound(),
1734 #[cfg(all(feature = "native", target_os = "windows"))]
1735 Self::Windows(txn) => txn.visible_db_size_bound(),
1736 Self::Mock(_) | Self::MemoryMock(_) | Self::Drained => 0,
1737 }
1738 }
1739}
1740
1741macro_rules! dispatch_transaction_kind {
1742 ($value:expr, $txn:ident => $body:expr) => {
1743 match $value {
1744 TransactionKind::Memory($txn) => $body,
1745 #[cfg(all(feature = "native", target_os = "linux"))]
1746 TransactionKind::IoUring($txn) => $body,
1747 #[cfg(all(feature = "native", unix))]
1748 TransactionKind::Unix($txn) => $body,
1749 #[cfg(all(feature = "native", target_os = "windows"))]
1750 TransactionKind::Windows($txn) => $body,
1751 TransactionKind::Mock($txn) => $body,
1752 TransactionKind::MemoryMock($txn) => $body,
1753 TransactionKind::Drained => {
1754 panic!("BUG: TransactionKind::Drained accessed while the transaction was extracted")
1755 }
1756 }
1757 };
1758}
1759
1760impl From<SimpleTransaction<MemoryVfs>> for TransactionKind {
1761 fn from(txn: SimpleTransaction<MemoryVfs>) -> Self {
1762 Self::Memory(txn)
1763 }
1764}
1765
1766#[cfg(all(feature = "native", target_os = "linux"))]
1767impl From<SimpleTransaction<IoUringVfs>> for TransactionKind {
1768 fn from(txn: SimpleTransaction<IoUringVfs>) -> Self {
1769 Self::IoUring(txn)
1770 }
1771}
1772
1773#[cfg(all(feature = "native", unix))]
1774impl From<SimpleTransaction<UnixVfs>> for TransactionKind {
1775 fn from(txn: SimpleTransaction<UnixVfs>) -> Self {
1776 Self::Unix(txn)
1777 }
1778}
1779
1780#[cfg(all(feature = "native", target_os = "windows"))]
1781impl From<SimpleTransaction<WindowsVfs>> for TransactionKind {
1782 fn from(txn: SimpleTransaction<WindowsVfs>) -> Self {
1783 Self::Windows(txn)
1784 }
1785}
1786
1787impl From<MockTransaction> for TransactionKind {
1788 fn from(txn: MockTransaction) -> Self {
1789 Self::Mock(txn)
1790 }
1791}
1792
1793impl From<MemoryMockTransaction> for TransactionKind {
1794 fn from(txn: MemoryMockTransaction) -> Self {
1795 Self::MemoryMock(txn)
1796 }
1797}
1798
1799impl sealed::Sealed for TransactionKind {}
1800
1801impl TransactionHandle for TransactionKind {
1802 fn get_page<'a>(
1810 &'a self,
1811 cx: &'a Cx,
1812 page_no: PageNumber,
1813 ) -> impl Future<Output = Result<PageData>> + 'a {
1814 async move { dispatch_transaction_kind!(self, txn => txn.get_page(cx, page_no).await) }
1815 }
1816
1817 fn prefetch_page_hint(&self, cx: &Cx, page_no: PageNumber) {
1818 dispatch_transaction_kind!(self, txn => txn.prefetch_page_hint(cx, page_no));
1819 }
1820
1821 fn write_page<'a>(
1822 &'a mut self,
1823 cx: &'a Cx,
1824 page_no: PageNumber,
1825 data: &'a [u8],
1826 ) -> impl Future<Output = Result<()>> + 'a {
1827 async move { dispatch_transaction_kind!(self, txn => txn.write_page(cx, page_no, data).await) }
1828 }
1829
1830 fn write_page_data<'a>(
1831 &'a mut self,
1832 cx: &'a Cx,
1833 page_no: PageNumber,
1834 data: PageData,
1835 ) -> impl Future<Output = Result<()>> + 'a {
1836 async move {
1837 dispatch_transaction_kind!(self, txn => txn.write_page_data(cx, page_no, data).await)
1838 }
1839 }
1840
1841 fn try_mutate_staged_page_data(
1842 &mut self,
1843 page_no: PageNumber,
1844 f: &mut dyn FnMut(&mut PageData),
1845 ) -> bool {
1846 dispatch_transaction_kind!(self, txn => txn.try_mutate_staged_page_data(page_no, f))
1847 }
1848
1849 fn allocate_page<'a>(
1850 &'a mut self,
1851 cx: &'a Cx,
1852 ) -> impl Future<Output = Result<PageNumber>> + 'a {
1853 async move { dispatch_transaction_kind!(self, txn => txn.allocate_page(cx).await) }
1854 }
1855
1856 fn free_page<'a>(
1857 &'a mut self,
1858 cx: &'a Cx,
1859 page_no: PageNumber,
1860 ) -> impl Future<Output = Result<()>> + 'a {
1861 async move { dispatch_transaction_kind!(self, txn => txn.free_page(cx, page_no).await) }
1862 }
1863
1864 fn commit<'a>(&'a mut self, cx: &'a Cx) -> impl Future<Output = Result<()>> + 'a {
1865 async move { dispatch_transaction_kind!(self, txn => txn.commit(cx).await) }
1866 }
1867
1868 fn pager_commit_state(&self) -> PagerCommitState {
1869 dispatch_transaction_kind!(self, txn => txn.pager_commit_state())
1870 }
1871
1872 fn commit_and_retain<'a>(&'a mut self, cx: &'a Cx) -> impl Future<Output = Result<bool>> + 'a {
1873 async move { dispatch_transaction_kind!(self, txn => txn.commit_and_retain(cx).await) }
1874 }
1875
1876 fn is_writer(&self) -> bool {
1877 dispatch_transaction_kind!(self, txn => txn.is_writer())
1878 }
1879
1880 fn has_pending_writes(&self) -> bool {
1881 dispatch_transaction_kind!(self, txn => txn.has_pending_writes())
1882 }
1883
1884 fn published_visible_commit_seq_hint(&self) -> Option<fsqlite_types::CommitSeq> {
1885 dispatch_transaction_kind!(self, txn => txn.published_visible_commit_seq_hint())
1886 }
1887
1888 fn pending_commit_pages(&self) -> Result<Vec<PageNumber>> {
1889 dispatch_transaction_kind!(self, txn => txn.pending_commit_pages())
1890 }
1891
1892 fn pending_conflict_pages(&self) -> Result<Vec<PageNumber>> {
1893 dispatch_transaction_kind!(self, txn => txn.pending_conflict_pages())
1894 }
1895
1896 fn pending_conflict_pages_conservative(&self) -> Vec<PageNumber> {
1897 dispatch_transaction_kind!(self, txn => txn.pending_conflict_pages_conservative())
1898 }
1899
1900 fn write_set_page_numbers(&self) -> Vec<PageNumber> {
1901 dispatch_transaction_kind!(self, txn => txn.write_set_page_numbers())
1902 }
1903
1904 fn page_one_in_pending_commit_surface(&self) -> Result<bool> {
1905 dispatch_transaction_kind!(self, txn => txn.page_one_in_pending_commit_surface())
1906 }
1907
1908 fn page_size(&self) -> PageSize {
1909 dispatch_transaction_kind!(self, txn => txn.page_size())
1910 }
1911
1912 fn allocate_page_requires_page_one_conflict_tracking(&self) -> Result<bool> {
1913 dispatch_transaction_kind!(self, txn => txn.allocate_page_requires_page_one_conflict_tracking())
1914 }
1915
1916 fn free_page_requires_page_one_conflict_tracking(&self, page_no: PageNumber) -> Result<bool> {
1917 dispatch_transaction_kind!(self, txn => txn.free_page_requires_page_one_conflict_tracking(page_no))
1918 }
1919
1920 fn write_page_requires_page_one_conflict_tracking(&self, page_no: PageNumber) -> Result<bool> {
1921 dispatch_transaction_kind!(self, txn => txn.write_page_requires_page_one_conflict_tracking(page_no))
1922 }
1923
1924 fn rollback<'a>(&'a mut self, cx: &'a Cx) -> impl Future<Output = Result<()>> + 'a {
1925 async move { dispatch_transaction_kind!(self, txn => txn.rollback(cx).await) }
1926 }
1927
1928 fn record_write_witness(&mut self, cx: &Cx, key: fsqlite_types::WitnessKey) {
1929 dispatch_transaction_kind!(self, txn => txn.record_write_witness(cx, key));
1930 }
1931
1932 fn savepoint(&mut self, cx: &Cx, name: &str) -> Result<()> {
1933 dispatch_transaction_kind!(self, txn => txn.savepoint(cx, name))
1934 }
1935
1936 fn release_savepoint(&mut self, cx: &Cx, name: &str) -> Result<()> {
1937 dispatch_transaction_kind!(self, txn => txn.release_savepoint(cx, name))
1938 }
1939
1940 fn rollback_to_savepoint(&mut self, cx: &Cx, name: &str) -> Result<()> {
1941 dispatch_transaction_kind!(self, txn => txn.rollback_to_savepoint(cx, name))
1942 }
1943}
1944
1945#[derive(Debug, Default, Clone, Copy)]
1947pub struct MockCheckpointPageWriter;
1948
1949impl sealed::Sealed for MockCheckpointPageWriter {}
1950
1951impl CheckpointPageWriter for MockCheckpointPageWriter {
1952 fn write_page<'a>(
1953 &'a mut self,
1954 _cx: &'a Cx,
1955 _page_no: PageNumber,
1956 _data: &'a [u8],
1957 ) -> WalFuture<'a, ()> {
1958 Box::pin(async { Ok(()) })
1959 }
1960
1961 fn truncate<'a>(&'a mut self, _cx: &'a Cx, _n_pages: u32) -> WalFuture<'a, ()> {
1962 Box::pin(async { Ok(()) })
1963 }
1964
1965 fn sync<'a>(&'a mut self, _cx: &'a Cx) -> WalFuture<'a, ()> {
1966 Box::pin(async { Ok(()) })
1967 }
1968}
1969
1970#[cfg(test)]
1975mod tests {
1976 use super::*;
1977 use fsqlite_vfs::VfsWriteCompletionState;
1978 use std::task::Poll;
1979
1980 const fn test_wal_generation_identity() -> WalGenerationIdentity {
1983 WalGenerationIdentity {
1984 checkpoint_seq: 0,
1985 salts: fsqlite_wal::checksum::WalSalts { salt1: 0, salt2: 0 },
1986 }
1987 }
1988
1989 struct PendingTrackedWalBackend;
1990
1991 impl WalBackend for PendingTrackedWalBackend {
1992 fn append_frame<'a>(
1993 &'a mut self,
1994 _cx: &'a Cx,
1995 _page_number: u32,
1996 _page_data: &'a [u8],
1997 _db_size_if_commit: u32,
1998 ) -> WalFuture<'a, ()> {
1999 Box::pin(std::future::pending())
2000 }
2001
2002 fn read_page<'a>(
2003 &'a mut self,
2004 _cx: &'a Cx,
2005 _page_number: u32,
2006 ) -> WalFuture<'a, Option<Vec<u8>>> {
2007 Box::pin(async { Ok(None) })
2008 }
2009
2010 fn sync(&mut self, _cx: &Cx) -> Result<()> {
2011 Ok(())
2012 }
2013
2014 fn frame_count(&self) -> usize {
2015 0
2016 }
2017
2018 fn checkpoint<'a>(
2019 &'a mut self,
2020 _cx: &'a Cx,
2021 mode: CheckpointMode,
2022 _writer: &'a mut dyn CheckpointPageWriter,
2023 _backfilled_frames: u32,
2024 _oldest_reader_frame: Option<u32>,
2025 ) -> WalFuture<'a, CheckpointResult> {
2026 Box::pin(async move {
2027 Ok(CheckpointResult {
2028 total_frames: 0,
2029 frames_backfilled: 0,
2030 completed: true,
2031 wal_was_reset: false,
2032 requested_mode: mode,
2033 effective_mode: mode,
2034 })
2035 })
2036 }
2037 }
2038
2039 #[test]
2040 fn tracked_default_marks_unpolled_drop_terminal_error() {
2041 let cx = Cx::new();
2042 let data = [0_u8; 16];
2043 let frames = [WalFrameRef {
2044 page_number: 1,
2045 page_data: &data,
2046 db_size_if_commit: 1,
2047 }];
2048 let completion = VfsWriteCompletion::new();
2049 let mut backend = PendingTrackedWalBackend;
2050
2051 let future = backend.append_frames_tracked(&cx, &frames, completion.clone());
2052 assert_eq!(completion.state(), VfsWriteCompletionState::Pending);
2053 drop(future);
2054 assert_eq!(completion.state(), VfsWriteCompletionState::Error);
2055 }
2056
2057 #[test]
2058 fn tracked_default_marks_polled_drop_terminal_error() {
2059 let cx = Cx::new();
2060 let data = [0_u8; 16];
2061 let frames = [WalFrameRef {
2062 page_number: 1,
2063 page_data: &data,
2064 db_size_if_commit: 1,
2065 }];
2066 let completion = VfsWriteCompletion::new();
2067 let mut backend = PendingTrackedWalBackend;
2068 let mut future = Box::pin(backend.append_frames_tracked(&cx, &frames, completion.clone()));
2069
2070 let polled = std::future::poll_fn(|poll_cx| {
2071 assert!(future.as_mut().poll(poll_cx).is_pending());
2072 Poll::Ready(())
2073 });
2074 let runtime = asupersync::runtime::RuntimeBuilder::current_thread()
2075 .blocking_threads(1, 1)
2076 .build()
2077 .expect("tracked-default test runtime should build");
2078 runtime.block_on(polled);
2079 assert_eq!(completion.state(), VfsWriteCompletionState::Pending);
2080 drop(future);
2081 assert_eq!(completion.state(), VfsWriteCompletionState::Error);
2082 }
2083
2084 #[test]
2085 fn test_pager_trait_is_sealed_mock_impl() {
2086 asupersync::test_utils::run_test(|| async {
2087 let pager = MockMvccPager;
2090 let cx = Cx::new();
2091 let _txn = pager.begin(&cx, TransactionMode::Deferred).await.unwrap();
2092 });
2093 }
2094
2095 #[test]
2096 fn test_mvccpager_begin_commit_rollback_signatures() {
2097 asupersync::test_utils::run_test(|| async {
2098 let pager = MockMvccPager;
2099 let cx = Cx::new();
2100
2101 let mut txn = pager.begin(&cx, TransactionMode::ReadOnly).await.unwrap();
2103
2104 let page_no = PageNumber::new(1).unwrap();
2106 let data = txn.get_page(&cx, page_no).await.unwrap();
2107 assert_eq!(
2108 u32::from_le_bytes(data.as_bytes()[..4].try_into().unwrap()),
2109 1
2110 );
2111
2112 txn.write_page(&cx, page_no, &[0u8; 4096]).await.unwrap();
2113 let new_page = txn.allocate_page(&cx).await.unwrap();
2114 assert_eq!(new_page.get(), 2);
2115 txn.free_page(&cx, new_page).await.unwrap();
2116
2117 txn.commit(&cx).await.unwrap();
2118 });
2119 }
2120
2121 #[test]
2122 fn test_transaction_rollback_is_infallible() {
2123 asupersync::test_utils::run_test(|| async {
2124 let pager = MockMvccPager;
2125 let cx = Cx::new();
2126 let mut txn = pager.begin(&cx, TransactionMode::Deferred).await.unwrap();
2127 txn.rollback(&cx).await.unwrap();
2129 });
2130 }
2131
2132 #[test]
2133 fn test_checkpoint_page_writer_signatures() {
2134 asupersync::test_utils::run_test(|| async {
2135 let mut writer = MockCheckpointPageWriter;
2136 let cx = Cx::new();
2137 let page1 = PageNumber::new(1).unwrap();
2138
2139 writer.write_page(&cx, page1, &[0u8; 4096]).await.unwrap();
2140 writer.truncate(&cx, 10).await.unwrap();
2141 writer.sync(&cx).await.unwrap();
2142 });
2143 }
2144
2145 #[test]
2146 fn test_transaction_mode_default_is_deferred() {
2147 assert_eq!(TransactionMode::default(), TransactionMode::Deferred);
2148 }
2149
2150 #[test]
2151 fn test_open_traits_are_extensible() {
2152 fn assert_is_mvcc_pager<P: MvccPager<Txn = MockTransaction>>(_pager: &P) {}
2163 let pager = MockMvccPager;
2164 assert_is_mvcc_pager(&pager);
2165 }
2166
2167 #[test]
2168 fn test_memory_mock_transaction_persists_writes() {
2169 asupersync::test_utils::run_test(|| async {
2170 let pager = MemoryMockMvccPager;
2171 let cx = Cx::new();
2172 let mut txn = pager.begin(&cx, TransactionMode::Immediate).await.unwrap();
2173 let page_no = PageNumber::new(256).unwrap();
2174
2175 let mut bytes = vec![0_u8; fsqlite_types::PageSize::default().as_usize()];
2176 bytes[0] = 0x0A;
2177 txn.write_page(&cx, page_no, &bytes).await.unwrap();
2178
2179 let page = txn.get_page(&cx, page_no).await.unwrap();
2180 assert_eq!(page.as_bytes()[0], 0x0A);
2181 assert!(txn.has_pending_writes());
2182 assert!(txn.is_writer());
2183 });
2184 }
2185
2186 #[test]
2187 fn test_memory_mock_transaction_commit_clears_pending_writes() {
2188 asupersync::test_utils::run_test(|| async {
2189 let pager = MemoryMockMvccPager;
2190 let cx = Cx::new();
2191 let mut txn = pager.begin(&cx, TransactionMode::Immediate).await.unwrap();
2192 let page_no = PageNumber::new(2).unwrap();
2193
2194 txn.write_page(&cx, page_no, &[1_u8; 4096]).await.unwrap();
2195 assert!(txn.has_pending_writes());
2196
2197 txn.commit(&cx).await.unwrap();
2198 assert!(
2199 !txn.has_pending_writes(),
2200 "committed mock transactions must not report pending writes"
2201 );
2202 });
2203 }
2204
2205 #[test]
2206 fn test_memory_mock_transaction_rollback_resets_allocator() {
2207 asupersync::test_utils::run_test(|| async {
2208 let pager = MemoryMockMvccPager;
2209 let cx = Cx::new();
2210 let mut txn = pager.begin(&cx, TransactionMode::Immediate).await.unwrap();
2211
2212 assert_eq!(txn.allocate_page(&cx).await.unwrap().get(), 2);
2213 assert_eq!(txn.allocate_page(&cx).await.unwrap().get(), 3);
2214
2215 txn.rollback(&cx).await.unwrap();
2216
2217 assert_eq!(
2218 txn.allocate_page(&cx).await.unwrap().get(),
2219 2,
2220 "rollback should restore the mock allocator to its initial state"
2221 );
2222 });
2223 }
2224
2225 #[test]
2226 fn test_checkpoint_mode_default_is_passive() {
2227 assert_eq!(CheckpointMode::default(), CheckpointMode::Passive);
2228 }
2229
2230 #[test]
2231 fn test_journal_mode_default_is_delete() {
2232 assert_eq!(JournalMode::default(), JournalMode::Delete);
2233 }
2234
2235 #[test]
2236 fn test_wal_publication_snapshot_authoritative_when_index_full() {
2237 let snap = WalPublicationSnapshot {
2238 publication_seq: 1,
2239 generation: test_wal_generation_identity(),
2240 last_commit_frame: Some(10),
2241 commit_count: 5,
2242 latest_frame_entries: 10,
2243 index_is_partial: false,
2244 };
2245 assert!(
2246 snap.lookup_contract_is_authoritative(),
2247 "full index must be authoritative"
2248 );
2249 }
2250
2251 #[test]
2252 fn test_wal_publication_snapshot_not_authoritative_when_partial() {
2253 let snap = WalPublicationSnapshot {
2254 publication_seq: 1,
2255 generation: test_wal_generation_identity(),
2256 last_commit_frame: None,
2257 commit_count: 0,
2258 latest_frame_entries: 0,
2259 index_is_partial: true,
2260 };
2261 assert!(
2262 !snap.lookup_contract_is_authoritative(),
2263 "partial index must not be authoritative"
2264 );
2265 }
2266
2267 #[test]
2268 fn test_prepared_wal_frame_batch_frame_count_and_page_size() {
2269 let batch = PreparedWalFrameBatch {
2270 frame_size: 4120,
2271 page_data_offset: 24,
2272 big_endian_checksum: false,
2273 frame_metas: vec![
2274 PreparedWalFrameMeta {
2275 page_number: 1,
2276 db_size_if_commit: 0,
2277 },
2278 PreparedWalFrameMeta {
2279 page_number: 2,
2280 db_size_if_commit: 10,
2281 },
2282 ],
2283 checksum_transforms: Vec::new(),
2284 frame_bytes: vec![0u8; 4120 * 2],
2285 last_commit_frame_offset: Some(4120),
2286 finalized_for: None,
2287 finalized_running_checksum: None,
2288 };
2289 assert_eq!(batch.frame_count(), 2);
2290 assert_eq!(batch.page_size(), 4096);
2291 }
2292
2293 #[test]
2294 fn test_prepared_wal_frame_batch_set_db_size_clears_finalized() {
2295 let mut batch = PreparedWalFrameBatch {
2296 frame_size: 32,
2297 page_data_offset: 8,
2298 big_endian_checksum: false,
2299 frame_metas: vec![PreparedWalFrameMeta {
2300 page_number: 1,
2301 db_size_if_commit: 0,
2302 }],
2303 checksum_transforms: Vec::new(),
2304 frame_bytes: vec![0u8; 32],
2305 last_commit_frame_offset: None,
2306 finalized_for: Some(PreparedWalFinalizationState {
2307 checkpoint_seq: 1,
2308 salt1: 0xAA,
2309 salt2: 0xBB,
2310 start_frame_index: 0,
2311 seed: PreparedWalChecksumSeed::default(),
2312 }),
2313 finalized_running_checksum: Some(PreparedWalChecksumSeed { s1: 1, s2: 2 }),
2314 };
2315
2316 batch.set_db_size_if_commit(0, 42);
2317
2318 assert_eq!(batch.frame_metas[0].db_size_if_commit, 42);
2319 assert!(
2320 batch.finalized_for.is_none(),
2321 "set_db_size_if_commit must invalidate finalized_for"
2322 );
2323 assert!(
2324 batch.finalized_running_checksum.is_none(),
2325 "set_db_size_if_commit must invalidate finalized_running_checksum"
2326 );
2327 let db_bytes = &batch.frame_bytes[4..8];
2328 assert_eq!(u32::from_be_bytes(db_bytes.try_into().unwrap()), 42);
2329 }
2330
2331 #[test]
2332 fn test_mock_release_savepoint_unknown_name_returns_error() {
2333 asupersync::test_utils::run_test(|| async {
2334 let pager = MockMvccPager;
2335 let cx = Cx::new();
2336 let mut txn = pager.begin(&cx, TransactionMode::Deferred).await.unwrap();
2337
2338 let result = txn.release_savepoint(&cx, "nonexistent");
2339 assert!(result.is_err(), "releasing unknown savepoint must fail");
2340 });
2341 }
2342
2343 #[test]
2344 fn test_memory_mock_savepoint_rollback_restores_pages() {
2345 asupersync::test_utils::run_test(|| async {
2346 let pager = MemoryMockMvccPager;
2347 let cx = Cx::new();
2348 let mut txn = pager.begin(&cx, TransactionMode::Immediate).await.unwrap();
2349
2350 let p1 = PageNumber::new(1).unwrap();
2351 let page_size = fsqlite_types::PageSize::default().as_usize();
2352 let mut data_a = vec![0u8; page_size];
2353 data_a[0] = 0xAA;
2354 txn.write_page(&cx, p1, &data_a).await.unwrap();
2355
2356 txn.savepoint(&cx, "sp1").unwrap();
2357
2358 let mut data_b = vec![0u8; page_size];
2359 data_b[0] = 0xBB;
2360 txn.write_page(&cx, p1, &data_b).await.unwrap();
2361 assert_eq!(txn.get_page(&cx, p1).await.unwrap().as_bytes()[0], 0xBB);
2362
2363 txn.rollback_to_savepoint(&cx, "sp1").unwrap();
2364 assert_eq!(
2365 txn.get_page(&cx, p1).await.unwrap().as_bytes()[0],
2366 0xAA,
2367 "rollback_to_savepoint must restore page state"
2368 );
2369 });
2370 }
2371
2372 #[test]
2373 fn test_transaction_mode_default_trait_contract_is_deferred() {
2374 assert_eq!(TransactionMode::default(), TransactionMode::Deferred);
2375 }
2376
2377 #[test]
2378 fn test_checkpoint_result_fields() {
2379 let result = CheckpointResult {
2380 total_frames: 100,
2381 frames_backfilled: 80,
2382 completed: false,
2383 wal_was_reset: false,
2384 requested_mode: CheckpointMode::Full,
2385 effective_mode: CheckpointMode::Passive,
2386 };
2387 assert_eq!(result.total_frames, 100);
2388 assert_eq!(result.frames_backfilled, 80);
2389 assert!(!result.completed);
2390 assert_ne!(result.requested_mode, result.effective_mode);
2391 }
2392
2393 #[test]
2394 fn test_journal_mode_debug_clone_copy_eq() {
2395 let a = JournalMode::Wal;
2396 let b = a;
2397 assert_eq!(a, b);
2398 assert_ne!(JournalMode::Delete, JournalMode::Wal);
2399 let dbg = format!("{a:?}");
2400 assert!(dbg.contains("Wal"));
2401 }
2402
2403 #[test]
2404 fn test_checkpoint_result_clone_debug() {
2405 let result = CheckpointResult {
2406 total_frames: 50,
2407 frames_backfilled: 50,
2408 completed: true,
2409 wal_was_reset: true,
2410 requested_mode: CheckpointMode::Truncate,
2411 effective_mode: CheckpointMode::Truncate,
2412 };
2413 let cloned = result.clone();
2414 assert_eq!(result, cloned);
2415 let dbg = format!("{result:?}");
2416 assert!(dbg.contains("CheckpointResult"));
2417 assert!(dbg.contains("Truncate"));
2418 assert!(dbg.contains("wal_was_reset"));
2419 }
2420
2421 #[test]
2422 fn test_wal_publication_snapshot_clone_copy_debug() {
2423 let snap = WalPublicationSnapshot {
2424 publication_seq: 42,
2425 generation: test_wal_generation_identity(),
2426 last_commit_frame: Some(100),
2427 commit_count: 7,
2428 latest_frame_entries: 50,
2429 index_is_partial: false,
2430 };
2431 let copied = snap;
2432 assert_eq!(copied, snap);
2433 let dbg = format!("{snap:?}");
2434 assert!(dbg.contains("WalPublicationSnapshot"));
2435 assert!(dbg.contains("publication_seq"));
2436 assert!(dbg.contains("42"));
2437 }
2438
2439 #[test]
2440 fn test_checkpoint_mode_all_variants_debug() {
2441 for (mode, expected) in [
2442 (CheckpointMode::Passive, "Passive"),
2443 (CheckpointMode::Full, "Full"),
2444 (CheckpointMode::Restart, "Restart"),
2445 (CheckpointMode::Truncate, "Truncate"),
2446 ] {
2447 let dbg = format!("{mode:?}");
2448 assert!(dbg.contains(expected), "expected {expected} in {dbg}");
2449 let copy = mode;
2450 assert_eq!(mode, copy);
2451 }
2452 }
2453
2454 #[test]
2455 fn test_prepared_wal_frame_batch_page_data_and_frame_slice() {
2456 let frame_size = 32;
2457 let page_data_offset = 8;
2458 let mut frame_bytes = vec![0u8; frame_size * 2];
2459 frame_bytes[8] = 0xAA;
2460 frame_bytes[frame_size + 8] = 0xBB;
2461
2462 let batch = PreparedWalFrameBatch {
2463 frame_size,
2464 page_data_offset,
2465 big_endian_checksum: false,
2466 frame_metas: vec![
2467 PreparedWalFrameMeta {
2468 page_number: 1,
2469 db_size_if_commit: 0,
2470 },
2471 PreparedWalFrameMeta {
2472 page_number: 2,
2473 db_size_if_commit: 5,
2474 },
2475 ],
2476 checksum_transforms: Vec::new(),
2477 frame_bytes,
2478 last_commit_frame_offset: None,
2479 finalized_for: None,
2480 finalized_running_checksum: None,
2481 };
2482
2483 assert_eq!(batch.page_data(0)[0], 0xAA);
2484 assert_eq!(batch.page_data(1)[0], 0xBB);
2485 assert_eq!(batch.frame_slice(0).len(), frame_size);
2486 assert_eq!(batch.frame_slice(1).len(), frame_size);
2487
2488 let refs = batch.frame_refs();
2489 assert_eq!(refs.len(), 2);
2490 assert_eq!(refs[0].page_number, 1);
2491 assert_eq!(refs[1].db_size_if_commit, 5);
2492 assert_eq!(refs[0].page_data[0], 0xAA);
2493 assert_eq!(refs[1].page_data[0], 0xBB);
2494 }
2495
2496 #[test]
2497 fn prepared_wal_frame_meta_debug_clone_copy_eq() {
2498 let a = PreparedWalFrameMeta {
2499 page_number: 5,
2500 db_size_if_commit: 0,
2501 };
2502 let b = PreparedWalFrameMeta {
2503 page_number: 5,
2504 db_size_if_commit: 10,
2505 };
2506 let copied = a;
2507 assert_eq!(copied, a);
2508 assert_ne!(a, b);
2509 let dbg = format!("{a:?}");
2510 assert!(dbg.contains("PreparedWalFrameMeta"));
2511 assert!(dbg.contains("5"));
2512 }
2513
2514 #[test]
2515 fn prepared_wal_checksum_seed_default_and_eq() {
2516 let def = PreparedWalChecksumSeed::default();
2517 assert_eq!(def.s1, 0);
2518 assert_eq!(def.s2, 0);
2519 let other = PreparedWalChecksumSeed { s1: 1, s2: 2 };
2520 assert_ne!(def, other);
2521 let copied = other;
2522 assert_eq!(copied, other);
2523 let dbg = format!("{def:?}");
2524 assert!(dbg.contains("PreparedWalChecksumSeed"));
2525 }
2526
2527 #[test]
2528 fn prepared_wal_finalization_state_default_and_eq() {
2529 let def = PreparedWalFinalizationState::default();
2530 assert_eq!(def.checkpoint_seq, 0);
2531 assert_eq!(def.salt1, 0);
2532 assert_eq!(def.salt2, 0);
2533 assert_eq!(def.start_frame_index, 0);
2534 assert_eq!(def.seed, PreparedWalChecksumSeed::default());
2535 let other = PreparedWalFinalizationState {
2536 checkpoint_seq: 1,
2537 salt1: 0xAA,
2538 salt2: 0xBB,
2539 start_frame_index: 42,
2540 seed: PreparedWalChecksumSeed { s1: 10, s2: 20 },
2541 };
2542 assert_ne!(def, other);
2543 let copied = other;
2544 assert_eq!(copied, other);
2545 let dbg = format!("{other:?}");
2546 assert!(dbg.contains("PreparedWalFinalizationState"));
2547 }
2548
2549 #[test]
2550 fn transaction_mode_all_variants_debug_copy_eq() {
2551 let variants = [
2552 (TransactionMode::Deferred, "Deferred"),
2553 (TransactionMode::Immediate, "Immediate"),
2554 (TransactionMode::Exclusive, "Exclusive"),
2555 (TransactionMode::Concurrent, "Concurrent"),
2556 (TransactionMode::ReadOnly, "ReadOnly"),
2557 ];
2558 for (mode, expected) in &variants {
2559 let dbg = format!("{mode:?}");
2560 assert!(dbg.contains(expected), "expected {expected} in {dbg}");
2561 let copied = *mode;
2562 assert_eq!(copied, *mode);
2563 }
2564 assert_ne!(TransactionMode::Deferred, TransactionMode::Concurrent);
2565 }
2566
2567 #[test]
2568 fn wal_frame_ref_debug_clone_copy() {
2569 let data = [0xABu8; 16];
2570 let frame = WalFrameRef {
2571 page_number: 3,
2572 page_data: &data,
2573 db_size_if_commit: 0,
2574 };
2575 let copied = frame;
2576 assert_eq!(copied.page_number, 3);
2577 assert_eq!(copied.page_data.len(), 16);
2578 assert_eq!(copied.db_size_if_commit, 0);
2579 let dbg = format!("{frame:?}");
2580 assert!(dbg.contains("WalFrameRef"));
2581 }
2582
2583 #[test]
2584 fn mock_checkpoint_page_writer_default_and_trait_methods() {
2585 asupersync::test_utils::run_test(|| async {
2586 let mut writer = MockCheckpointPageWriter;
2587 let cx = Cx::new();
2588 let page = PageNumber::new(1).unwrap();
2589 writer.write_page(&cx, page, &[0u8; 4096]).await.unwrap();
2590 writer.truncate(&cx, 10).await.unwrap();
2591 writer.sync(&cx).await.unwrap();
2592 let dbg = format!("{writer:?}");
2593 assert!(dbg.contains("MockCheckpointPageWriter"));
2594 });
2595 }
2596
2597 #[test]
2598 fn transaction_kind_drained_debug() {
2599 let kind = TransactionKind::Drained;
2600 let dbg = format!("{kind:?}");
2601 assert!(dbg.contains("Drained"));
2602 }
2603
2604 #[test]
2605 fn wal_publication_snapshot_authoritative_boundary() {
2606 let base = WalPublicationSnapshot {
2607 publication_seq: 1,
2608 generation: test_wal_generation_identity(),
2609 last_commit_frame: Some(10),
2610 commit_count: 5,
2611 latest_frame_entries: 10,
2612 index_is_partial: false,
2613 };
2614 assert!(base.lookup_contract_is_authoritative());
2615 let partial = WalPublicationSnapshot {
2616 index_is_partial: true,
2617 ..base
2618 };
2619 assert!(!partial.lookup_contract_is_authoritative());
2620 }
2621}