diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 40946b9..ab37a66 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -385,7 +385,11 @@ jobs: --mountpoint ./fusemnt \ --namespaces "db=test" & sleep 1 - ./speedtest1 ./fusemnt/db + set +e + ./speedtest1 ./fusemnt/db 2>&1 | tee log + set -e + grep "SQL error: no such table: t1" log + echo "Got the expected error." build-ycsb: name: Build ycsb runs-on: ubuntu-20.04 diff --git a/mvfs/src/vfs.rs b/mvfs/src/vfs.rs index a18b003..13b5152 100644 --- a/mvfs/src/vfs.rs +++ b/mvfs/src/vfs.rs @@ -114,6 +114,7 @@ impl MultiVersionVfs { fixed_version, txn: None, lock: LockKind::None, + commit_confirmed: false, history: TransitionHistory::default(), sector_size: self.sector_size, first_page, @@ -136,6 +137,7 @@ pub struct Connection { fixed_version: Option, txn: Option, lock: LockKind, + commit_confirmed: bool, history: TransitionHistory, sector_size: usize, @@ -368,7 +370,7 @@ impl Connection { Ok(()) } - async fn finalize_transaction(&mut self) -> bool { + async fn finalize_transaction(&mut self, commit: bool) -> bool { if self.write_buffer.len() > WRITE_CHUNK_SIZE.load(Ordering::Relaxed) { self.force_flush_write_buffer().await; } @@ -394,15 +396,25 @@ impl Connection { .chain(self.write_buffer.keys().copied()) .collect::>(); - if self.fixed_version.is_some() { + let mut discard_reason: Option<&'static str> = None; + + if !commit { + // sqlite didn't confirm to commit + discard_reason = Some("did not receive confirmation from sqlite"); + } else if self.fixed_version.is_some() { // Disallow write to snapshot + discard_reason = Some("write to snapshot is dropped"); + } + + if let Some(discard_reason) = discard_reason { for index in &dirty_pages { self.page_cache.invalidate(index); } self.write_buffer.clear(); tracing::warn!( dirty_page_count = dirty_pages.len(), - "discarding write to snapshot" + reason = discard_reason, + "discarding transaction" ); return true; } @@ -654,6 +666,7 @@ impl Connection { self.virtual_version_counter = self.virtual_version_counter.wrapping_add(2); } self.lock = lock; + assert!(!self.commit_confirmed); Ok(true) } @@ -669,7 +682,9 @@ impl Connection { let reserved_level = LockKind::Reserved.level(); let commit_ok = if prev_lock.level() >= reserved_level && lock.level() < reserved_level { - self.finalize_transaction().await + let commit_confirmed = self.commit_confirmed; + self.commit_confirmed = false; + self.finalize_transaction(commit_confirmed).await } else { true }; @@ -683,6 +698,11 @@ impl Connection { Ok(commit_ok) } + pub fn confirm_commit(&mut self) { + assert!(!self.commit_confirmed); + self.commit_confirmed = true; + } + pub fn reserved(&mut self) -> Result { Ok(false) } diff --git a/mvsqlite/src/sqlite_vfs/mod.rs b/mvsqlite/src/sqlite_vfs/mod.rs index ba954c0..4543246 100644 --- a/mvsqlite/src/sqlite_vfs/mod.rs +++ b/mvsqlite/src/sqlite_vfs/mod.rs @@ -52,6 +52,8 @@ pub trait DatabaseHandle: Sync { self.lock(lock) } + fn commit_phasetwo(&mut self) {} + /// Check if the database this handle points to holds a [LockKind::Reserved], /// [LockKind::Pending] or [LockKind::Exclusive] lock. fn reserved(&mut self) -> Result; @@ -1392,7 +1394,10 @@ mod io { // Sent to the VFS after a transaction has been committed immediately but before the // database is unlocked. Silently ignored. - ffi::SQLITE_FCNTL_COMMIT_PHASETWO => ffi::SQLITE_OK, + ffi::SQLITE_FCNTL_COMMIT_PHASETWO => { + state.file.commit_phasetwo(); + ffi::SQLITE_OK + } // Used for debugging. Swap the file handle with the one pointed to by the pArg // argument. This capability is used during testing and only needs to be supported when diff --git a/mvsqlite/src/vfs.rs b/mvsqlite/src/vfs.rs index 2a3e658..513ddbd 100644 --- a/mvsqlite/src/vfs.rs +++ b/mvsqlite/src/vfs.rs @@ -103,6 +103,10 @@ impl DatabaseHandle for Connection { self.io.run(async { self.inner.unlock(lock.into()).await }) } + fn commit_phasetwo(&mut self) { + self.inner.confirm_commit(); + } + fn reserved(&mut self) -> Result { Ok(false) } @@ -155,6 +159,10 @@ impl DatabaseHandle for Box Result { (**self).reserved() }