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 forget_cached_page(&self, _page_no: PageNumber) {}
954
955 fn write_page<'a>(
960 &'a mut self,
961 cx: &'a Cx,
962 page_no: PageNumber,
963 data: &'a [u8],
964 ) -> impl Future<Output = Result<()>> + 'a;
965
966 fn write_page_data<'a>(
971 &'a mut self,
972 cx: &'a Cx,
973 page_no: PageNumber,
974 data: PageData,
975 ) -> impl Future<Output = Result<()>> + 'a {
976 async move { self.write_page(cx, page_no, data.as_bytes()).await }
977 }
978
979 fn try_take_staged_page_data(&mut self, _page_no: PageNumber) -> Option<PageData> {
986 None
987 }
988
989 fn try_mutate_staged_page_data(
995 &mut self,
996 _page_no: PageNumber,
997 _f: &mut dyn FnMut(&mut PageData),
998 ) -> bool {
999 false
1000 }
1001
1002 fn restore_staged_page_data<'a>(
1008 &'a mut self,
1009 cx: &'a Cx,
1010 page_no: PageNumber,
1011 data: PageData,
1012 ) -> impl Future<Output = Result<()>> + 'a {
1013 async move { self.write_page_data(cx, page_no, data).await }
1014 }
1015
1016 fn allocate_page<'a>(&'a mut self, cx: &'a Cx)
1020 -> impl Future<Output = Result<PageNumber>> + 'a;
1021
1022 fn free_page<'a>(
1024 &'a mut self,
1025 cx: &'a Cx,
1026 page_no: PageNumber,
1027 ) -> impl Future<Output = Result<()>> + 'a;
1028
1029 fn commit<'a>(&'a mut self, cx: &'a Cx) -> impl Future<Output = Result<()>> + 'a;
1035
1036 fn pager_commit_state(&self) -> PagerCommitState {
1042 PagerCommitState::NotCommitted
1043 }
1044
1045 fn commit_and_retain<'a>(&'a mut self, cx: &'a Cx) -> impl Future<Output = Result<bool>> + 'a {
1061 async move {
1062 self.commit(cx).await?;
1063 Ok(false)
1064 }
1065 }
1066
1067 fn is_writer(&self) -> bool;
1073
1074 fn has_pending_writes(&self) -> bool;
1079
1080 fn published_visible_commit_seq_hint(&self) -> Option<fsqlite_types::CommitSeq> {
1086 None
1087 }
1088
1089 fn pending_commit_pages(&self) -> Result<Vec<PageNumber>> {
1093 Ok(Vec::new())
1094 }
1095
1096 fn pending_conflict_pages(&self) -> Result<Vec<PageNumber>> {
1104 self.pending_commit_pages()
1105 }
1106
1107 fn pending_conflict_pages_conservative(&self) -> Vec<PageNumber> {
1122 self.write_set_page_numbers()
1123 }
1124
1125 fn write_set_page_numbers(&self) -> Vec<PageNumber> {
1128 Vec::new()
1129 }
1130
1131 fn page_one_in_pending_commit_surface(&self) -> Result<bool> {
1134 Ok(self.pending_commit_pages()?.contains(&PageNumber::ONE))
1135 }
1136
1137 fn page_size(&self) -> PageSize {
1142 PageSize::default()
1143 }
1144
1145 fn allocate_page_requires_page_one_conflict_tracking(&self) -> Result<bool> {
1154 Ok(true)
1155 }
1156
1157 fn free_page_requires_page_one_conflict_tracking(&self, _page_no: PageNumber) -> Result<bool> {
1166 Ok(true)
1167 }
1168
1169 fn write_page_requires_page_one_conflict_tracking(&self, _page_no: PageNumber) -> Result<bool> {
1179 Ok(true)
1180 }
1181
1182 fn rollback<'a>(&'a mut self, cx: &'a Cx) -> impl Future<Output = Result<()>> + 'a;
1188
1189 fn record_write_witness(&mut self, _cx: &Cx, _key: fsqlite_types::WitnessKey) {}
1194
1195 fn savepoint(&mut self, cx: &Cx, name: &str) -> Result<()>;
1201
1202 fn release_savepoint(&mut self, cx: &Cx, name: &str) -> Result<()>;
1208
1209 fn rollback_to_savepoint(&mut self, cx: &Cx, name: &str) -> Result<()>;
1216}
1217
1218pub trait CheckpointPageWriter: sealed::Sealed + Send {
1232 fn write_page<'a>(
1234 &'a mut self,
1235 cx: &'a Cx,
1236 page_no: PageNumber,
1237 data: &'a [u8],
1238 ) -> WalFuture<'a, ()>;
1239
1240 fn truncate<'a>(&'a mut self, cx: &'a Cx, n_pages: u32) -> WalFuture<'a, ()>;
1242
1243 fn sync<'a>(&'a mut self, cx: &'a Cx) -> WalFuture<'a, ()>;
1245}
1246
1247#[derive(Debug, Default, Clone, Copy)]
1253pub struct MockMvccPager;
1254
1255impl sealed::Sealed for MockMvccPager {}
1256
1257impl MvccPager for MockMvccPager {
1258 type Txn = MockTransaction;
1259
1260 fn begin<'a>(
1261 &'a self,
1262 _cx: &'a Cx,
1263 _mode: TransactionMode,
1264 ) -> impl Future<Output = Result<Self::Txn>> + 'a {
1265 async {
1266 Ok(MockTransaction {
1267 committed: false,
1268 next_page: 2,
1269 savepoint_names: Vec::new(),
1270 })
1271 }
1272 }
1273
1274 fn journal_mode(&self) -> JournalMode {
1275 JournalMode::Delete
1276 }
1277
1278 fn is_readonly(&self) -> bool {
1279 false
1280 }
1281
1282 fn set_journal_mode<'a>(
1283 &'a self,
1284 _cx: &'a Cx,
1285 mode: JournalMode,
1286 ) -> impl Future<Output = Result<JournalMode>> + 'a {
1287 async move { Ok(mode) }
1288 }
1289
1290 fn set_wal_backend(&self, _backend: Box<dyn WalBackend>) -> Result<()> {
1291 Ok(())
1292 }
1293}
1294
1295#[derive(Debug, Clone)]
1297pub struct MockTransaction {
1298 committed: bool,
1299 next_page: u32,
1300 savepoint_names: Vec<String>,
1301}
1302
1303impl sealed::Sealed for MockTransaction {}
1304
1305impl TransactionHandle for MockTransaction {
1306 fn get_page<'a>(
1307 &'a self,
1308 _cx: &'a Cx,
1309 page_no: PageNumber,
1310 ) -> impl Future<Output = Result<PageData>> + 'a {
1311 async move {
1312 let size = fsqlite_types::PageSize::default();
1313 let mut data = PageData::zeroed(size);
1314 data.as_bytes_mut()[..4].copy_from_slice(&page_no.get().to_le_bytes());
1315 Ok(data)
1316 }
1317 }
1318
1319 fn write_page<'a>(
1320 &'a mut self,
1321 _cx: &'a Cx,
1322 _page_no: PageNumber,
1323 _data: &'a [u8],
1324 ) -> impl Future<Output = Result<()>> + 'a {
1325 async { Ok(()) }
1326 }
1327
1328 fn allocate_page<'a>(
1329 &'a mut self,
1330 _cx: &'a Cx,
1331 ) -> impl Future<Output = Result<PageNumber>> + 'a {
1332 async move {
1333 let page = PageNumber::new(self.next_page)
1334 .expect("mock allocator must always produce non-zero page numbers");
1335 self.next_page += 1;
1336 Ok(page)
1337 }
1338 }
1339
1340 fn free_page<'a>(
1341 &'a mut self,
1342 _cx: &'a Cx,
1343 _page_no: PageNumber,
1344 ) -> impl Future<Output = Result<()>> + 'a {
1345 async { Ok(()) }
1346 }
1347
1348 fn commit<'a>(&'a mut self, _cx: &'a Cx) -> impl Future<Output = Result<()>> + 'a {
1349 async move {
1350 self.committed = true;
1351 Ok(())
1352 }
1353 }
1354
1355 fn pager_commit_state(&self) -> PagerCommitState {
1356 if self.committed {
1357 PagerCommitState::Committed
1358 } else {
1359 PagerCommitState::NotCommitted
1360 }
1361 }
1362
1363 fn is_writer(&self) -> bool {
1364 false
1365 }
1366
1367 fn has_pending_writes(&self) -> bool {
1368 false
1369 }
1370
1371 fn pending_commit_pages(&self) -> Result<Vec<PageNumber>> {
1372 Ok(Vec::new())
1373 }
1374
1375 fn rollback<'a>(&'a mut self, _cx: &'a Cx) -> impl Future<Output = Result<()>> + 'a {
1376 async { Ok(()) }
1377 }
1378
1379 fn record_write_witness(&mut self, _cx: &Cx, _key: fsqlite_types::WitnessKey) {}
1380
1381 fn savepoint(&mut self, _cx: &Cx, name: &str) -> Result<()> {
1382 self.savepoint_names.push(name.to_owned());
1383 Ok(())
1384 }
1385
1386 fn release_savepoint(&mut self, _cx: &Cx, name: &str) -> Result<()> {
1387 if let Some(pos) = self.savepoint_names.iter().rposition(|n| n == name) {
1388 self.savepoint_names.truncate(pos);
1389 Ok(())
1390 } else {
1391 Err(fsqlite_error::FrankenError::internal(format!(
1392 "no savepoint named '{name}'"
1393 )))
1394 }
1395 }
1396
1397 fn rollback_to_savepoint(&mut self, _cx: &Cx, name: &str) -> Result<()> {
1398 if let Some(pos) = self.savepoint_names.iter().rposition(|n| n == name) {
1399 self.savepoint_names.truncate(pos + 1);
1400 Ok(())
1401 } else {
1402 Err(fsqlite_error::FrankenError::internal(format!(
1403 "no savepoint named '{name}'"
1404 )))
1405 }
1406 }
1407}
1408
1409#[derive(Debug, Default, Clone, Copy)]
1412pub struct MemoryMockMvccPager;
1413
1414impl sealed::Sealed for MemoryMockMvccPager {}
1415
1416impl MvccPager for MemoryMockMvccPager {
1417 type Txn = MemoryMockTransaction;
1418
1419 fn begin<'a>(
1420 &'a self,
1421 _cx: &'a Cx,
1422 _mode: TransactionMode,
1423 ) -> impl Future<Output = Result<Self::Txn>> + 'a {
1424 async {
1425 Ok(MemoryMockTransaction {
1426 committed: false,
1427 next_page: 2,
1428 pages: HashMap::new(),
1429 savepoints: Vec::new(),
1430 })
1431 }
1432 }
1433
1434 fn journal_mode(&self) -> JournalMode {
1435 JournalMode::Delete
1436 }
1437
1438 fn is_readonly(&self) -> bool {
1439 false
1440 }
1441
1442 fn set_journal_mode<'a>(
1443 &'a self,
1444 _cx: &'a Cx,
1445 mode: JournalMode,
1446 ) -> impl Future<Output = Result<JournalMode>> + 'a {
1447 async move { Ok(mode) }
1448 }
1449
1450 fn set_wal_backend(&self, _backend: Box<dyn WalBackend>) -> Result<()> {
1451 Ok(())
1452 }
1453}
1454
1455#[derive(Debug, Clone)]
1456struct MemoryMockSavepoint {
1457 name: String,
1458 next_page: u32,
1459 pages: HashMap<PageNumber, PageData>,
1460}
1461
1462#[derive(Debug, Clone)]
1465pub struct MemoryMockTransaction {
1466 committed: bool,
1467 next_page: u32,
1468 pages: HashMap<PageNumber, PageData>,
1469 savepoints: Vec<MemoryMockSavepoint>,
1470}
1471
1472impl sealed::Sealed for MemoryMockTransaction {}
1473
1474impl TransactionHandle for MemoryMockTransaction {
1475 fn get_page<'a>(
1476 &'a self,
1477 _cx: &'a Cx,
1478 page_no: PageNumber,
1479 ) -> impl Future<Output = Result<PageData>> + 'a {
1480 async move {
1481 Ok(self
1482 .pages
1483 .get(&page_no)
1484 .cloned()
1485 .unwrap_or_else(|| PageData::zeroed(fsqlite_types::PageSize::default())))
1486 }
1487 }
1488
1489 fn write_page<'a>(
1490 &'a mut self,
1491 _cx: &'a Cx,
1492 page_no: PageNumber,
1493 data: &'a [u8],
1494 ) -> impl Future<Output = Result<()>> + 'a {
1495 async move {
1496 self.committed = false;
1497 let page_size = fsqlite_types::PageSize::default().as_usize();
1498 let mut page = vec![0_u8; page_size];
1499 let copy_len = data.len().min(page_size);
1500 page[..copy_len].copy_from_slice(&data[..copy_len]);
1501 self.pages.insert(page_no, PageData::from_vec(page));
1502 Ok(())
1503 }
1504 }
1505
1506 fn write_page_data<'a>(
1507 &'a mut self,
1508 _cx: &'a Cx,
1509 page_no: PageNumber,
1510 data: PageData,
1511 ) -> impl Future<Output = Result<()>> + 'a {
1512 async move {
1513 self.committed = false;
1514 let page_size = fsqlite_types::PageSize::default().as_usize();
1515 let mut page = vec![0_u8; page_size];
1516 let copy_len = data.len().min(page_size);
1517 page[..copy_len].copy_from_slice(&data.as_bytes()[..copy_len]);
1518 self.pages.insert(page_no, PageData::from_vec(page));
1519 Ok(())
1520 }
1521 }
1522
1523 fn allocate_page<'a>(
1524 &'a mut self,
1525 _cx: &'a Cx,
1526 ) -> impl Future<Output = Result<PageNumber>> + 'a {
1527 async move {
1528 self.committed = false;
1529 let page = PageNumber::new(self.next_page)
1530 .expect("mock allocator must always produce non-zero page numbers");
1531 self.next_page += 1;
1532 self.pages
1533 .entry(page)
1534 .or_insert_with(|| PageData::zeroed(fsqlite_types::PageSize::default()));
1535 Ok(page)
1536 }
1537 }
1538
1539 fn free_page<'a>(
1540 &'a mut self,
1541 _cx: &'a Cx,
1542 page_no: PageNumber,
1543 ) -> impl Future<Output = Result<()>> + 'a {
1544 async move {
1545 self.committed = false;
1546 self.pages.remove(&page_no);
1547 Ok(())
1548 }
1549 }
1550
1551 fn commit<'a>(&'a mut self, _cx: &'a Cx) -> impl Future<Output = Result<()>> + 'a {
1552 async move {
1553 self.committed = true;
1554 Ok(())
1555 }
1556 }
1557
1558 fn pager_commit_state(&self) -> PagerCommitState {
1559 if self.committed {
1560 PagerCommitState::Committed
1561 } else {
1562 PagerCommitState::NotCommitted
1563 }
1564 }
1565
1566 fn is_writer(&self) -> bool {
1567 !self.pages.is_empty()
1568 }
1569
1570 fn has_pending_writes(&self) -> bool {
1571 !self.committed && !self.pages.is_empty()
1572 }
1573
1574 fn pending_commit_pages(&self) -> Result<Vec<PageNumber>> {
1575 let mut pages = self.pages.keys().copied().collect::<Vec<_>>();
1576 pages.sort_unstable();
1577 Ok(pages)
1578 }
1579
1580 fn rollback<'a>(&'a mut self, _cx: &'a Cx) -> impl Future<Output = Result<()>> + 'a {
1581 async move {
1582 self.committed = false;
1583 self.next_page = 2;
1584 self.pages.clear();
1585 self.savepoints.clear();
1586 Ok(())
1587 }
1588 }
1589
1590 fn record_write_witness(&mut self, _cx: &Cx, _key: fsqlite_types::WitnessKey) {}
1591
1592 fn savepoint(&mut self, _cx: &Cx, name: &str) -> Result<()> {
1593 self.savepoints.push(MemoryMockSavepoint {
1594 name: name.to_owned(),
1595 next_page: self.next_page,
1596 pages: self.pages.clone(),
1597 });
1598 Ok(())
1599 }
1600
1601 fn release_savepoint(&mut self, _cx: &Cx, name: &str) -> Result<()> {
1602 if let Some(pos) = self.savepoints.iter().rposition(|sp| sp.name == name) {
1603 self.savepoints.truncate(pos);
1604 Ok(())
1605 } else {
1606 Err(fsqlite_error::FrankenError::internal(format!(
1607 "no savepoint named '{name}'"
1608 )))
1609 }
1610 }
1611
1612 fn rollback_to_savepoint(&mut self, _cx: &Cx, name: &str) -> Result<()> {
1613 if let Some(pos) = self.savepoints.iter().rposition(|sp| sp.name == name) {
1614 let snapshot = self.savepoints[pos].clone();
1615 self.next_page = snapshot.next_page;
1616 self.pages = snapshot.pages;
1617 self.savepoints.truncate(pos + 1);
1618 Ok(())
1619 } else {
1620 Err(fsqlite_error::FrankenError::internal(format!(
1621 "no savepoint named '{name}'"
1622 )))
1623 }
1624 }
1625}
1626
1627#[cfg_attr(
1630 target_arch = "wasm32",
1631 expect(
1632 clippy::large_enum_variant,
1633 reason = "native transaction variants are absent on wasm, making the intentional inline memory transaction an apparent size outlier"
1634 )
1635)]
1636pub enum TransactionKind {
1637 Memory(SimpleTransaction<MemoryVfs>),
1639 #[cfg(all(feature = "native", target_os = "linux"))]
1641 IoUring(SimpleTransaction<IoUringVfs>),
1642 #[cfg(all(feature = "native", unix))]
1644 Unix(SimpleTransaction<UnixVfs>),
1645 #[cfg(all(feature = "native", target_os = "windows"))]
1647 Windows(SimpleTransaction<WindowsVfs>),
1648 Mock(MockTransaction),
1650 MemoryMock(MemoryMockTransaction),
1652 Drained,
1657}
1658
1659impl std::fmt::Debug for TransactionKind {
1660 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1661 match self {
1662 Self::Memory(_) => f.write_str("TransactionKind::Memory"),
1663 #[cfg(all(feature = "native", target_os = "linux"))]
1664 Self::IoUring(_) => f.write_str("TransactionKind::IoUring"),
1665 #[cfg(all(feature = "native", unix))]
1666 Self::Unix(_) => f.write_str("TransactionKind::Unix"),
1667 #[cfg(all(feature = "native", target_os = "windows"))]
1668 Self::Windows(_) => f.write_str("TransactionKind::Windows"),
1669 Self::Mock(_) => f.write_str("TransactionKind::Mock"),
1670 Self::MemoryMock(_) => f.write_str("TransactionKind::MemoryMock"),
1671 Self::Drained => f.write_str("TransactionKind::Drained"),
1672 }
1673 }
1674}
1675
1676impl TransactionKind {
1677 #[must_use]
1684 pub fn live_freelist_pages(&self) -> Vec<PageNumber> {
1685 match self {
1686 Self::Memory(txn) => txn.live_freelist_pages(),
1687 #[cfg(all(feature = "native", target_os = "linux"))]
1688 Self::IoUring(txn) => txn.live_freelist_pages(),
1689 #[cfg(all(feature = "native", unix))]
1690 Self::Unix(txn) => txn.live_freelist_pages(),
1691 #[cfg(all(feature = "native", target_os = "windows"))]
1692 Self::Windows(txn) => txn.live_freelist_pages(),
1693 Self::Mock(_) | Self::MemoryMock(_) | Self::Drained => Vec::new(),
1694 }
1695 }
1696
1697 #[must_use]
1703 pub fn live_db_size(&self) -> u32 {
1704 match self {
1705 Self::Memory(txn) => txn.live_db_size(),
1706 #[cfg(all(feature = "native", target_os = "linux"))]
1707 Self::IoUring(txn) => txn.live_db_size(),
1708 #[cfg(all(feature = "native", unix))]
1709 Self::Unix(txn) => txn.live_db_size(),
1710 #[cfg(all(feature = "native", target_os = "windows"))]
1711 Self::Windows(txn) => txn.live_db_size(),
1712 Self::Mock(_) | Self::MemoryMock(_) | Self::Drained => 0,
1713 }
1714 }
1715
1716 #[must_use]
1720 pub fn snapshot_db_size(&self) -> u32 {
1721 match self {
1722 Self::Memory(txn) => txn.snapshot_db_size(),
1723 #[cfg(all(feature = "native", target_os = "linux"))]
1724 Self::IoUring(txn) => txn.snapshot_db_size(),
1725 #[cfg(all(feature = "native", unix))]
1726 Self::Unix(txn) => txn.snapshot_db_size(),
1727 #[cfg(all(feature = "native", target_os = "windows"))]
1728 Self::Windows(txn) => txn.snapshot_db_size(),
1729 Self::Mock(_) | Self::MemoryMock(_) | Self::Drained => 0,
1730 }
1731 }
1732
1733 #[must_use]
1737 pub fn visible_db_size_bound(&self) -> u32 {
1738 match self {
1739 Self::Memory(txn) => txn.visible_db_size_bound(),
1740 #[cfg(all(feature = "native", target_os = "linux"))]
1741 Self::IoUring(txn) => txn.visible_db_size_bound(),
1742 #[cfg(all(feature = "native", unix))]
1743 Self::Unix(txn) => txn.visible_db_size_bound(),
1744 #[cfg(all(feature = "native", target_os = "windows"))]
1745 Self::Windows(txn) => txn.visible_db_size_bound(),
1746 Self::Mock(_) | Self::MemoryMock(_) | Self::Drained => 0,
1747 }
1748 }
1749}
1750
1751macro_rules! dispatch_transaction_kind {
1752 ($value:expr, $txn:ident => $body:expr) => {
1753 match $value {
1754 TransactionKind::Memory($txn) => $body,
1755 #[cfg(all(feature = "native", target_os = "linux"))]
1756 TransactionKind::IoUring($txn) => $body,
1757 #[cfg(all(feature = "native", unix))]
1758 TransactionKind::Unix($txn) => $body,
1759 #[cfg(all(feature = "native", target_os = "windows"))]
1760 TransactionKind::Windows($txn) => $body,
1761 TransactionKind::Mock($txn) => $body,
1762 TransactionKind::MemoryMock($txn) => $body,
1763 TransactionKind::Drained => {
1764 panic!("BUG: TransactionKind::Drained accessed while the transaction was extracted")
1765 }
1766 }
1767 };
1768}
1769
1770impl From<SimpleTransaction<MemoryVfs>> for TransactionKind {
1771 fn from(txn: SimpleTransaction<MemoryVfs>) -> Self {
1772 Self::Memory(txn)
1773 }
1774}
1775
1776#[cfg(all(feature = "native", target_os = "linux"))]
1777impl From<SimpleTransaction<IoUringVfs>> for TransactionKind {
1778 fn from(txn: SimpleTransaction<IoUringVfs>) -> Self {
1779 Self::IoUring(txn)
1780 }
1781}
1782
1783#[cfg(all(feature = "native", unix))]
1784impl From<SimpleTransaction<UnixVfs>> for TransactionKind {
1785 fn from(txn: SimpleTransaction<UnixVfs>) -> Self {
1786 Self::Unix(txn)
1787 }
1788}
1789
1790#[cfg(all(feature = "native", target_os = "windows"))]
1791impl From<SimpleTransaction<WindowsVfs>> for TransactionKind {
1792 fn from(txn: SimpleTransaction<WindowsVfs>) -> Self {
1793 Self::Windows(txn)
1794 }
1795}
1796
1797impl From<MockTransaction> for TransactionKind {
1798 fn from(txn: MockTransaction) -> Self {
1799 Self::Mock(txn)
1800 }
1801}
1802
1803impl From<MemoryMockTransaction> for TransactionKind {
1804 fn from(txn: MemoryMockTransaction) -> Self {
1805 Self::MemoryMock(txn)
1806 }
1807}
1808
1809impl sealed::Sealed for TransactionKind {}
1810
1811impl TransactionHandle for TransactionKind {
1812 fn get_page<'a>(
1820 &'a self,
1821 cx: &'a Cx,
1822 page_no: PageNumber,
1823 ) -> impl Future<Output = Result<PageData>> + 'a {
1824 async move { dispatch_transaction_kind!(self, txn => txn.get_page(cx, page_no).await) }
1825 }
1826
1827 fn prefetch_page_hint(&self, cx: &Cx, page_no: PageNumber) {
1828 dispatch_transaction_kind!(self, txn => txn.prefetch_page_hint(cx, page_no));
1829 }
1830
1831 fn forget_cached_page(&self, page_no: PageNumber) {
1832 dispatch_transaction_kind!(self, txn => txn.forget_cached_page(page_no));
1833 }
1834
1835 fn write_page<'a>(
1836 &'a mut self,
1837 cx: &'a Cx,
1838 page_no: PageNumber,
1839 data: &'a [u8],
1840 ) -> impl Future<Output = Result<()>> + 'a {
1841 async move { dispatch_transaction_kind!(self, txn => txn.write_page(cx, page_no, data).await) }
1842 }
1843
1844 fn write_page_data<'a>(
1845 &'a mut self,
1846 cx: &'a Cx,
1847 page_no: PageNumber,
1848 data: PageData,
1849 ) -> impl Future<Output = Result<()>> + 'a {
1850 async move {
1851 dispatch_transaction_kind!(self, txn => txn.write_page_data(cx, page_no, data).await)
1852 }
1853 }
1854
1855 fn try_mutate_staged_page_data(
1856 &mut self,
1857 page_no: PageNumber,
1858 f: &mut dyn FnMut(&mut PageData),
1859 ) -> bool {
1860 dispatch_transaction_kind!(self, txn => txn.try_mutate_staged_page_data(page_no, f))
1861 }
1862
1863 fn allocate_page<'a>(
1864 &'a mut self,
1865 cx: &'a Cx,
1866 ) -> impl Future<Output = Result<PageNumber>> + 'a {
1867 async move { dispatch_transaction_kind!(self, txn => txn.allocate_page(cx).await) }
1868 }
1869
1870 fn free_page<'a>(
1871 &'a mut self,
1872 cx: &'a Cx,
1873 page_no: PageNumber,
1874 ) -> impl Future<Output = Result<()>> + 'a {
1875 async move { dispatch_transaction_kind!(self, txn => txn.free_page(cx, page_no).await) }
1876 }
1877
1878 fn commit<'a>(&'a mut self, cx: &'a Cx) -> impl Future<Output = Result<()>> + 'a {
1879 async move { dispatch_transaction_kind!(self, txn => txn.commit(cx).await) }
1880 }
1881
1882 fn pager_commit_state(&self) -> PagerCommitState {
1883 dispatch_transaction_kind!(self, txn => txn.pager_commit_state())
1884 }
1885
1886 fn commit_and_retain<'a>(&'a mut self, cx: &'a Cx) -> impl Future<Output = Result<bool>> + 'a {
1887 async move { dispatch_transaction_kind!(self, txn => txn.commit_and_retain(cx).await) }
1888 }
1889
1890 fn is_writer(&self) -> bool {
1891 dispatch_transaction_kind!(self, txn => txn.is_writer())
1892 }
1893
1894 fn has_pending_writes(&self) -> bool {
1895 dispatch_transaction_kind!(self, txn => txn.has_pending_writes())
1896 }
1897
1898 fn published_visible_commit_seq_hint(&self) -> Option<fsqlite_types::CommitSeq> {
1899 dispatch_transaction_kind!(self, txn => txn.published_visible_commit_seq_hint())
1900 }
1901
1902 fn pending_commit_pages(&self) -> Result<Vec<PageNumber>> {
1903 dispatch_transaction_kind!(self, txn => txn.pending_commit_pages())
1904 }
1905
1906 fn pending_conflict_pages(&self) -> Result<Vec<PageNumber>> {
1907 dispatch_transaction_kind!(self, txn => txn.pending_conflict_pages())
1908 }
1909
1910 fn pending_conflict_pages_conservative(&self) -> Vec<PageNumber> {
1911 dispatch_transaction_kind!(self, txn => txn.pending_conflict_pages_conservative())
1912 }
1913
1914 fn write_set_page_numbers(&self) -> Vec<PageNumber> {
1915 dispatch_transaction_kind!(self, txn => txn.write_set_page_numbers())
1916 }
1917
1918 fn page_one_in_pending_commit_surface(&self) -> Result<bool> {
1919 dispatch_transaction_kind!(self, txn => txn.page_one_in_pending_commit_surface())
1920 }
1921
1922 fn page_size(&self) -> PageSize {
1923 dispatch_transaction_kind!(self, txn => txn.page_size())
1924 }
1925
1926 fn allocate_page_requires_page_one_conflict_tracking(&self) -> Result<bool> {
1927 dispatch_transaction_kind!(self, txn => txn.allocate_page_requires_page_one_conflict_tracking())
1928 }
1929
1930 fn free_page_requires_page_one_conflict_tracking(&self, page_no: PageNumber) -> Result<bool> {
1931 dispatch_transaction_kind!(self, txn => txn.free_page_requires_page_one_conflict_tracking(page_no))
1932 }
1933
1934 fn write_page_requires_page_one_conflict_tracking(&self, page_no: PageNumber) -> Result<bool> {
1935 dispatch_transaction_kind!(self, txn => txn.write_page_requires_page_one_conflict_tracking(page_no))
1936 }
1937
1938 fn rollback<'a>(&'a mut self, cx: &'a Cx) -> impl Future<Output = Result<()>> + 'a {
1939 async move { dispatch_transaction_kind!(self, txn => txn.rollback(cx).await) }
1940 }
1941
1942 fn record_write_witness(&mut self, cx: &Cx, key: fsqlite_types::WitnessKey) {
1943 dispatch_transaction_kind!(self, txn => txn.record_write_witness(cx, key));
1944 }
1945
1946 fn savepoint(&mut self, cx: &Cx, name: &str) -> Result<()> {
1947 dispatch_transaction_kind!(self, txn => txn.savepoint(cx, name))
1948 }
1949
1950 fn release_savepoint(&mut self, cx: &Cx, name: &str) -> Result<()> {
1951 dispatch_transaction_kind!(self, txn => txn.release_savepoint(cx, name))
1952 }
1953
1954 fn rollback_to_savepoint(&mut self, cx: &Cx, name: &str) -> Result<()> {
1955 dispatch_transaction_kind!(self, txn => txn.rollback_to_savepoint(cx, name))
1956 }
1957}
1958
1959#[derive(Debug, Default, Clone, Copy)]
1961pub struct MockCheckpointPageWriter;
1962
1963impl sealed::Sealed for MockCheckpointPageWriter {}
1964
1965impl CheckpointPageWriter for MockCheckpointPageWriter {
1966 fn write_page<'a>(
1967 &'a mut self,
1968 _cx: &'a Cx,
1969 _page_no: PageNumber,
1970 _data: &'a [u8],
1971 ) -> WalFuture<'a, ()> {
1972 Box::pin(async { Ok(()) })
1973 }
1974
1975 fn truncate<'a>(&'a mut self, _cx: &'a Cx, _n_pages: u32) -> WalFuture<'a, ()> {
1976 Box::pin(async { Ok(()) })
1977 }
1978
1979 fn sync<'a>(&'a mut self, _cx: &'a Cx) -> WalFuture<'a, ()> {
1980 Box::pin(async { Ok(()) })
1981 }
1982}
1983
1984#[cfg(test)]
1989mod tests {
1990 use super::*;
1991 use fsqlite_vfs::VfsWriteCompletionState;
1992 use std::task::Poll;
1993
1994 const fn test_wal_generation_identity() -> WalGenerationIdentity {
1997 WalGenerationIdentity {
1998 checkpoint_seq: 0,
1999 salts: fsqlite_wal::checksum::WalSalts { salt1: 0, salt2: 0 },
2000 }
2001 }
2002
2003 struct PendingTrackedWalBackend;
2004
2005 impl WalBackend for PendingTrackedWalBackend {
2006 fn append_frame<'a>(
2007 &'a mut self,
2008 _cx: &'a Cx,
2009 _page_number: u32,
2010 _page_data: &'a [u8],
2011 _db_size_if_commit: u32,
2012 ) -> WalFuture<'a, ()> {
2013 Box::pin(std::future::pending())
2014 }
2015
2016 fn read_page<'a>(
2017 &'a mut self,
2018 _cx: &'a Cx,
2019 _page_number: u32,
2020 ) -> WalFuture<'a, Option<Vec<u8>>> {
2021 Box::pin(async { Ok(None) })
2022 }
2023
2024 fn sync(&mut self, _cx: &Cx) -> Result<()> {
2025 Ok(())
2026 }
2027
2028 fn frame_count(&self) -> usize {
2029 0
2030 }
2031
2032 fn checkpoint<'a>(
2033 &'a mut self,
2034 _cx: &'a Cx,
2035 mode: CheckpointMode,
2036 _writer: &'a mut dyn CheckpointPageWriter,
2037 _backfilled_frames: u32,
2038 _oldest_reader_frame: Option<u32>,
2039 ) -> WalFuture<'a, CheckpointResult> {
2040 Box::pin(async move {
2041 Ok(CheckpointResult {
2042 total_frames: 0,
2043 frames_backfilled: 0,
2044 completed: true,
2045 wal_was_reset: false,
2046 requested_mode: mode,
2047 effective_mode: mode,
2048 })
2049 })
2050 }
2051 }
2052
2053 #[test]
2054 fn tracked_default_marks_unpolled_drop_terminal_error() {
2055 let cx = Cx::new();
2056 let data = [0_u8; 16];
2057 let frames = [WalFrameRef {
2058 page_number: 1,
2059 page_data: &data,
2060 db_size_if_commit: 1,
2061 }];
2062 let completion = VfsWriteCompletion::new();
2063 let mut backend = PendingTrackedWalBackend;
2064
2065 let future = backend.append_frames_tracked(&cx, &frames, completion.clone());
2066 assert_eq!(completion.state(), VfsWriteCompletionState::Pending);
2067 drop(future);
2068 assert_eq!(completion.state(), VfsWriteCompletionState::Error);
2069 }
2070
2071 #[test]
2072 fn tracked_default_marks_polled_drop_terminal_error() {
2073 let cx = Cx::new();
2074 let data = [0_u8; 16];
2075 let frames = [WalFrameRef {
2076 page_number: 1,
2077 page_data: &data,
2078 db_size_if_commit: 1,
2079 }];
2080 let completion = VfsWriteCompletion::new();
2081 let mut backend = PendingTrackedWalBackend;
2082 let mut future = Box::pin(backend.append_frames_tracked(&cx, &frames, completion.clone()));
2083
2084 let polled = std::future::poll_fn(|poll_cx| {
2085 assert!(future.as_mut().poll(poll_cx).is_pending());
2086 Poll::Ready(())
2087 });
2088 let runtime = asupersync::runtime::RuntimeBuilder::current_thread()
2089 .blocking_threads(1, 1)
2090 .build()
2091 .expect("tracked-default test runtime should build");
2092 runtime.block_on(polled);
2093 assert_eq!(completion.state(), VfsWriteCompletionState::Pending);
2094 drop(future);
2095 assert_eq!(completion.state(), VfsWriteCompletionState::Error);
2096 }
2097
2098 #[test]
2099 fn test_pager_trait_is_sealed_mock_impl() {
2100 asupersync::test_utils::run_test(|| async {
2101 let pager = MockMvccPager;
2104 let cx = Cx::new();
2105 let _txn = pager.begin(&cx, TransactionMode::Deferred).await.unwrap();
2106 });
2107 }
2108
2109 #[test]
2110 fn test_mvccpager_begin_commit_rollback_signatures() {
2111 asupersync::test_utils::run_test(|| async {
2112 let pager = MockMvccPager;
2113 let cx = Cx::new();
2114
2115 let mut txn = pager.begin(&cx, TransactionMode::ReadOnly).await.unwrap();
2117
2118 let page_no = PageNumber::new(1).unwrap();
2120 let data = txn.get_page(&cx, page_no).await.unwrap();
2121 assert_eq!(
2122 u32::from_le_bytes(data.as_bytes()[..4].try_into().unwrap()),
2123 1
2124 );
2125
2126 txn.write_page(&cx, page_no, &[0u8; 4096]).await.unwrap();
2127 let new_page = txn.allocate_page(&cx).await.unwrap();
2128 assert_eq!(new_page.get(), 2);
2129 txn.free_page(&cx, new_page).await.unwrap();
2130
2131 txn.commit(&cx).await.unwrap();
2132 });
2133 }
2134
2135 #[test]
2136 fn test_transaction_rollback_is_infallible() {
2137 asupersync::test_utils::run_test(|| async {
2138 let pager = MockMvccPager;
2139 let cx = Cx::new();
2140 let mut txn = pager.begin(&cx, TransactionMode::Deferred).await.unwrap();
2141 txn.rollback(&cx).await.unwrap();
2143 });
2144 }
2145
2146 #[test]
2147 fn test_checkpoint_page_writer_signatures() {
2148 asupersync::test_utils::run_test(|| async {
2149 let mut writer = MockCheckpointPageWriter;
2150 let cx = Cx::new();
2151 let page1 = PageNumber::new(1).unwrap();
2152
2153 writer.write_page(&cx, page1, &[0u8; 4096]).await.unwrap();
2154 writer.truncate(&cx, 10).await.unwrap();
2155 writer.sync(&cx).await.unwrap();
2156 });
2157 }
2158
2159 #[test]
2160 fn test_transaction_mode_default_is_deferred() {
2161 assert_eq!(TransactionMode::default(), TransactionMode::Deferred);
2162 }
2163
2164 #[test]
2165 fn test_open_traits_are_extensible() {
2166 fn assert_is_mvcc_pager<P: MvccPager<Txn = MockTransaction>>(_pager: &P) {}
2177 let pager = MockMvccPager;
2178 assert_is_mvcc_pager(&pager);
2179 }
2180
2181 #[test]
2182 fn test_memory_mock_transaction_persists_writes() {
2183 asupersync::test_utils::run_test(|| async {
2184 let pager = MemoryMockMvccPager;
2185 let cx = Cx::new();
2186 let mut txn = pager.begin(&cx, TransactionMode::Immediate).await.unwrap();
2187 let page_no = PageNumber::new(256).unwrap();
2188
2189 let mut bytes = vec![0_u8; fsqlite_types::PageSize::default().as_usize()];
2190 bytes[0] = 0x0A;
2191 txn.write_page(&cx, page_no, &bytes).await.unwrap();
2192
2193 let page = txn.get_page(&cx, page_no).await.unwrap();
2194 assert_eq!(page.as_bytes()[0], 0x0A);
2195 assert!(txn.has_pending_writes());
2196 assert!(txn.is_writer());
2197 });
2198 }
2199
2200 #[test]
2201 fn test_memory_mock_transaction_commit_clears_pending_writes() {
2202 asupersync::test_utils::run_test(|| async {
2203 let pager = MemoryMockMvccPager;
2204 let cx = Cx::new();
2205 let mut txn = pager.begin(&cx, TransactionMode::Immediate).await.unwrap();
2206 let page_no = PageNumber::new(2).unwrap();
2207
2208 txn.write_page(&cx, page_no, &[1_u8; 4096]).await.unwrap();
2209 assert!(txn.has_pending_writes());
2210
2211 txn.commit(&cx).await.unwrap();
2212 assert!(
2213 !txn.has_pending_writes(),
2214 "committed mock transactions must not report pending writes"
2215 );
2216 });
2217 }
2218
2219 #[test]
2220 fn test_memory_mock_transaction_rollback_resets_allocator() {
2221 asupersync::test_utils::run_test(|| async {
2222 let pager = MemoryMockMvccPager;
2223 let cx = Cx::new();
2224 let mut txn = pager.begin(&cx, TransactionMode::Immediate).await.unwrap();
2225
2226 assert_eq!(txn.allocate_page(&cx).await.unwrap().get(), 2);
2227 assert_eq!(txn.allocate_page(&cx).await.unwrap().get(), 3);
2228
2229 txn.rollback(&cx).await.unwrap();
2230
2231 assert_eq!(
2232 txn.allocate_page(&cx).await.unwrap().get(),
2233 2,
2234 "rollback should restore the mock allocator to its initial state"
2235 );
2236 });
2237 }
2238
2239 #[test]
2240 fn test_checkpoint_mode_default_is_passive() {
2241 assert_eq!(CheckpointMode::default(), CheckpointMode::Passive);
2242 }
2243
2244 #[test]
2245 fn test_journal_mode_default_is_delete() {
2246 assert_eq!(JournalMode::default(), JournalMode::Delete);
2247 }
2248
2249 #[test]
2250 fn test_wal_publication_snapshot_authoritative_when_index_full() {
2251 let snap = WalPublicationSnapshot {
2252 publication_seq: 1,
2253 generation: test_wal_generation_identity(),
2254 last_commit_frame: Some(10),
2255 commit_count: 5,
2256 latest_frame_entries: 10,
2257 index_is_partial: false,
2258 };
2259 assert!(
2260 snap.lookup_contract_is_authoritative(),
2261 "full index must be authoritative"
2262 );
2263 }
2264
2265 #[test]
2266 fn test_wal_publication_snapshot_not_authoritative_when_partial() {
2267 let snap = WalPublicationSnapshot {
2268 publication_seq: 1,
2269 generation: test_wal_generation_identity(),
2270 last_commit_frame: None,
2271 commit_count: 0,
2272 latest_frame_entries: 0,
2273 index_is_partial: true,
2274 };
2275 assert!(
2276 !snap.lookup_contract_is_authoritative(),
2277 "partial index must not be authoritative"
2278 );
2279 }
2280
2281 #[test]
2282 fn test_prepared_wal_frame_batch_frame_count_and_page_size() {
2283 let batch = PreparedWalFrameBatch {
2284 frame_size: 4120,
2285 page_data_offset: 24,
2286 big_endian_checksum: false,
2287 frame_metas: vec![
2288 PreparedWalFrameMeta {
2289 page_number: 1,
2290 db_size_if_commit: 0,
2291 },
2292 PreparedWalFrameMeta {
2293 page_number: 2,
2294 db_size_if_commit: 10,
2295 },
2296 ],
2297 checksum_transforms: Vec::new(),
2298 frame_bytes: vec![0u8; 4120 * 2],
2299 last_commit_frame_offset: Some(4120),
2300 finalized_for: None,
2301 finalized_running_checksum: None,
2302 };
2303 assert_eq!(batch.frame_count(), 2);
2304 assert_eq!(batch.page_size(), 4096);
2305 }
2306
2307 #[test]
2308 fn test_prepared_wal_frame_batch_set_db_size_clears_finalized() {
2309 let mut batch = PreparedWalFrameBatch {
2310 frame_size: 32,
2311 page_data_offset: 8,
2312 big_endian_checksum: false,
2313 frame_metas: vec![PreparedWalFrameMeta {
2314 page_number: 1,
2315 db_size_if_commit: 0,
2316 }],
2317 checksum_transforms: Vec::new(),
2318 frame_bytes: vec![0u8; 32],
2319 last_commit_frame_offset: None,
2320 finalized_for: Some(PreparedWalFinalizationState {
2321 checkpoint_seq: 1,
2322 salt1: 0xAA,
2323 salt2: 0xBB,
2324 start_frame_index: 0,
2325 seed: PreparedWalChecksumSeed::default(),
2326 }),
2327 finalized_running_checksum: Some(PreparedWalChecksumSeed { s1: 1, s2: 2 }),
2328 };
2329
2330 batch.set_db_size_if_commit(0, 42);
2331
2332 assert_eq!(batch.frame_metas[0].db_size_if_commit, 42);
2333 assert!(
2334 batch.finalized_for.is_none(),
2335 "set_db_size_if_commit must invalidate finalized_for"
2336 );
2337 assert!(
2338 batch.finalized_running_checksum.is_none(),
2339 "set_db_size_if_commit must invalidate finalized_running_checksum"
2340 );
2341 let db_bytes = &batch.frame_bytes[4..8];
2342 assert_eq!(u32::from_be_bytes(db_bytes.try_into().unwrap()), 42);
2343 }
2344
2345 #[test]
2346 fn test_mock_release_savepoint_unknown_name_returns_error() {
2347 asupersync::test_utils::run_test(|| async {
2348 let pager = MockMvccPager;
2349 let cx = Cx::new();
2350 let mut txn = pager.begin(&cx, TransactionMode::Deferred).await.unwrap();
2351
2352 let result = txn.release_savepoint(&cx, "nonexistent");
2353 assert!(result.is_err(), "releasing unknown savepoint must fail");
2354 });
2355 }
2356
2357 #[test]
2358 fn test_memory_mock_savepoint_rollback_restores_pages() {
2359 asupersync::test_utils::run_test(|| async {
2360 let pager = MemoryMockMvccPager;
2361 let cx = Cx::new();
2362 let mut txn = pager.begin(&cx, TransactionMode::Immediate).await.unwrap();
2363
2364 let p1 = PageNumber::new(1).unwrap();
2365 let page_size = fsqlite_types::PageSize::default().as_usize();
2366 let mut data_a = vec![0u8; page_size];
2367 data_a[0] = 0xAA;
2368 txn.write_page(&cx, p1, &data_a).await.unwrap();
2369
2370 txn.savepoint(&cx, "sp1").unwrap();
2371
2372 let mut data_b = vec![0u8; page_size];
2373 data_b[0] = 0xBB;
2374 txn.write_page(&cx, p1, &data_b).await.unwrap();
2375 assert_eq!(txn.get_page(&cx, p1).await.unwrap().as_bytes()[0], 0xBB);
2376
2377 txn.rollback_to_savepoint(&cx, "sp1").unwrap();
2378 assert_eq!(
2379 txn.get_page(&cx, p1).await.unwrap().as_bytes()[0],
2380 0xAA,
2381 "rollback_to_savepoint must restore page state"
2382 );
2383 });
2384 }
2385
2386 #[test]
2387 fn test_transaction_mode_default_trait_contract_is_deferred() {
2388 assert_eq!(TransactionMode::default(), TransactionMode::Deferred);
2389 }
2390
2391 #[test]
2392 fn test_checkpoint_result_fields() {
2393 let result = CheckpointResult {
2394 total_frames: 100,
2395 frames_backfilled: 80,
2396 completed: false,
2397 wal_was_reset: false,
2398 requested_mode: CheckpointMode::Full,
2399 effective_mode: CheckpointMode::Passive,
2400 };
2401 assert_eq!(result.total_frames, 100);
2402 assert_eq!(result.frames_backfilled, 80);
2403 assert!(!result.completed);
2404 assert_ne!(result.requested_mode, result.effective_mode);
2405 }
2406
2407 #[test]
2408 fn test_journal_mode_debug_clone_copy_eq() {
2409 let a = JournalMode::Wal;
2410 let b = a;
2411 assert_eq!(a, b);
2412 assert_ne!(JournalMode::Delete, JournalMode::Wal);
2413 let dbg = format!("{a:?}");
2414 assert!(dbg.contains("Wal"));
2415 }
2416
2417 #[test]
2418 fn test_checkpoint_result_clone_debug() {
2419 let result = CheckpointResult {
2420 total_frames: 50,
2421 frames_backfilled: 50,
2422 completed: true,
2423 wal_was_reset: true,
2424 requested_mode: CheckpointMode::Truncate,
2425 effective_mode: CheckpointMode::Truncate,
2426 };
2427 let cloned = result.clone();
2428 assert_eq!(result, cloned);
2429 let dbg = format!("{result:?}");
2430 assert!(dbg.contains("CheckpointResult"));
2431 assert!(dbg.contains("Truncate"));
2432 assert!(dbg.contains("wal_was_reset"));
2433 }
2434
2435 #[test]
2436 fn test_wal_publication_snapshot_clone_copy_debug() {
2437 let snap = WalPublicationSnapshot {
2438 publication_seq: 42,
2439 generation: test_wal_generation_identity(),
2440 last_commit_frame: Some(100),
2441 commit_count: 7,
2442 latest_frame_entries: 50,
2443 index_is_partial: false,
2444 };
2445 let copied = snap;
2446 assert_eq!(copied, snap);
2447 let dbg = format!("{snap:?}");
2448 assert!(dbg.contains("WalPublicationSnapshot"));
2449 assert!(dbg.contains("publication_seq"));
2450 assert!(dbg.contains("42"));
2451 }
2452
2453 #[test]
2454 fn test_checkpoint_mode_all_variants_debug() {
2455 for (mode, expected) in [
2456 (CheckpointMode::Passive, "Passive"),
2457 (CheckpointMode::Full, "Full"),
2458 (CheckpointMode::Restart, "Restart"),
2459 (CheckpointMode::Truncate, "Truncate"),
2460 ] {
2461 let dbg = format!("{mode:?}");
2462 assert!(dbg.contains(expected), "expected {expected} in {dbg}");
2463 let copy = mode;
2464 assert_eq!(mode, copy);
2465 }
2466 }
2467
2468 #[test]
2469 fn test_prepared_wal_frame_batch_page_data_and_frame_slice() {
2470 let frame_size = 32;
2471 let page_data_offset = 8;
2472 let mut frame_bytes = vec![0u8; frame_size * 2];
2473 frame_bytes[8] = 0xAA;
2474 frame_bytes[frame_size + 8] = 0xBB;
2475
2476 let batch = PreparedWalFrameBatch {
2477 frame_size,
2478 page_data_offset,
2479 big_endian_checksum: false,
2480 frame_metas: vec![
2481 PreparedWalFrameMeta {
2482 page_number: 1,
2483 db_size_if_commit: 0,
2484 },
2485 PreparedWalFrameMeta {
2486 page_number: 2,
2487 db_size_if_commit: 5,
2488 },
2489 ],
2490 checksum_transforms: Vec::new(),
2491 frame_bytes,
2492 last_commit_frame_offset: None,
2493 finalized_for: None,
2494 finalized_running_checksum: None,
2495 };
2496
2497 assert_eq!(batch.page_data(0)[0], 0xAA);
2498 assert_eq!(batch.page_data(1)[0], 0xBB);
2499 assert_eq!(batch.frame_slice(0).len(), frame_size);
2500 assert_eq!(batch.frame_slice(1).len(), frame_size);
2501
2502 let refs = batch.frame_refs();
2503 assert_eq!(refs.len(), 2);
2504 assert_eq!(refs[0].page_number, 1);
2505 assert_eq!(refs[1].db_size_if_commit, 5);
2506 assert_eq!(refs[0].page_data[0], 0xAA);
2507 assert_eq!(refs[1].page_data[0], 0xBB);
2508 }
2509
2510 #[test]
2511 fn prepared_wal_frame_meta_debug_clone_copy_eq() {
2512 let a = PreparedWalFrameMeta {
2513 page_number: 5,
2514 db_size_if_commit: 0,
2515 };
2516 let b = PreparedWalFrameMeta {
2517 page_number: 5,
2518 db_size_if_commit: 10,
2519 };
2520 let copied = a;
2521 assert_eq!(copied, a);
2522 assert_ne!(a, b);
2523 let dbg = format!("{a:?}");
2524 assert!(dbg.contains("PreparedWalFrameMeta"));
2525 assert!(dbg.contains("5"));
2526 }
2527
2528 #[test]
2529 fn prepared_wal_checksum_seed_default_and_eq() {
2530 let def = PreparedWalChecksumSeed::default();
2531 assert_eq!(def.s1, 0);
2532 assert_eq!(def.s2, 0);
2533 let other = PreparedWalChecksumSeed { s1: 1, s2: 2 };
2534 assert_ne!(def, other);
2535 let copied = other;
2536 assert_eq!(copied, other);
2537 let dbg = format!("{def:?}");
2538 assert!(dbg.contains("PreparedWalChecksumSeed"));
2539 }
2540
2541 #[test]
2542 fn prepared_wal_finalization_state_default_and_eq() {
2543 let def = PreparedWalFinalizationState::default();
2544 assert_eq!(def.checkpoint_seq, 0);
2545 assert_eq!(def.salt1, 0);
2546 assert_eq!(def.salt2, 0);
2547 assert_eq!(def.start_frame_index, 0);
2548 assert_eq!(def.seed, PreparedWalChecksumSeed::default());
2549 let other = PreparedWalFinalizationState {
2550 checkpoint_seq: 1,
2551 salt1: 0xAA,
2552 salt2: 0xBB,
2553 start_frame_index: 42,
2554 seed: PreparedWalChecksumSeed { s1: 10, s2: 20 },
2555 };
2556 assert_ne!(def, other);
2557 let copied = other;
2558 assert_eq!(copied, other);
2559 let dbg = format!("{other:?}");
2560 assert!(dbg.contains("PreparedWalFinalizationState"));
2561 }
2562
2563 #[test]
2564 fn transaction_mode_all_variants_debug_copy_eq() {
2565 let variants = [
2566 (TransactionMode::Deferred, "Deferred"),
2567 (TransactionMode::Immediate, "Immediate"),
2568 (TransactionMode::Exclusive, "Exclusive"),
2569 (TransactionMode::Concurrent, "Concurrent"),
2570 (TransactionMode::ReadOnly, "ReadOnly"),
2571 ];
2572 for (mode, expected) in &variants {
2573 let dbg = format!("{mode:?}");
2574 assert!(dbg.contains(expected), "expected {expected} in {dbg}");
2575 let copied = *mode;
2576 assert_eq!(copied, *mode);
2577 }
2578 assert_ne!(TransactionMode::Deferred, TransactionMode::Concurrent);
2579 }
2580
2581 #[test]
2582 fn wal_frame_ref_debug_clone_copy() {
2583 let data = [0xABu8; 16];
2584 let frame = WalFrameRef {
2585 page_number: 3,
2586 page_data: &data,
2587 db_size_if_commit: 0,
2588 };
2589 let copied = frame;
2590 assert_eq!(copied.page_number, 3);
2591 assert_eq!(copied.page_data.len(), 16);
2592 assert_eq!(copied.db_size_if_commit, 0);
2593 let dbg = format!("{frame:?}");
2594 assert!(dbg.contains("WalFrameRef"));
2595 }
2596
2597 #[test]
2598 fn mock_checkpoint_page_writer_default_and_trait_methods() {
2599 asupersync::test_utils::run_test(|| async {
2600 let mut writer = MockCheckpointPageWriter;
2601 let cx = Cx::new();
2602 let page = PageNumber::new(1).unwrap();
2603 writer.write_page(&cx, page, &[0u8; 4096]).await.unwrap();
2604 writer.truncate(&cx, 10).await.unwrap();
2605 writer.sync(&cx).await.unwrap();
2606 let dbg = format!("{writer:?}");
2607 assert!(dbg.contains("MockCheckpointPageWriter"));
2608 });
2609 }
2610
2611 #[test]
2612 fn transaction_kind_drained_debug() {
2613 let kind = TransactionKind::Drained;
2614 let dbg = format!("{kind:?}");
2615 assert!(dbg.contains("Drained"));
2616 }
2617
2618 #[test]
2619 fn wal_publication_snapshot_authoritative_boundary() {
2620 let base = WalPublicationSnapshot {
2621 publication_seq: 1,
2622 generation: test_wal_generation_identity(),
2623 last_commit_frame: Some(10),
2624 commit_count: 5,
2625 latest_frame_entries: 10,
2626 index_is_partial: false,
2627 };
2628 assert!(base.lookup_contract_is_authoritative());
2629 let partial = WalPublicationSnapshot {
2630 index_is_partial: true,
2631 ..base
2632 };
2633 assert!(!partial.lookup_contract_is_authoritative());
2634 }
2635}