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::{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)]
145pub enum ParallelWalCommitReconciliation {
146 Authorized,
149 NotCommitted,
151}
152
153pub type WalFuture<'a, T> = Pin<Box<dyn Future<Output = Result<T>> + Send + 'a>>;
162
163struct WalTrackedCompletionGuard(VfsWriteCompletion);
170
171impl WalTrackedCompletionGuard {
172 fn complete_success(&self) {
173 self.0.complete_success();
174 }
175
176 fn complete_error(&self) {
177 self.0.complete_error();
178 }
179}
180
181impl Drop for WalTrackedCompletionGuard {
182 fn drop(&mut self) {
183 self.0.complete_error();
184 }
185}
186
187pub trait WalBackend: Send + Sync {
188 fn begin_transaction<'a>(&'a mut self, _cx: &'a Cx) -> WalFuture<'a, ()> {
193 Box::pin(async { Ok(()) })
194 }
195
196 #[must_use]
201 fn published_snapshot(&self) -> Option<WalPublicationSnapshot> {
202 None
203 }
204
205 #[must_use]
210 fn pinned_read_snapshot(&self) -> Option<WalPublicationSnapshot> {
211 None
212 }
213
214 fn refresh_published_snapshot<'a>(
220 &'a mut self,
221 _cx: &'a Cx,
222 ) -> WalFuture<'a, Option<WalPublicationSnapshot>> {
223 Box::pin(async { Ok(self.published_snapshot()) })
224 }
225
226 fn publish_authorized_deferred_commit<'a>(&'a mut self, _cx: &'a Cx) -> WalFuture<'a, ()> {
234 Box::pin(async { Ok(()) })
235 }
236
237 fn append_frame<'a>(
244 &'a mut self,
245 cx: &'a Cx,
246 page_number: u32,
247 page_data: &'a [u8],
248 db_size_if_commit: u32,
249 ) -> WalFuture<'a, ()>;
250
251 fn append_frames<'a>(
256 &'a mut self,
257 cx: &'a Cx,
258 frames: &'a [WalFrameRef<'a>],
259 ) -> WalFuture<'a, ()> {
260 Box::pin(async move {
261 for frame in frames {
262 self.append_frame(
263 cx,
264 frame.page_number,
265 frame.page_data,
266 frame.db_size_if_commit,
267 )
268 .await?;
269 }
270 Ok(())
271 })
272 }
273
274 fn append_frames_tracked<'a>(
283 &'a mut self,
284 cx: &'a Cx,
285 frames: &'a [WalFrameRef<'a>],
286 completion: VfsWriteCompletion,
287 ) -> WalFuture<'a, ()> {
288 let completion = WalTrackedCompletionGuard(completion);
289 Box::pin(async move {
290 let result = self.append_frames(cx, frames).await;
291 if result.is_ok() {
292 completion.complete_success();
293 } else {
294 completion.complete_error();
295 }
296 result
297 })
298 }
299
300 fn prepare_append_frames(
306 &self,
307 _frames: &[WalFrameRef<'_>],
308 ) -> Result<Option<PreparedWalFrameBatch>> {
309 Ok(None)
310 }
311
312 fn finalize_prepared_frames(
319 &self,
320 _cx: &Cx,
321 _prepared: &mut PreparedWalFrameBatch,
322 ) -> Result<()> {
323 Ok(())
324 }
325
326 fn append_prepared_frames<'a>(
332 &'a mut self,
333 cx: &'a Cx,
334 prepared: &'a mut PreparedWalFrameBatch,
335 ) -> WalFuture<'a, ()> {
336 Box::pin(async move {
337 for index in 0..prepared.frame_count() {
338 let meta = prepared.frame_metas[index];
339 self.append_frame(
340 cx,
341 meta.page_number,
342 prepared.page_data(index),
343 meta.db_size_if_commit,
344 )
345 .await?;
346 }
347 Ok(())
348 })
349 }
350
351 fn append_prepared_frames_tracked<'a>(
353 &'a mut self,
354 cx: &'a Cx,
355 prepared: &'a mut PreparedWalFrameBatch,
356 completion: VfsWriteCompletion,
357 ) -> WalFuture<'a, ()> {
358 let completion = WalTrackedCompletionGuard(completion);
359 Box::pin(async move {
360 let result = self.append_prepared_frames(cx, prepared).await;
361 if result.is_ok() {
362 completion.complete_success();
363 } else {
364 completion.complete_error();
365 }
366 result
367 })
368 }
369
370 fn persist_parallel_wal_commit_certificate<'a>(
385 &'a mut self,
386 _cx: &'a Cx,
387 _certificate: &'a ParallelWalCommitCertificate,
388 _wal_frame_start: u64,
389 _wal_frame_end: u64,
390 _sync: bool,
391 ) -> WalFuture<'a, ()> {
392 Box::pin(async { Err(FrankenError::Unsupported) })
393 }
394
395 fn persist_parallel_wal_commit_certificate_tracked<'a>(
398 &'a mut self,
399 cx: &'a Cx,
400 certificate: &'a ParallelWalCommitCertificate,
401 wal_frame_start: u64,
402 wal_frame_end: u64,
403 sync: bool,
404 completion: VfsWriteCompletion,
405 ) -> WalFuture<'a, ()> {
406 let completion = WalTrackedCompletionGuard(completion);
407 Box::pin(async move {
408 let result = self
409 .persist_parallel_wal_commit_certificate(
410 cx,
411 certificate,
412 wal_frame_start,
413 wal_frame_end,
414 sync,
415 )
416 .await;
417 if result.is_ok() {
418 completion.complete_success();
419 } else {
420 completion.complete_error();
421 }
422 result
423 })
424 }
425
426 fn reconcile_parallel_wal_commit<'a>(
437 &'a mut self,
438 _cx: &'a Cx,
439 _certificate: &'a ParallelWalCommitCertificate,
440 _wal_frame_start: u64,
441 _wal_frame_end: u64,
442 _sync: bool,
443 ) -> WalFuture<'a, ParallelWalCommitReconciliation> {
444 Box::pin(async { Err(FrankenError::Unsupported) })
445 }
446
447 fn latest_authorized_parallel_wal_commit_certificate<'a>(
455 &'a mut self,
456 _cx: &'a Cx,
457 ) -> WalFuture<'a, Option<ParallelWalCommitCertificate>> {
458 Box::pin(async { Ok(None) })
459 }
460
461 fn read_page<'a>(&'a mut self, cx: &'a Cx, page_number: u32) -> WalFuture<'a, Option<Vec<u8>>>;
468
469 fn read_page_pinned<'a>(
482 &'a self,
483 _cx: &'a Cx,
484 _page_number: u32,
485 ) -> WalFuture<'a, Option<Vec<u8>>> {
486 Box::pin(async {
487 Err(FrankenError::internal(
490 "read_page_pinned not supported by this WalBackend; use read_page",
491 ))
492 })
493 }
494
495 fn supports_pinned_reads(&self) -> bool {
499 false
500 }
501
502 fn committed_txns_since_page<'a>(
509 &'a mut self,
510 _cx: &'a Cx,
511 _page_number: u32,
512 ) -> WalFuture<'a, u64> {
513 Box::pin(async { Ok(0) })
514 }
515
516 fn conflicting_pages_since_snapshot<'a>(
525 &'a mut self,
526 _cx: &'a Cx,
527 _snapshot: TransactionConflictSnapshot,
528 _page_numbers: &'a [u32],
529 _page_baselines: &'a [TransactionConflictPageBaseline],
530 ) -> WalFuture<'a, Vec<u32>> {
531 Box::pin(async { Ok(Vec::new()) })
532 }
533
534 fn committed_txn_count<'a>(&'a mut self, _cx: &'a Cx) -> WalFuture<'a, u64> {
541 Box::pin(async { Ok(0) })
542 }
543
544 fn sync(&mut self, cx: &Cx) -> Result<()>;
546
547 fn frame_count(&self) -> usize;
549
550 fn checkpoint<'a>(
567 &'a mut self,
568 cx: &'a Cx,
569 mode: CheckpointMode,
570 writer: &'a mut dyn CheckpointPageWriter,
571 backfilled_frames: u32,
572 oldest_reader_frame: Option<u32>,
573 ) -> WalFuture<'a, CheckpointResult>;
574}
575
576#[derive(Debug, Clone, Copy)]
578pub struct WalFrameRef<'a> {
579 pub page_number: u32,
581 pub page_data: &'a [u8],
583 pub db_size_if_commit: u32,
585}
586
587#[derive(Debug, Clone, Copy, PartialEq, Eq)]
589pub struct PreparedWalFrameMeta {
590 pub page_number: u32,
592 pub db_size_if_commit: u32,
594}
595
596pub type PreparedWalChecksumTransform = WalChecksumTransform;
601
602#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
604pub struct PreparedWalChecksumSeed {
605 pub s1: u32,
607 pub s2: u32,
609}
610
611#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
616pub struct PreparedWalFinalizationState {
617 pub checkpoint_seq: u32,
619 pub salt1: u32,
621 pub salt2: u32,
623 pub start_frame_index: usize,
625 pub seed: PreparedWalChecksumSeed,
627}
628
629#[derive(Debug, Clone, PartialEq, Eq)]
631pub struct PreparedWalFrameBatch {
632 pub frame_size: usize,
634 pub page_data_offset: usize,
636 pub big_endian_checksum: bool,
638 pub frame_metas: Vec<PreparedWalFrameMeta>,
640 pub checksum_transforms: Vec<PreparedWalChecksumTransform>,
642 pub frame_bytes: Vec<u8>,
644 pub last_commit_frame_offset: Option<usize>,
646 pub finalized_for: Option<PreparedWalFinalizationState>,
648 pub finalized_running_checksum: Option<PreparedWalChecksumSeed>,
650}
651
652impl PreparedWalFrameBatch {
653 #[must_use]
655 pub fn frame_count(&self) -> usize {
656 self.frame_metas.len()
657 }
658
659 #[must_use]
661 pub fn page_size(&self) -> usize {
662 self.frame_size.saturating_sub(self.page_data_offset)
663 }
664
665 #[must_use]
667 pub fn frame_refs(&self) -> Vec<WalFrameRef<'_>> {
668 self.frame_metas
669 .iter()
670 .enumerate()
671 .map(|(index, meta)| {
672 let frame_start = index * self.frame_size;
673 let page_start = frame_start + self.page_data_offset;
674 let page_end = frame_start + self.frame_size;
675 WalFrameRef {
676 page_number: meta.page_number,
677 page_data: &self.frame_bytes[page_start..page_end],
678 db_size_if_commit: meta.db_size_if_commit,
679 }
680 })
681 .collect()
682 }
683
684 #[must_use]
686 pub fn page_data(&self, index: usize) -> &[u8] {
687 let frame_start = index * self.frame_size;
688 let page_start = frame_start + self.page_data_offset;
689 let page_end = frame_start + self.frame_size;
690 &self.frame_bytes[page_start..page_end]
691 }
692
693 #[must_use]
695 pub fn frame_slice(&self, index: usize) -> &[u8] {
696 let frame_start = index * self.frame_size;
697 let frame_end = frame_start + self.frame_size;
698 &self.frame_bytes[frame_start..frame_end]
699 }
700
701 pub fn set_db_size_if_commit(&mut self, index: usize, db_size_if_commit: u32) {
703 self.frame_metas[index].db_size_if_commit = db_size_if_commit;
704 let frame_start = index * self.frame_size;
705 let db_size_offset = frame_start + 4;
706 self.frame_bytes[db_size_offset..db_size_offset + 4]
707 .copy_from_slice(&db_size_if_commit.to_be_bytes());
708 self.finalized_for = None;
709 self.finalized_running_checksum = None;
710 }
711
712 pub fn recompute_checksum_transforms(&mut self) -> Result<()> {
714 let page_size = self.page_size();
715 self.checksum_transforms = (0..self.frame_count())
716 .map(|index| {
717 WalChecksumTransform::for_wal_frame(
718 self.frame_slice(index),
719 page_size,
720 self.big_endian_checksum,
721 )
722 })
723 .collect::<Result<Vec<_>>>()?;
724 self.finalized_for = None;
725 self.finalized_running_checksum = None;
726 Ok(())
727 }
728}
729
730#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
739pub enum TransactionMode {
740 #[default]
743 Deferred,
744 Immediate,
748 Exclusive,
751 Concurrent,
758 ReadOnly,
761}
762
763pub trait MvccPager: sealed::Sealed + Send + Sync {
784 type Txn: TransactionHandle;
786
787 fn begin<'a>(
793 &'a self,
794 cx: &'a Cx,
795 mode: TransactionMode,
796 ) -> impl Future<Output = Result<Self::Txn>> + 'a;
797
798 fn journal_mode(&self) -> JournalMode;
800
801 fn is_readonly(&self) -> bool;
803
804 fn set_journal_mode<'a>(
812 &'a self,
813 cx: &'a Cx,
814 mode: JournalMode,
815 ) -> impl Future<Output = Result<JournalMode>> + 'a;
816
817 fn set_wal_backend(&self, backend: Box<dyn WalBackend>) -> Result<()>;
822}
823
824#[derive(Debug, Clone, Copy, PartialEq, Eq)]
836pub enum PagerCommitState {
837 NotCommitted,
839 InDoubt,
841 DurableNeedsPublication,
843 Committed,
845}
846
847impl PagerCommitState {
848 #[must_use]
850 pub const fn retains_commit_obligation(self) -> bool {
851 !matches!(self, Self::NotCommitted)
852 }
853}
854
855pub trait TransactionHandle: sealed::Sealed + Send {
870 fn get_page<'a>(
876 &'a self,
877 cx: &'a Cx,
878 page_no: PageNumber,
879 ) -> impl Future<Output = Result<PageData>> + 'a;
880
881 fn prefetch_page_hint(&self, _cx: &Cx, _page_no: PageNumber) {}
886
887 fn write_page<'a>(
892 &'a mut self,
893 cx: &'a Cx,
894 page_no: PageNumber,
895 data: &'a [u8],
896 ) -> impl Future<Output = Result<()>> + 'a;
897
898 fn write_page_data<'a>(
903 &'a mut self,
904 cx: &'a Cx,
905 page_no: PageNumber,
906 data: PageData,
907 ) -> impl Future<Output = Result<()>> + 'a {
908 async move { self.write_page(cx, page_no, data.as_bytes()).await }
909 }
910
911 fn try_take_staged_page_data(&mut self, _page_no: PageNumber) -> Option<PageData> {
918 None
919 }
920
921 fn try_mutate_staged_page_data(
927 &mut self,
928 _page_no: PageNumber,
929 _f: &mut dyn FnMut(&mut PageData),
930 ) -> bool {
931 false
932 }
933
934 fn restore_staged_page_data<'a>(
940 &'a mut self,
941 cx: &'a Cx,
942 page_no: PageNumber,
943 data: PageData,
944 ) -> impl Future<Output = Result<()>> + 'a {
945 async move { self.write_page_data(cx, page_no, data).await }
946 }
947
948 fn allocate_page<'a>(&'a mut self, cx: &'a Cx)
952 -> impl Future<Output = Result<PageNumber>> + 'a;
953
954 fn free_page<'a>(
956 &'a mut self,
957 cx: &'a Cx,
958 page_no: PageNumber,
959 ) -> impl Future<Output = Result<()>> + 'a;
960
961 fn commit<'a>(&'a mut self, cx: &'a Cx) -> impl Future<Output = Result<()>> + 'a;
967
968 fn pager_commit_state(&self) -> PagerCommitState {
974 PagerCommitState::NotCommitted
975 }
976
977 fn commit_and_retain<'a>(&'a mut self, cx: &'a Cx) -> impl Future<Output = Result<bool>> + 'a {
993 async move {
994 self.commit(cx).await?;
995 Ok(false)
996 }
997 }
998
999 fn is_writer(&self) -> bool;
1005
1006 fn has_pending_writes(&self) -> bool;
1011
1012 fn published_visible_commit_seq_hint(&self) -> Option<fsqlite_types::CommitSeq> {
1018 None
1019 }
1020
1021 fn pending_commit_pages(&self) -> Result<Vec<PageNumber>> {
1025 Ok(Vec::new())
1026 }
1027
1028 fn pending_conflict_pages(&self) -> Result<Vec<PageNumber>> {
1036 self.pending_commit_pages()
1037 }
1038
1039 fn pending_conflict_pages_conservative(&self) -> Vec<PageNumber> {
1054 self.write_set_page_numbers()
1055 }
1056
1057 fn write_set_page_numbers(&self) -> Vec<PageNumber> {
1060 Vec::new()
1061 }
1062
1063 fn page_one_in_pending_commit_surface(&self) -> Result<bool> {
1066 Ok(self.pending_commit_pages()?.contains(&PageNumber::ONE))
1067 }
1068
1069 fn page_size(&self) -> PageSize {
1074 PageSize::default()
1075 }
1076
1077 fn allocate_page_requires_page_one_conflict_tracking(&self) -> Result<bool> {
1086 Ok(true)
1087 }
1088
1089 fn free_page_requires_page_one_conflict_tracking(&self, _page_no: PageNumber) -> Result<bool> {
1098 Ok(true)
1099 }
1100
1101 fn write_page_requires_page_one_conflict_tracking(&self, _page_no: PageNumber) -> Result<bool> {
1111 Ok(true)
1112 }
1113
1114 fn rollback<'a>(&'a mut self, cx: &'a Cx) -> impl Future<Output = Result<()>> + 'a;
1120
1121 fn record_write_witness(&mut self, _cx: &Cx, _key: fsqlite_types::WitnessKey) {}
1126
1127 fn savepoint(&mut self, cx: &Cx, name: &str) -> Result<()>;
1133
1134 fn release_savepoint(&mut self, cx: &Cx, name: &str) -> Result<()>;
1140
1141 fn rollback_to_savepoint(&mut self, cx: &Cx, name: &str) -> Result<()>;
1148}
1149
1150pub trait CheckpointPageWriter: sealed::Sealed + Send {
1164 fn write_page<'a>(
1166 &'a mut self,
1167 cx: &'a Cx,
1168 page_no: PageNumber,
1169 data: &'a [u8],
1170 ) -> WalFuture<'a, ()>;
1171
1172 fn truncate<'a>(&'a mut self, cx: &'a Cx, n_pages: u32) -> WalFuture<'a, ()>;
1174
1175 fn sync<'a>(&'a mut self, cx: &'a Cx) -> WalFuture<'a, ()>;
1177}
1178
1179#[derive(Debug, Default, Clone, Copy)]
1185pub struct MockMvccPager;
1186
1187impl sealed::Sealed for MockMvccPager {}
1188
1189impl MvccPager for MockMvccPager {
1190 type Txn = MockTransaction;
1191
1192 fn begin<'a>(
1193 &'a self,
1194 _cx: &'a Cx,
1195 _mode: TransactionMode,
1196 ) -> impl Future<Output = Result<Self::Txn>> + 'a {
1197 async {
1198 Ok(MockTransaction {
1199 committed: false,
1200 next_page: 2,
1201 savepoint_names: Vec::new(),
1202 })
1203 }
1204 }
1205
1206 fn journal_mode(&self) -> JournalMode {
1207 JournalMode::Delete
1208 }
1209
1210 fn is_readonly(&self) -> bool {
1211 false
1212 }
1213
1214 fn set_journal_mode<'a>(
1215 &'a self,
1216 _cx: &'a Cx,
1217 mode: JournalMode,
1218 ) -> impl Future<Output = Result<JournalMode>> + 'a {
1219 async move { Ok(mode) }
1220 }
1221
1222 fn set_wal_backend(&self, _backend: Box<dyn WalBackend>) -> Result<()> {
1223 Ok(())
1224 }
1225}
1226
1227#[derive(Debug, Clone)]
1229pub struct MockTransaction {
1230 committed: bool,
1231 next_page: u32,
1232 savepoint_names: Vec<String>,
1233}
1234
1235impl sealed::Sealed for MockTransaction {}
1236
1237impl TransactionHandle for MockTransaction {
1238 fn get_page<'a>(
1239 &'a self,
1240 _cx: &'a Cx,
1241 page_no: PageNumber,
1242 ) -> impl Future<Output = Result<PageData>> + 'a {
1243 async move {
1244 let size = fsqlite_types::PageSize::default();
1245 let mut data = PageData::zeroed(size);
1246 data.as_bytes_mut()[..4].copy_from_slice(&page_no.get().to_le_bytes());
1247 Ok(data)
1248 }
1249 }
1250
1251 fn write_page<'a>(
1252 &'a mut self,
1253 _cx: &'a Cx,
1254 _page_no: PageNumber,
1255 _data: &'a [u8],
1256 ) -> impl Future<Output = Result<()>> + 'a {
1257 async { Ok(()) }
1258 }
1259
1260 fn allocate_page<'a>(
1261 &'a mut self,
1262 _cx: &'a Cx,
1263 ) -> impl Future<Output = Result<PageNumber>> + 'a {
1264 async move {
1265 let page = PageNumber::new(self.next_page)
1266 .expect("mock allocator must always produce non-zero page numbers");
1267 self.next_page += 1;
1268 Ok(page)
1269 }
1270 }
1271
1272 fn free_page<'a>(
1273 &'a mut self,
1274 _cx: &'a Cx,
1275 _page_no: PageNumber,
1276 ) -> impl Future<Output = Result<()>> + 'a {
1277 async { Ok(()) }
1278 }
1279
1280 fn commit<'a>(&'a mut self, _cx: &'a Cx) -> impl Future<Output = Result<()>> + 'a {
1281 async move {
1282 self.committed = true;
1283 Ok(())
1284 }
1285 }
1286
1287 fn pager_commit_state(&self) -> PagerCommitState {
1288 if self.committed {
1289 PagerCommitState::Committed
1290 } else {
1291 PagerCommitState::NotCommitted
1292 }
1293 }
1294
1295 fn is_writer(&self) -> bool {
1296 false
1297 }
1298
1299 fn has_pending_writes(&self) -> bool {
1300 false
1301 }
1302
1303 fn pending_commit_pages(&self) -> Result<Vec<PageNumber>> {
1304 Ok(Vec::new())
1305 }
1306
1307 fn rollback<'a>(&'a mut self, _cx: &'a Cx) -> impl Future<Output = Result<()>> + 'a {
1308 async { Ok(()) }
1309 }
1310
1311 fn record_write_witness(&mut self, _cx: &Cx, _key: fsqlite_types::WitnessKey) {}
1312
1313 fn savepoint(&mut self, _cx: &Cx, name: &str) -> Result<()> {
1314 self.savepoint_names.push(name.to_owned());
1315 Ok(())
1316 }
1317
1318 fn release_savepoint(&mut self, _cx: &Cx, name: &str) -> Result<()> {
1319 if let Some(pos) = self.savepoint_names.iter().rposition(|n| n == name) {
1320 self.savepoint_names.truncate(pos);
1321 Ok(())
1322 } else {
1323 Err(fsqlite_error::FrankenError::internal(format!(
1324 "no savepoint named '{name}'"
1325 )))
1326 }
1327 }
1328
1329 fn rollback_to_savepoint(&mut self, _cx: &Cx, name: &str) -> Result<()> {
1330 if let Some(pos) = self.savepoint_names.iter().rposition(|n| n == name) {
1331 self.savepoint_names.truncate(pos + 1);
1332 Ok(())
1333 } else {
1334 Err(fsqlite_error::FrankenError::internal(format!(
1335 "no savepoint named '{name}'"
1336 )))
1337 }
1338 }
1339}
1340
1341#[derive(Debug, Default, Clone, Copy)]
1344pub struct MemoryMockMvccPager;
1345
1346impl sealed::Sealed for MemoryMockMvccPager {}
1347
1348impl MvccPager for MemoryMockMvccPager {
1349 type Txn = MemoryMockTransaction;
1350
1351 fn begin<'a>(
1352 &'a self,
1353 _cx: &'a Cx,
1354 _mode: TransactionMode,
1355 ) -> impl Future<Output = Result<Self::Txn>> + 'a {
1356 async {
1357 Ok(MemoryMockTransaction {
1358 committed: false,
1359 next_page: 2,
1360 pages: HashMap::new(),
1361 savepoints: Vec::new(),
1362 })
1363 }
1364 }
1365
1366 fn journal_mode(&self) -> JournalMode {
1367 JournalMode::Delete
1368 }
1369
1370 fn is_readonly(&self) -> bool {
1371 false
1372 }
1373
1374 fn set_journal_mode<'a>(
1375 &'a self,
1376 _cx: &'a Cx,
1377 mode: JournalMode,
1378 ) -> impl Future<Output = Result<JournalMode>> + 'a {
1379 async move { Ok(mode) }
1380 }
1381
1382 fn set_wal_backend(&self, _backend: Box<dyn WalBackend>) -> Result<()> {
1383 Ok(())
1384 }
1385}
1386
1387#[derive(Debug, Clone)]
1388struct MemoryMockSavepoint {
1389 name: String,
1390 next_page: u32,
1391 pages: HashMap<PageNumber, PageData>,
1392}
1393
1394#[derive(Debug, Clone)]
1397pub struct MemoryMockTransaction {
1398 committed: bool,
1399 next_page: u32,
1400 pages: HashMap<PageNumber, PageData>,
1401 savepoints: Vec<MemoryMockSavepoint>,
1402}
1403
1404impl sealed::Sealed for MemoryMockTransaction {}
1405
1406impl TransactionHandle for MemoryMockTransaction {
1407 fn get_page<'a>(
1408 &'a self,
1409 _cx: &'a Cx,
1410 page_no: PageNumber,
1411 ) -> impl Future<Output = Result<PageData>> + 'a {
1412 async move {
1413 Ok(self
1414 .pages
1415 .get(&page_no)
1416 .cloned()
1417 .unwrap_or_else(|| PageData::zeroed(fsqlite_types::PageSize::default())))
1418 }
1419 }
1420
1421 fn write_page<'a>(
1422 &'a mut self,
1423 _cx: &'a Cx,
1424 page_no: PageNumber,
1425 data: &'a [u8],
1426 ) -> impl Future<Output = Result<()>> + 'a {
1427 async move {
1428 self.committed = false;
1429 let page_size = fsqlite_types::PageSize::default().as_usize();
1430 let mut page = vec![0_u8; page_size];
1431 let copy_len = data.len().min(page_size);
1432 page[..copy_len].copy_from_slice(&data[..copy_len]);
1433 self.pages.insert(page_no, PageData::from_vec(page));
1434 Ok(())
1435 }
1436 }
1437
1438 fn write_page_data<'a>(
1439 &'a mut self,
1440 _cx: &'a Cx,
1441 page_no: PageNumber,
1442 data: PageData,
1443 ) -> impl Future<Output = Result<()>> + 'a {
1444 async move {
1445 self.committed = false;
1446 let page_size = fsqlite_types::PageSize::default().as_usize();
1447 let mut page = vec![0_u8; page_size];
1448 let copy_len = data.len().min(page_size);
1449 page[..copy_len].copy_from_slice(&data.as_bytes()[..copy_len]);
1450 self.pages.insert(page_no, PageData::from_vec(page));
1451 Ok(())
1452 }
1453 }
1454
1455 fn allocate_page<'a>(
1456 &'a mut self,
1457 _cx: &'a Cx,
1458 ) -> impl Future<Output = Result<PageNumber>> + 'a {
1459 async move {
1460 self.committed = false;
1461 let page = PageNumber::new(self.next_page)
1462 .expect("mock allocator must always produce non-zero page numbers");
1463 self.next_page += 1;
1464 self.pages
1465 .entry(page)
1466 .or_insert_with(|| PageData::zeroed(fsqlite_types::PageSize::default()));
1467 Ok(page)
1468 }
1469 }
1470
1471 fn free_page<'a>(
1472 &'a mut self,
1473 _cx: &'a Cx,
1474 page_no: PageNumber,
1475 ) -> impl Future<Output = Result<()>> + 'a {
1476 async move {
1477 self.committed = false;
1478 self.pages.remove(&page_no);
1479 Ok(())
1480 }
1481 }
1482
1483 fn commit<'a>(&'a mut self, _cx: &'a Cx) -> impl Future<Output = Result<()>> + 'a {
1484 async move {
1485 self.committed = true;
1486 Ok(())
1487 }
1488 }
1489
1490 fn pager_commit_state(&self) -> PagerCommitState {
1491 if self.committed {
1492 PagerCommitState::Committed
1493 } else {
1494 PagerCommitState::NotCommitted
1495 }
1496 }
1497
1498 fn is_writer(&self) -> bool {
1499 !self.pages.is_empty()
1500 }
1501
1502 fn has_pending_writes(&self) -> bool {
1503 !self.committed && !self.pages.is_empty()
1504 }
1505
1506 fn pending_commit_pages(&self) -> Result<Vec<PageNumber>> {
1507 let mut pages = self.pages.keys().copied().collect::<Vec<_>>();
1508 pages.sort_unstable();
1509 Ok(pages)
1510 }
1511
1512 fn rollback<'a>(&'a mut self, _cx: &'a Cx) -> impl Future<Output = Result<()>> + 'a {
1513 async move {
1514 self.committed = false;
1515 self.next_page = 2;
1516 self.pages.clear();
1517 self.savepoints.clear();
1518 Ok(())
1519 }
1520 }
1521
1522 fn record_write_witness(&mut self, _cx: &Cx, _key: fsqlite_types::WitnessKey) {}
1523
1524 fn savepoint(&mut self, _cx: &Cx, name: &str) -> Result<()> {
1525 self.savepoints.push(MemoryMockSavepoint {
1526 name: name.to_owned(),
1527 next_page: self.next_page,
1528 pages: self.pages.clone(),
1529 });
1530 Ok(())
1531 }
1532
1533 fn release_savepoint(&mut self, _cx: &Cx, name: &str) -> Result<()> {
1534 if let Some(pos) = self.savepoints.iter().rposition(|sp| sp.name == name) {
1535 self.savepoints.truncate(pos);
1536 Ok(())
1537 } else {
1538 Err(fsqlite_error::FrankenError::internal(format!(
1539 "no savepoint named '{name}'"
1540 )))
1541 }
1542 }
1543
1544 fn rollback_to_savepoint(&mut self, _cx: &Cx, name: &str) -> Result<()> {
1545 if let Some(pos) = self.savepoints.iter().rposition(|sp| sp.name == name) {
1546 let snapshot = self.savepoints[pos].clone();
1547 self.next_page = snapshot.next_page;
1548 self.pages = snapshot.pages;
1549 self.savepoints.truncate(pos + 1);
1550 Ok(())
1551 } else {
1552 Err(fsqlite_error::FrankenError::internal(format!(
1553 "no savepoint named '{name}'"
1554 )))
1555 }
1556 }
1557}
1558
1559#[cfg_attr(
1562 target_arch = "wasm32",
1563 expect(
1564 clippy::large_enum_variant,
1565 reason = "native transaction variants are absent on wasm, making the intentional inline memory transaction an apparent size outlier"
1566 )
1567)]
1568pub enum TransactionKind {
1569 Memory(SimpleTransaction<MemoryVfs>),
1571 #[cfg(all(feature = "native", target_os = "linux"))]
1573 IoUring(SimpleTransaction<IoUringVfs>),
1574 #[cfg(all(feature = "native", unix))]
1576 Unix(SimpleTransaction<UnixVfs>),
1577 #[cfg(all(feature = "native", target_os = "windows"))]
1579 Windows(SimpleTransaction<WindowsVfs>),
1580 Mock(MockTransaction),
1582 MemoryMock(MemoryMockTransaction),
1584 Drained,
1589}
1590
1591impl std::fmt::Debug for TransactionKind {
1592 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1593 match self {
1594 Self::Memory(_) => f.write_str("TransactionKind::Memory"),
1595 #[cfg(all(feature = "native", target_os = "linux"))]
1596 Self::IoUring(_) => f.write_str("TransactionKind::IoUring"),
1597 #[cfg(all(feature = "native", unix))]
1598 Self::Unix(_) => f.write_str("TransactionKind::Unix"),
1599 #[cfg(all(feature = "native", target_os = "windows"))]
1600 Self::Windows(_) => f.write_str("TransactionKind::Windows"),
1601 Self::Mock(_) => f.write_str("TransactionKind::Mock"),
1602 Self::MemoryMock(_) => f.write_str("TransactionKind::MemoryMock"),
1603 Self::Drained => f.write_str("TransactionKind::Drained"),
1604 }
1605 }
1606}
1607
1608impl TransactionKind {
1609 #[must_use]
1616 pub fn live_freelist_pages(&self) -> Vec<PageNumber> {
1617 match self {
1618 Self::Memory(txn) => txn.live_freelist_pages(),
1619 #[cfg(all(feature = "native", target_os = "linux"))]
1620 Self::IoUring(txn) => txn.live_freelist_pages(),
1621 #[cfg(all(feature = "native", unix))]
1622 Self::Unix(txn) => txn.live_freelist_pages(),
1623 #[cfg(all(feature = "native", target_os = "windows"))]
1624 Self::Windows(txn) => txn.live_freelist_pages(),
1625 Self::Mock(_) | Self::MemoryMock(_) | Self::Drained => Vec::new(),
1626 }
1627 }
1628
1629 #[must_use]
1635 pub fn live_db_size(&self) -> u32 {
1636 match self {
1637 Self::Memory(txn) => txn.live_db_size(),
1638 #[cfg(all(feature = "native", target_os = "linux"))]
1639 Self::IoUring(txn) => txn.live_db_size(),
1640 #[cfg(all(feature = "native", unix))]
1641 Self::Unix(txn) => txn.live_db_size(),
1642 #[cfg(all(feature = "native", target_os = "windows"))]
1643 Self::Windows(txn) => txn.live_db_size(),
1644 Self::Mock(_) | Self::MemoryMock(_) | Self::Drained => 0,
1645 }
1646 }
1647
1648 #[must_use]
1652 pub fn snapshot_db_size(&self) -> u32 {
1653 match self {
1654 Self::Memory(txn) => txn.snapshot_db_size(),
1655 #[cfg(all(feature = "native", target_os = "linux"))]
1656 Self::IoUring(txn) => txn.snapshot_db_size(),
1657 #[cfg(all(feature = "native", unix))]
1658 Self::Unix(txn) => txn.snapshot_db_size(),
1659 #[cfg(all(feature = "native", target_os = "windows"))]
1660 Self::Windows(txn) => txn.snapshot_db_size(),
1661 Self::Mock(_) | Self::MemoryMock(_) | Self::Drained => 0,
1662 }
1663 }
1664
1665 #[must_use]
1669 pub fn visible_db_size_bound(&self) -> u32 {
1670 match self {
1671 Self::Memory(txn) => txn.visible_db_size_bound(),
1672 #[cfg(all(feature = "native", target_os = "linux"))]
1673 Self::IoUring(txn) => txn.visible_db_size_bound(),
1674 #[cfg(all(feature = "native", unix))]
1675 Self::Unix(txn) => txn.visible_db_size_bound(),
1676 #[cfg(all(feature = "native", target_os = "windows"))]
1677 Self::Windows(txn) => txn.visible_db_size_bound(),
1678 Self::Mock(_) | Self::MemoryMock(_) | Self::Drained => 0,
1679 }
1680 }
1681}
1682
1683macro_rules! dispatch_transaction_kind {
1684 ($value:expr, $txn:ident => $body:expr) => {
1685 match $value {
1686 TransactionKind::Memory($txn) => $body,
1687 #[cfg(all(feature = "native", target_os = "linux"))]
1688 TransactionKind::IoUring($txn) => $body,
1689 #[cfg(all(feature = "native", unix))]
1690 TransactionKind::Unix($txn) => $body,
1691 #[cfg(all(feature = "native", target_os = "windows"))]
1692 TransactionKind::Windows($txn) => $body,
1693 TransactionKind::Mock($txn) => $body,
1694 TransactionKind::MemoryMock($txn) => $body,
1695 TransactionKind::Drained => {
1696 panic!("BUG: TransactionKind::Drained accessed while the transaction was extracted")
1697 }
1698 }
1699 };
1700}
1701
1702impl From<SimpleTransaction<MemoryVfs>> for TransactionKind {
1703 fn from(txn: SimpleTransaction<MemoryVfs>) -> Self {
1704 Self::Memory(txn)
1705 }
1706}
1707
1708#[cfg(all(feature = "native", target_os = "linux"))]
1709impl From<SimpleTransaction<IoUringVfs>> for TransactionKind {
1710 fn from(txn: SimpleTransaction<IoUringVfs>) -> Self {
1711 Self::IoUring(txn)
1712 }
1713}
1714
1715#[cfg(all(feature = "native", unix))]
1716impl From<SimpleTransaction<UnixVfs>> for TransactionKind {
1717 fn from(txn: SimpleTransaction<UnixVfs>) -> Self {
1718 Self::Unix(txn)
1719 }
1720}
1721
1722#[cfg(all(feature = "native", target_os = "windows"))]
1723impl From<SimpleTransaction<WindowsVfs>> for TransactionKind {
1724 fn from(txn: SimpleTransaction<WindowsVfs>) -> Self {
1725 Self::Windows(txn)
1726 }
1727}
1728
1729impl From<MockTransaction> for TransactionKind {
1730 fn from(txn: MockTransaction) -> Self {
1731 Self::Mock(txn)
1732 }
1733}
1734
1735impl From<MemoryMockTransaction> for TransactionKind {
1736 fn from(txn: MemoryMockTransaction) -> Self {
1737 Self::MemoryMock(txn)
1738 }
1739}
1740
1741impl sealed::Sealed for TransactionKind {}
1742
1743impl TransactionHandle for TransactionKind {
1744 fn get_page<'a>(
1752 &'a self,
1753 cx: &'a Cx,
1754 page_no: PageNumber,
1755 ) -> impl Future<Output = Result<PageData>> + 'a {
1756 async move { dispatch_transaction_kind!(self, txn => txn.get_page(cx, page_no).await) }
1757 }
1758
1759 fn prefetch_page_hint(&self, cx: &Cx, page_no: PageNumber) {
1760 dispatch_transaction_kind!(self, txn => txn.prefetch_page_hint(cx, page_no));
1761 }
1762
1763 fn write_page<'a>(
1764 &'a mut self,
1765 cx: &'a Cx,
1766 page_no: PageNumber,
1767 data: &'a [u8],
1768 ) -> impl Future<Output = Result<()>> + 'a {
1769 async move { dispatch_transaction_kind!(self, txn => txn.write_page(cx, page_no, data).await) }
1770 }
1771
1772 fn write_page_data<'a>(
1773 &'a mut self,
1774 cx: &'a Cx,
1775 page_no: PageNumber,
1776 data: PageData,
1777 ) -> impl Future<Output = Result<()>> + 'a {
1778 async move {
1779 dispatch_transaction_kind!(self, txn => txn.write_page_data(cx, page_no, data).await)
1780 }
1781 }
1782
1783 fn try_mutate_staged_page_data(
1784 &mut self,
1785 page_no: PageNumber,
1786 f: &mut dyn FnMut(&mut PageData),
1787 ) -> bool {
1788 dispatch_transaction_kind!(self, txn => txn.try_mutate_staged_page_data(page_no, f))
1789 }
1790
1791 fn allocate_page<'a>(
1792 &'a mut self,
1793 cx: &'a Cx,
1794 ) -> impl Future<Output = Result<PageNumber>> + 'a {
1795 async move { dispatch_transaction_kind!(self, txn => txn.allocate_page(cx).await) }
1796 }
1797
1798 fn free_page<'a>(
1799 &'a mut self,
1800 cx: &'a Cx,
1801 page_no: PageNumber,
1802 ) -> impl Future<Output = Result<()>> + 'a {
1803 async move { dispatch_transaction_kind!(self, txn => txn.free_page(cx, page_no).await) }
1804 }
1805
1806 fn commit<'a>(&'a mut self, cx: &'a Cx) -> impl Future<Output = Result<()>> + 'a {
1807 async move { dispatch_transaction_kind!(self, txn => txn.commit(cx).await) }
1808 }
1809
1810 fn pager_commit_state(&self) -> PagerCommitState {
1811 dispatch_transaction_kind!(self, txn => txn.pager_commit_state())
1812 }
1813
1814 fn commit_and_retain<'a>(&'a mut self, cx: &'a Cx) -> impl Future<Output = Result<bool>> + 'a {
1815 async move { dispatch_transaction_kind!(self, txn => txn.commit_and_retain(cx).await) }
1816 }
1817
1818 fn is_writer(&self) -> bool {
1819 dispatch_transaction_kind!(self, txn => txn.is_writer())
1820 }
1821
1822 fn has_pending_writes(&self) -> bool {
1823 dispatch_transaction_kind!(self, txn => txn.has_pending_writes())
1824 }
1825
1826 fn published_visible_commit_seq_hint(&self) -> Option<fsqlite_types::CommitSeq> {
1827 dispatch_transaction_kind!(self, txn => txn.published_visible_commit_seq_hint())
1828 }
1829
1830 fn pending_commit_pages(&self) -> Result<Vec<PageNumber>> {
1831 dispatch_transaction_kind!(self, txn => txn.pending_commit_pages())
1832 }
1833
1834 fn pending_conflict_pages(&self) -> Result<Vec<PageNumber>> {
1835 dispatch_transaction_kind!(self, txn => txn.pending_conflict_pages())
1836 }
1837
1838 fn pending_conflict_pages_conservative(&self) -> Vec<PageNumber> {
1839 dispatch_transaction_kind!(self, txn => txn.pending_conflict_pages_conservative())
1840 }
1841
1842 fn write_set_page_numbers(&self) -> Vec<PageNumber> {
1843 dispatch_transaction_kind!(self, txn => txn.write_set_page_numbers())
1844 }
1845
1846 fn page_one_in_pending_commit_surface(&self) -> Result<bool> {
1847 dispatch_transaction_kind!(self, txn => txn.page_one_in_pending_commit_surface())
1848 }
1849
1850 fn page_size(&self) -> PageSize {
1851 dispatch_transaction_kind!(self, txn => txn.page_size())
1852 }
1853
1854 fn allocate_page_requires_page_one_conflict_tracking(&self) -> Result<bool> {
1855 dispatch_transaction_kind!(self, txn => txn.allocate_page_requires_page_one_conflict_tracking())
1856 }
1857
1858 fn free_page_requires_page_one_conflict_tracking(&self, page_no: PageNumber) -> Result<bool> {
1859 dispatch_transaction_kind!(self, txn => txn.free_page_requires_page_one_conflict_tracking(page_no))
1860 }
1861
1862 fn write_page_requires_page_one_conflict_tracking(&self, page_no: PageNumber) -> Result<bool> {
1863 dispatch_transaction_kind!(self, txn => txn.write_page_requires_page_one_conflict_tracking(page_no))
1864 }
1865
1866 fn rollback<'a>(&'a mut self, cx: &'a Cx) -> impl Future<Output = Result<()>> + 'a {
1867 async move { dispatch_transaction_kind!(self, txn => txn.rollback(cx).await) }
1868 }
1869
1870 fn record_write_witness(&mut self, cx: &Cx, key: fsqlite_types::WitnessKey) {
1871 dispatch_transaction_kind!(self, txn => txn.record_write_witness(cx, key));
1872 }
1873
1874 fn savepoint(&mut self, cx: &Cx, name: &str) -> Result<()> {
1875 dispatch_transaction_kind!(self, txn => txn.savepoint(cx, name))
1876 }
1877
1878 fn release_savepoint(&mut self, cx: &Cx, name: &str) -> Result<()> {
1879 dispatch_transaction_kind!(self, txn => txn.release_savepoint(cx, name))
1880 }
1881
1882 fn rollback_to_savepoint(&mut self, cx: &Cx, name: &str) -> Result<()> {
1883 dispatch_transaction_kind!(self, txn => txn.rollback_to_savepoint(cx, name))
1884 }
1885}
1886
1887#[derive(Debug, Default, Clone, Copy)]
1889pub struct MockCheckpointPageWriter;
1890
1891impl sealed::Sealed for MockCheckpointPageWriter {}
1892
1893impl CheckpointPageWriter for MockCheckpointPageWriter {
1894 fn write_page<'a>(
1895 &'a mut self,
1896 _cx: &'a Cx,
1897 _page_no: PageNumber,
1898 _data: &'a [u8],
1899 ) -> WalFuture<'a, ()> {
1900 Box::pin(async { Ok(()) })
1901 }
1902
1903 fn truncate<'a>(&'a mut self, _cx: &'a Cx, _n_pages: u32) -> WalFuture<'a, ()> {
1904 Box::pin(async { Ok(()) })
1905 }
1906
1907 fn sync<'a>(&'a mut self, _cx: &'a Cx) -> WalFuture<'a, ()> {
1908 Box::pin(async { Ok(()) })
1909 }
1910}
1911
1912#[cfg(test)]
1917mod tests {
1918 use super::*;
1919 use fsqlite_vfs::VfsWriteCompletionState;
1920 use std::task::Poll;
1921
1922 const fn test_wal_generation_identity() -> WalGenerationIdentity {
1925 WalGenerationIdentity {
1926 checkpoint_seq: 0,
1927 salts: fsqlite_wal::checksum::WalSalts { salt1: 0, salt2: 0 },
1928 }
1929 }
1930
1931 struct PendingTrackedWalBackend;
1932
1933 impl WalBackend for PendingTrackedWalBackend {
1934 fn append_frame<'a>(
1935 &'a mut self,
1936 _cx: &'a Cx,
1937 _page_number: u32,
1938 _page_data: &'a [u8],
1939 _db_size_if_commit: u32,
1940 ) -> WalFuture<'a, ()> {
1941 Box::pin(std::future::pending())
1942 }
1943
1944 fn read_page<'a>(
1945 &'a mut self,
1946 _cx: &'a Cx,
1947 _page_number: u32,
1948 ) -> WalFuture<'a, Option<Vec<u8>>> {
1949 Box::pin(async { Ok(None) })
1950 }
1951
1952 fn sync(&mut self, _cx: &Cx) -> Result<()> {
1953 Ok(())
1954 }
1955
1956 fn frame_count(&self) -> usize {
1957 0
1958 }
1959
1960 fn checkpoint<'a>(
1961 &'a mut self,
1962 _cx: &'a Cx,
1963 mode: CheckpointMode,
1964 _writer: &'a mut dyn CheckpointPageWriter,
1965 _backfilled_frames: u32,
1966 _oldest_reader_frame: Option<u32>,
1967 ) -> WalFuture<'a, CheckpointResult> {
1968 Box::pin(async move {
1969 Ok(CheckpointResult {
1970 total_frames: 0,
1971 frames_backfilled: 0,
1972 completed: true,
1973 wal_was_reset: false,
1974 requested_mode: mode,
1975 effective_mode: mode,
1976 })
1977 })
1978 }
1979 }
1980
1981 #[test]
1982 fn tracked_default_marks_unpolled_drop_terminal_error() {
1983 let cx = Cx::new();
1984 let data = [0_u8; 16];
1985 let frames = [WalFrameRef {
1986 page_number: 1,
1987 page_data: &data,
1988 db_size_if_commit: 1,
1989 }];
1990 let completion = VfsWriteCompletion::new();
1991 let mut backend = PendingTrackedWalBackend;
1992
1993 let future = backend.append_frames_tracked(&cx, &frames, completion.clone());
1994 assert_eq!(completion.state(), VfsWriteCompletionState::Pending);
1995 drop(future);
1996 assert_eq!(completion.state(), VfsWriteCompletionState::Error);
1997 }
1998
1999 #[test]
2000 fn tracked_default_marks_polled_drop_terminal_error() {
2001 let cx = Cx::new();
2002 let data = [0_u8; 16];
2003 let frames = [WalFrameRef {
2004 page_number: 1,
2005 page_data: &data,
2006 db_size_if_commit: 1,
2007 }];
2008 let completion = VfsWriteCompletion::new();
2009 let mut backend = PendingTrackedWalBackend;
2010 let mut future = Box::pin(backend.append_frames_tracked(&cx, &frames, completion.clone()));
2011
2012 let polled = std::future::poll_fn(|poll_cx| {
2013 assert!(future.as_mut().poll(poll_cx).is_pending());
2014 Poll::Ready(())
2015 });
2016 let runtime = asupersync::runtime::RuntimeBuilder::current_thread()
2017 .blocking_threads(1, 1)
2018 .build()
2019 .expect("tracked-default test runtime should build");
2020 runtime.block_on(polled);
2021 assert_eq!(completion.state(), VfsWriteCompletionState::Pending);
2022 drop(future);
2023 assert_eq!(completion.state(), VfsWriteCompletionState::Error);
2024 }
2025
2026 #[test]
2027 fn test_pager_trait_is_sealed_mock_impl() {
2028 asupersync::test_utils::run_test(|| async {
2029 let pager = MockMvccPager;
2032 let cx = Cx::new();
2033 let _txn = pager.begin(&cx, TransactionMode::Deferred).await.unwrap();
2034 });
2035 }
2036
2037 #[test]
2038 fn test_mvccpager_begin_commit_rollback_signatures() {
2039 asupersync::test_utils::run_test(|| async {
2040 let pager = MockMvccPager;
2041 let cx = Cx::new();
2042
2043 let mut txn = pager.begin(&cx, TransactionMode::ReadOnly).await.unwrap();
2045
2046 let page_no = PageNumber::new(1).unwrap();
2048 let data = txn.get_page(&cx, page_no).await.unwrap();
2049 assert_eq!(
2050 u32::from_le_bytes(data.as_bytes()[..4].try_into().unwrap()),
2051 1
2052 );
2053
2054 txn.write_page(&cx, page_no, &[0u8; 4096]).await.unwrap();
2055 let new_page = txn.allocate_page(&cx).await.unwrap();
2056 assert_eq!(new_page.get(), 2);
2057 txn.free_page(&cx, new_page).await.unwrap();
2058
2059 txn.commit(&cx).await.unwrap();
2060 });
2061 }
2062
2063 #[test]
2064 fn test_transaction_rollback_is_infallible() {
2065 asupersync::test_utils::run_test(|| async {
2066 let pager = MockMvccPager;
2067 let cx = Cx::new();
2068 let mut txn = pager.begin(&cx, TransactionMode::Deferred).await.unwrap();
2069 txn.rollback(&cx).await.unwrap();
2071 });
2072 }
2073
2074 #[test]
2075 fn test_checkpoint_page_writer_signatures() {
2076 asupersync::test_utils::run_test(|| async {
2077 let mut writer = MockCheckpointPageWriter;
2078 let cx = Cx::new();
2079 let page1 = PageNumber::new(1).unwrap();
2080
2081 writer.write_page(&cx, page1, &[0u8; 4096]).await.unwrap();
2082 writer.truncate(&cx, 10).await.unwrap();
2083 writer.sync(&cx).await.unwrap();
2084 });
2085 }
2086
2087 #[test]
2088 fn test_transaction_mode_default_is_deferred() {
2089 assert_eq!(TransactionMode::default(), TransactionMode::Deferred);
2090 }
2091
2092 #[test]
2093 fn test_open_traits_are_extensible() {
2094 fn assert_is_mvcc_pager<P: MvccPager<Txn = MockTransaction>>(_pager: &P) {}
2105 let pager = MockMvccPager;
2106 assert_is_mvcc_pager(&pager);
2107 }
2108
2109 #[test]
2110 fn test_memory_mock_transaction_persists_writes() {
2111 asupersync::test_utils::run_test(|| async {
2112 let pager = MemoryMockMvccPager;
2113 let cx = Cx::new();
2114 let mut txn = pager.begin(&cx, TransactionMode::Immediate).await.unwrap();
2115 let page_no = PageNumber::new(256).unwrap();
2116
2117 let mut bytes = vec![0_u8; fsqlite_types::PageSize::default().as_usize()];
2118 bytes[0] = 0x0A;
2119 txn.write_page(&cx, page_no, &bytes).await.unwrap();
2120
2121 let page = txn.get_page(&cx, page_no).await.unwrap();
2122 assert_eq!(page.as_bytes()[0], 0x0A);
2123 assert!(txn.has_pending_writes());
2124 assert!(txn.is_writer());
2125 });
2126 }
2127
2128 #[test]
2129 fn test_memory_mock_transaction_commit_clears_pending_writes() {
2130 asupersync::test_utils::run_test(|| async {
2131 let pager = MemoryMockMvccPager;
2132 let cx = Cx::new();
2133 let mut txn = pager.begin(&cx, TransactionMode::Immediate).await.unwrap();
2134 let page_no = PageNumber::new(2).unwrap();
2135
2136 txn.write_page(&cx, page_no, &[1_u8; 4096]).await.unwrap();
2137 assert!(txn.has_pending_writes());
2138
2139 txn.commit(&cx).await.unwrap();
2140 assert!(
2141 !txn.has_pending_writes(),
2142 "committed mock transactions must not report pending writes"
2143 );
2144 });
2145 }
2146
2147 #[test]
2148 fn test_memory_mock_transaction_rollback_resets_allocator() {
2149 asupersync::test_utils::run_test(|| async {
2150 let pager = MemoryMockMvccPager;
2151 let cx = Cx::new();
2152 let mut txn = pager.begin(&cx, TransactionMode::Immediate).await.unwrap();
2153
2154 assert_eq!(txn.allocate_page(&cx).await.unwrap().get(), 2);
2155 assert_eq!(txn.allocate_page(&cx).await.unwrap().get(), 3);
2156
2157 txn.rollback(&cx).await.unwrap();
2158
2159 assert_eq!(
2160 txn.allocate_page(&cx).await.unwrap().get(),
2161 2,
2162 "rollback should restore the mock allocator to its initial state"
2163 );
2164 });
2165 }
2166
2167 #[test]
2168 fn test_checkpoint_mode_default_is_passive() {
2169 assert_eq!(CheckpointMode::default(), CheckpointMode::Passive);
2170 }
2171
2172 #[test]
2173 fn test_journal_mode_default_is_delete() {
2174 assert_eq!(JournalMode::default(), JournalMode::Delete);
2175 }
2176
2177 #[test]
2178 fn test_wal_publication_snapshot_authoritative_when_index_full() {
2179 let snap = WalPublicationSnapshot {
2180 publication_seq: 1,
2181 generation: test_wal_generation_identity(),
2182 last_commit_frame: Some(10),
2183 commit_count: 5,
2184 latest_frame_entries: 10,
2185 index_is_partial: false,
2186 };
2187 assert!(
2188 snap.lookup_contract_is_authoritative(),
2189 "full index must be authoritative"
2190 );
2191 }
2192
2193 #[test]
2194 fn test_wal_publication_snapshot_not_authoritative_when_partial() {
2195 let snap = WalPublicationSnapshot {
2196 publication_seq: 1,
2197 generation: test_wal_generation_identity(),
2198 last_commit_frame: None,
2199 commit_count: 0,
2200 latest_frame_entries: 0,
2201 index_is_partial: true,
2202 };
2203 assert!(
2204 !snap.lookup_contract_is_authoritative(),
2205 "partial index must not be authoritative"
2206 );
2207 }
2208
2209 #[test]
2210 fn test_prepared_wal_frame_batch_frame_count_and_page_size() {
2211 let batch = PreparedWalFrameBatch {
2212 frame_size: 4120,
2213 page_data_offset: 24,
2214 big_endian_checksum: false,
2215 frame_metas: vec![
2216 PreparedWalFrameMeta {
2217 page_number: 1,
2218 db_size_if_commit: 0,
2219 },
2220 PreparedWalFrameMeta {
2221 page_number: 2,
2222 db_size_if_commit: 10,
2223 },
2224 ],
2225 checksum_transforms: Vec::new(),
2226 frame_bytes: vec![0u8; 4120 * 2],
2227 last_commit_frame_offset: Some(4120),
2228 finalized_for: None,
2229 finalized_running_checksum: None,
2230 };
2231 assert_eq!(batch.frame_count(), 2);
2232 assert_eq!(batch.page_size(), 4096);
2233 }
2234
2235 #[test]
2236 fn test_prepared_wal_frame_batch_set_db_size_clears_finalized() {
2237 let mut batch = PreparedWalFrameBatch {
2238 frame_size: 32,
2239 page_data_offset: 8,
2240 big_endian_checksum: false,
2241 frame_metas: vec![PreparedWalFrameMeta {
2242 page_number: 1,
2243 db_size_if_commit: 0,
2244 }],
2245 checksum_transforms: Vec::new(),
2246 frame_bytes: vec![0u8; 32],
2247 last_commit_frame_offset: None,
2248 finalized_for: Some(PreparedWalFinalizationState {
2249 checkpoint_seq: 1,
2250 salt1: 0xAA,
2251 salt2: 0xBB,
2252 start_frame_index: 0,
2253 seed: PreparedWalChecksumSeed::default(),
2254 }),
2255 finalized_running_checksum: Some(PreparedWalChecksumSeed { s1: 1, s2: 2 }),
2256 };
2257
2258 batch.set_db_size_if_commit(0, 42);
2259
2260 assert_eq!(batch.frame_metas[0].db_size_if_commit, 42);
2261 assert!(
2262 batch.finalized_for.is_none(),
2263 "set_db_size_if_commit must invalidate finalized_for"
2264 );
2265 assert!(
2266 batch.finalized_running_checksum.is_none(),
2267 "set_db_size_if_commit must invalidate finalized_running_checksum"
2268 );
2269 let db_bytes = &batch.frame_bytes[4..8];
2270 assert_eq!(u32::from_be_bytes(db_bytes.try_into().unwrap()), 42);
2271 }
2272
2273 #[test]
2274 fn test_mock_release_savepoint_unknown_name_returns_error() {
2275 asupersync::test_utils::run_test(|| async {
2276 let pager = MockMvccPager;
2277 let cx = Cx::new();
2278 let mut txn = pager.begin(&cx, TransactionMode::Deferred).await.unwrap();
2279
2280 let result = txn.release_savepoint(&cx, "nonexistent");
2281 assert!(result.is_err(), "releasing unknown savepoint must fail");
2282 });
2283 }
2284
2285 #[test]
2286 fn test_memory_mock_savepoint_rollback_restores_pages() {
2287 asupersync::test_utils::run_test(|| async {
2288 let pager = MemoryMockMvccPager;
2289 let cx = Cx::new();
2290 let mut txn = pager.begin(&cx, TransactionMode::Immediate).await.unwrap();
2291
2292 let p1 = PageNumber::new(1).unwrap();
2293 let page_size = fsqlite_types::PageSize::default().as_usize();
2294 let mut data_a = vec![0u8; page_size];
2295 data_a[0] = 0xAA;
2296 txn.write_page(&cx, p1, &data_a).await.unwrap();
2297
2298 txn.savepoint(&cx, "sp1").unwrap();
2299
2300 let mut data_b = vec![0u8; page_size];
2301 data_b[0] = 0xBB;
2302 txn.write_page(&cx, p1, &data_b).await.unwrap();
2303 assert_eq!(txn.get_page(&cx, p1).await.unwrap().as_bytes()[0], 0xBB);
2304
2305 txn.rollback_to_savepoint(&cx, "sp1").unwrap();
2306 assert_eq!(
2307 txn.get_page(&cx, p1).await.unwrap().as_bytes()[0],
2308 0xAA,
2309 "rollback_to_savepoint must restore page state"
2310 );
2311 });
2312 }
2313
2314 #[test]
2315 fn test_transaction_mode_default_trait_contract_is_deferred() {
2316 assert_eq!(TransactionMode::default(), TransactionMode::Deferred);
2317 }
2318
2319 #[test]
2320 fn test_checkpoint_result_fields() {
2321 let result = CheckpointResult {
2322 total_frames: 100,
2323 frames_backfilled: 80,
2324 completed: false,
2325 wal_was_reset: false,
2326 requested_mode: CheckpointMode::Full,
2327 effective_mode: CheckpointMode::Passive,
2328 };
2329 assert_eq!(result.total_frames, 100);
2330 assert_eq!(result.frames_backfilled, 80);
2331 assert!(!result.completed);
2332 assert_ne!(result.requested_mode, result.effective_mode);
2333 }
2334
2335 #[test]
2336 fn test_journal_mode_debug_clone_copy_eq() {
2337 let a = JournalMode::Wal;
2338 let b = a;
2339 assert_eq!(a, b);
2340 assert_ne!(JournalMode::Delete, JournalMode::Wal);
2341 let dbg = format!("{a:?}");
2342 assert!(dbg.contains("Wal"));
2343 }
2344
2345 #[test]
2346 fn test_checkpoint_result_clone_debug() {
2347 let result = CheckpointResult {
2348 total_frames: 50,
2349 frames_backfilled: 50,
2350 completed: true,
2351 wal_was_reset: true,
2352 requested_mode: CheckpointMode::Truncate,
2353 effective_mode: CheckpointMode::Truncate,
2354 };
2355 let cloned = result.clone();
2356 assert_eq!(result, cloned);
2357 let dbg = format!("{result:?}");
2358 assert!(dbg.contains("CheckpointResult"));
2359 assert!(dbg.contains("Truncate"));
2360 assert!(dbg.contains("wal_was_reset"));
2361 }
2362
2363 #[test]
2364 fn test_wal_publication_snapshot_clone_copy_debug() {
2365 let snap = WalPublicationSnapshot {
2366 publication_seq: 42,
2367 generation: test_wal_generation_identity(),
2368 last_commit_frame: Some(100),
2369 commit_count: 7,
2370 latest_frame_entries: 50,
2371 index_is_partial: false,
2372 };
2373 let copied = snap;
2374 assert_eq!(copied, snap);
2375 let dbg = format!("{snap:?}");
2376 assert!(dbg.contains("WalPublicationSnapshot"));
2377 assert!(dbg.contains("publication_seq"));
2378 assert!(dbg.contains("42"));
2379 }
2380
2381 #[test]
2382 fn test_checkpoint_mode_all_variants_debug() {
2383 for (mode, expected) in [
2384 (CheckpointMode::Passive, "Passive"),
2385 (CheckpointMode::Full, "Full"),
2386 (CheckpointMode::Restart, "Restart"),
2387 (CheckpointMode::Truncate, "Truncate"),
2388 ] {
2389 let dbg = format!("{mode:?}");
2390 assert!(dbg.contains(expected), "expected {expected} in {dbg}");
2391 let copy = mode;
2392 assert_eq!(mode, copy);
2393 }
2394 }
2395
2396 #[test]
2397 fn test_prepared_wal_frame_batch_page_data_and_frame_slice() {
2398 let frame_size = 32;
2399 let page_data_offset = 8;
2400 let mut frame_bytes = vec![0u8; frame_size * 2];
2401 frame_bytes[8] = 0xAA;
2402 frame_bytes[frame_size + 8] = 0xBB;
2403
2404 let batch = PreparedWalFrameBatch {
2405 frame_size,
2406 page_data_offset,
2407 big_endian_checksum: false,
2408 frame_metas: vec![
2409 PreparedWalFrameMeta {
2410 page_number: 1,
2411 db_size_if_commit: 0,
2412 },
2413 PreparedWalFrameMeta {
2414 page_number: 2,
2415 db_size_if_commit: 5,
2416 },
2417 ],
2418 checksum_transforms: Vec::new(),
2419 frame_bytes,
2420 last_commit_frame_offset: None,
2421 finalized_for: None,
2422 finalized_running_checksum: None,
2423 };
2424
2425 assert_eq!(batch.page_data(0)[0], 0xAA);
2426 assert_eq!(batch.page_data(1)[0], 0xBB);
2427 assert_eq!(batch.frame_slice(0).len(), frame_size);
2428 assert_eq!(batch.frame_slice(1).len(), frame_size);
2429
2430 let refs = batch.frame_refs();
2431 assert_eq!(refs.len(), 2);
2432 assert_eq!(refs[0].page_number, 1);
2433 assert_eq!(refs[1].db_size_if_commit, 5);
2434 assert_eq!(refs[0].page_data[0], 0xAA);
2435 assert_eq!(refs[1].page_data[0], 0xBB);
2436 }
2437
2438 #[test]
2439 fn prepared_wal_frame_meta_debug_clone_copy_eq() {
2440 let a = PreparedWalFrameMeta {
2441 page_number: 5,
2442 db_size_if_commit: 0,
2443 };
2444 let b = PreparedWalFrameMeta {
2445 page_number: 5,
2446 db_size_if_commit: 10,
2447 };
2448 let copied = a;
2449 assert_eq!(copied, a);
2450 assert_ne!(a, b);
2451 let dbg = format!("{a:?}");
2452 assert!(dbg.contains("PreparedWalFrameMeta"));
2453 assert!(dbg.contains("5"));
2454 }
2455
2456 #[test]
2457 fn prepared_wal_checksum_seed_default_and_eq() {
2458 let def = PreparedWalChecksumSeed::default();
2459 assert_eq!(def.s1, 0);
2460 assert_eq!(def.s2, 0);
2461 let other = PreparedWalChecksumSeed { s1: 1, s2: 2 };
2462 assert_ne!(def, other);
2463 let copied = other;
2464 assert_eq!(copied, other);
2465 let dbg = format!("{def:?}");
2466 assert!(dbg.contains("PreparedWalChecksumSeed"));
2467 }
2468
2469 #[test]
2470 fn prepared_wal_finalization_state_default_and_eq() {
2471 let def = PreparedWalFinalizationState::default();
2472 assert_eq!(def.checkpoint_seq, 0);
2473 assert_eq!(def.salt1, 0);
2474 assert_eq!(def.salt2, 0);
2475 assert_eq!(def.start_frame_index, 0);
2476 assert_eq!(def.seed, PreparedWalChecksumSeed::default());
2477 let other = PreparedWalFinalizationState {
2478 checkpoint_seq: 1,
2479 salt1: 0xAA,
2480 salt2: 0xBB,
2481 start_frame_index: 42,
2482 seed: PreparedWalChecksumSeed { s1: 10, s2: 20 },
2483 };
2484 assert_ne!(def, other);
2485 let copied = other;
2486 assert_eq!(copied, other);
2487 let dbg = format!("{other:?}");
2488 assert!(dbg.contains("PreparedWalFinalizationState"));
2489 }
2490
2491 #[test]
2492 fn transaction_mode_all_variants_debug_copy_eq() {
2493 let variants = [
2494 (TransactionMode::Deferred, "Deferred"),
2495 (TransactionMode::Immediate, "Immediate"),
2496 (TransactionMode::Exclusive, "Exclusive"),
2497 (TransactionMode::Concurrent, "Concurrent"),
2498 (TransactionMode::ReadOnly, "ReadOnly"),
2499 ];
2500 for (mode, expected) in &variants {
2501 let dbg = format!("{mode:?}");
2502 assert!(dbg.contains(expected), "expected {expected} in {dbg}");
2503 let copied = *mode;
2504 assert_eq!(copied, *mode);
2505 }
2506 assert_ne!(TransactionMode::Deferred, TransactionMode::Concurrent);
2507 }
2508
2509 #[test]
2510 fn wal_frame_ref_debug_clone_copy() {
2511 let data = [0xABu8; 16];
2512 let frame = WalFrameRef {
2513 page_number: 3,
2514 page_data: &data,
2515 db_size_if_commit: 0,
2516 };
2517 let copied = frame;
2518 assert_eq!(copied.page_number, 3);
2519 assert_eq!(copied.page_data.len(), 16);
2520 assert_eq!(copied.db_size_if_commit, 0);
2521 let dbg = format!("{frame:?}");
2522 assert!(dbg.contains("WalFrameRef"));
2523 }
2524
2525 #[test]
2526 fn mock_checkpoint_page_writer_default_and_trait_methods() {
2527 asupersync::test_utils::run_test(|| async {
2528 let mut writer = MockCheckpointPageWriter;
2529 let cx = Cx::new();
2530 let page = PageNumber::new(1).unwrap();
2531 writer.write_page(&cx, page, &[0u8; 4096]).await.unwrap();
2532 writer.truncate(&cx, 10).await.unwrap();
2533 writer.sync(&cx).await.unwrap();
2534 let dbg = format!("{writer:?}");
2535 assert!(dbg.contains("MockCheckpointPageWriter"));
2536 });
2537 }
2538
2539 #[test]
2540 fn transaction_kind_drained_debug() {
2541 let kind = TransactionKind::Drained;
2542 let dbg = format!("{kind:?}");
2543 assert!(dbg.contains("Drained"));
2544 }
2545
2546 #[test]
2547 fn wal_publication_snapshot_authoritative_boundary() {
2548 let base = WalPublicationSnapshot {
2549 publication_seq: 1,
2550 generation: test_wal_generation_identity(),
2551 last_commit_frame: Some(10),
2552 commit_count: 5,
2553 latest_frame_entries: 10,
2554 index_is_partial: false,
2555 };
2556 assert!(base.lookup_contract_is_authoritative());
2557 let partial = WalPublicationSnapshot {
2558 index_is_partial: true,
2559 ..base
2560 };
2561 assert!(!partial.lookup_contract_is_authoritative());
2562 }
2563}