From 757966379b73278d8c86c7eed1cad5d88ee5aa0d Mon Sep 17 00:00:00 2001 From: Hammed Abass Date: Sun, 12 Jul 2026 04:15:11 +0200 Subject: [PATCH] feat(core): add atomic receive history cleanup --- crates/vnidrop/src/repository.rs | 64 ++++++++ crates/vnidrop/src/runtime.rs | 17 +++ crates/vnidrop/src/tests/repository.rs | 143 ++++++++++++++++++ .../kotlin/com/vnidrop/app/core/CoreModels.kt | 1 + .../com/vnidrop/app/core/CoreRepository.kt | 6 + 5 files changed, 231 insertions(+) diff --git a/crates/vnidrop/src/repository.rs b/crates/vnidrop/src/repository.rs index fe347be..fedd924 100644 --- a/crates/vnidrop/src/repository.rs +++ b/crates/vnidrop/src/repository.rs @@ -27,6 +27,8 @@ pub(crate) struct Repository { pool: SqlitePool, #[cfg(test)] fail_next_write: Arc, + #[cfg(test)] + fail_receive_history_after_dependants: Arc, } pub(crate) struct TransferUpsert<'a> { @@ -80,6 +82,8 @@ impl Repository { pool, #[cfg(test)] fail_next_write: Arc::new(AtomicBool::new(false)), + #[cfg(test)] + fail_receive_history_after_dependants: Arc::new(AtomicBool::new(false)), }; repository.ensure_schema().await?; Ok(repository) @@ -482,6 +486,12 @@ impl Repository { self.fail_next_write.store(true, Ordering::SeqCst); } + #[cfg(test)] + pub(crate) fn fail_receive_history_after_dependants(&self) { + self.fail_receive_history_after_dependants + .store(true, Ordering::SeqCst); + } + #[cfg(test)] fn maybe_fail_write(&self) -> Result<()> { if self.fail_next_write.swap(false, Ordering::SeqCst) { @@ -759,6 +769,60 @@ impl Repository { Ok(()) } + pub(crate) async fn delete_receive_history(&self) -> Result { + self.maybe_fail_write()?; + let mut transaction = self.pool.begin().await?; + + // Delete dependants before their transfer rows. Keep the terminal-state + // predicate on every statement so receive work that is still active and + // every send record remain outside this transaction's scope. + sqlx::query( + r#" + DELETE FROM receiver_requests + WHERE transfer_id IN ( + SELECT transfer_id + FROM transfers + WHERE direction = 'receive' + AND status IN ('done', 'failed', 'cancelled') + ) + "#, + ) + .execute(&mut *transaction) + .await?; + sqlx::query( + r#" + DELETE FROM transfer_events + WHERE transfer_id IN ( + SELECT transfer_id + FROM transfers + WHERE direction = 'receive' + AND status IN ('done', 'failed', 'cancelled') + ) + "#, + ) + .execute(&mut *transaction) + .await?; + #[cfg(test)] + if self + .fail_receive_history_after_dependants + .swap(false, Ordering::SeqCst) + { + anyhow::bail!("injected receive history failure after dependant deletion"); + } + let deleted = sqlx::query( + r#" + DELETE FROM transfers + WHERE direction = 'receive' + AND status IN ('done', 'failed', 'cancelled') + "#, + ) + .execute(&mut *transaction) + .await?; + + transaction.commit().await?; + Ok(deleted.rows_affected()) + } + pub(crate) async fn list_events( &self, transfer_id: Option, diff --git a/crates/vnidrop/src/runtime.rs b/crates/vnidrop/src/runtime.rs index 311126f..9a8b75b 100644 --- a/crates/vnidrop/src/runtime.rs +++ b/crates/vnidrop/src/runtime.rs @@ -250,6 +250,12 @@ impl VnidropCore { .map_err(VnidropError::transfer) } + pub fn delete_receive_history(&self) -> Result { + self.runtime + .block_on(self.inner.delete_receive_history()) + .map_err(VnidropError::repository) + } + pub fn set_transfer_access_mode( &self, transfer_id: u64, @@ -992,9 +998,20 @@ impl CoreInner { .await .retain(|_, id| *id != transfer_id); self.access_policy.remove_transfer(transfer_id).await; + // Events are persisted asynchronously. Drain events emitted before this + // request so none can be written back after the transfer is deleted. + self.event_hub.flush().await; self.repository.delete_transfer(transfer_id).await } + async fn delete_receive_history(&self) -> Result { + // Transfer events are persisted on a background task. Drain everything + // emitted before this request so cleared history cannot be reinserted + // after the repository transaction commits. + self.event_hub.flush().await; + self.repository.delete_receive_history().await + } + async fn set_transfer_access_mode( &self, transfer_id: u64, diff --git a/crates/vnidrop/src/tests/repository.rs b/crates/vnidrop/src/tests/repository.rs index 1272591..63adcbf 100644 --- a/crates/vnidrop/src/tests/repository.rs +++ b/crates/vnidrop/src/tests/repository.rs @@ -585,6 +585,149 @@ async fn deleting_transfer_removes_related_history_transactionally() { .is_empty()); assert!(repository.delete_transfer(88).await.is_err()); } + +#[tokio::test] +async fn deleting_receive_history_only_removes_terminal_receives_and_dependants() { + let temp = tempfile::tempdir().unwrap(); + let repository = Repository::open(temp.path()).await.unwrap(); + let records = [ + (100, TransferDirection::Receive, TransferStatus::Done), + (101, TransferDirection::Receive, TransferStatus::Failed), + (102, TransferDirection::Receive, TransferStatus::Cancelled), + (103, TransferDirection::Receive, TransferStatus::Receiving), + (104, TransferDirection::Send, TransferStatus::Done), + (105, TransferDirection::Send, TransferStatus::Sharing), + ]; + + for (transfer_id, direction, status) in records { + repository + .insert_transfer(transfer(transfer_id, direction, status)) + .await + .unwrap(); + let request_id = format!("request-{transfer_id}"); + repository + .insert_receiver_request(ReceiverRequestInsert { + id: &request_id, + transfer_id, + remote_endpoint_id: "receiver", + transfer_name: "demo", + receiver_name: None, + receiver_device_name: None, + app_version: "1.0", + }) + .await + .unwrap(); + repository + .insert_event( + &CoreEvent { + id: format!("event-{transfer_id}"), + timestamp: transfer_id as i64, + scope: "transfer".to_string(), + transfer_id: Some(transfer_id), + direction: Some(direction.as_str().to_string()), + phase: "test".to_string(), + kind: "created".to_string(), + data_json: "{}".to_string(), + }, + 500, + ) + .await + .unwrap(); + } + + assert_eq!(repository.delete_receive_history().await.unwrap(), 3); + + let remaining = repository.list_transfers().await.unwrap(); + assert_eq!(remaining.len(), 3); + for transfer_id in [103, 104, 105] { + assert!(remaining + .iter() + .any(|transfer| transfer.transfer_id == transfer_id)); + assert_eq!( + repository + .list_receiver_requests(transfer_id) + .await + .unwrap() + .len(), + 1 + ); + assert_eq!( + repository + .list_events(Some(transfer_id), 500) + .await + .unwrap() + .len(), + 1 + ); + } + for transfer_id in [100, 101, 102] { + assert!(repository + .list_receiver_requests(transfer_id) + .await + .unwrap() + .is_empty()); + assert!(repository + .list_events(Some(transfer_id), 500) + .await + .unwrap() + .is_empty()); + } + assert_eq!(repository.delete_receive_history().await.unwrap(), 0); +} + +#[tokio::test] +async fn receive_history_mid_transaction_failure_preserves_all_related_rows() { + let temp = tempfile::tempdir().unwrap(); + let repository = Repository::open(temp.path()).await.unwrap(); + repository + .insert_transfer(transfer( + 106, + TransferDirection::Receive, + TransferStatus::Done, + )) + .await + .unwrap(); + repository + .insert_receiver_request(ReceiverRequestInsert { + id: "request-preserved", + transfer_id: 106, + remote_endpoint_id: "receiver", + transfer_name: "demo", + receiver_name: None, + receiver_device_name: None, + app_version: "1.0", + }) + .await + .unwrap(); + repository + .insert_event( + &CoreEvent { + id: "event-preserved".to_string(), + timestamp: 1, + scope: "transfer".to_string(), + transfer_id: Some(106), + direction: Some("receive".to_string()), + phase: "test".to_string(), + kind: "created".to_string(), + data_json: "{}".to_string(), + }, + 500, + ) + .await + .unwrap(); + repository.fail_receive_history_after_dependants(); + + assert!(repository.delete_receive_history().await.is_err()); + assert_eq!(repository.list_transfers().await.unwrap().len(), 1); + assert_eq!( + repository.list_receiver_requests(106).await.unwrap().len(), + 1 + ); + assert_eq!( + repository.list_events(Some(106), 500).await.unwrap().len(), + 1 + ); +} use std::str::FromStr; use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions}; diff --git a/shared/src/commonMain/kotlin/com/vnidrop/app/core/CoreModels.kt b/shared/src/commonMain/kotlin/com/vnidrop/app/core/CoreModels.kt index 2e62e26..1791ee5 100644 --- a/shared/src/commonMain/kotlin/com/vnidrop/app/core/CoreModels.kt +++ b/shared/src/commonMain/kotlin/com/vnidrop/app/core/CoreModels.kt @@ -146,6 +146,7 @@ interface CoreGateway { suspend fun receiveIntoSecurityScopedDirectory(ticket: String, outputDirectoryUrl: String, receiverName: String): Result suspend fun cancel(transferId: ULong): Result suspend fun delete(transferId: ULong): Result + suspend fun clearReceiveHistory(): Result suspend fun receiverRequests(transferId: ULong): Result> suspend fun respondReceiverRequest(requestId: String, accepted: Boolean, reason: String? = null): Result suspend fun refresh(): Result diff --git a/shared/src/commonMain/kotlin/com/vnidrop/app/core/CoreRepository.kt b/shared/src/commonMain/kotlin/com/vnidrop/app/core/CoreRepository.kt index 4b0867f..ce9ba71 100644 --- a/shared/src/commonMain/kotlin/com/vnidrop/app/core/CoreRepository.kt +++ b/shared/src/commonMain/kotlin/com/vnidrop/app/core/CoreRepository.kt @@ -173,6 +173,12 @@ class CoreRepository( _signals.tryEmit(CoreSignal.ReceiverHistoryChanged(transferId)) } + override suspend fun clearReceiveHistory(): Result = runCore { + val deleted = requireCore().deleteReceiveHistory() + refreshSnapshot() + deleted + } + override suspend fun receiverRequests(transferId: ULong): Result> = runCore { requireCore().listReceiverRequests(transferId).map(ReceiverRequest::toModel) }