From dc862bc01c995489a2fd9ead0630562fec0c9410 Mon Sep 17 00:00:00 2001 From: Hammed Abass Date: Sat, 11 Jul 2026 09:30:56 +0200 Subject: [PATCH] feat(core): track delivery receipts and transfer deletion --- Cargo.lock | 1 + crates/vnidrop/Cargo.toml | 1 + crates/vnidrop/src/api.rs | 1 + crates/vnidrop/src/approval.rs | 97 ++++++++++++++++++- crates/vnidrop/src/handshake.rs | 42 ++++++++- crates/vnidrop/src/repository.rs | 112 +++++++++++++++++++++- crates/vnidrop/src/runtime.rs | 126 ++++++++++++++++++++++--- crates/vnidrop/src/tests/repository.rs | 80 +++++++++++++++- crates/vnidrop/src/transfer_state.rs | 3 + crates/vnidrop/tests/approval.rs | 16 +++- crates/vnidrop/tests/lifecycle.rs | 20 ++++ 11 files changed, 471 insertions(+), 28 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 158af04..1261b9b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5056,6 +5056,7 @@ version = "0.1.0" dependencies = [ "anyhow", "async-channel", + "blake3", "bytes", "data-encoding", "futures", diff --git a/crates/vnidrop/Cargo.toml b/crates/vnidrop/Cargo.toml index 9925bab..7c1e77c 100644 --- a/crates/vnidrop/Cargo.toml +++ b/crates/vnidrop/Cargo.toml @@ -11,6 +11,7 @@ crate-type = ["cdylib", "staticlib", "rlib"] anyhow = "1.0.102" async-channel = "2.5.0" bytes = "1.11.1" +blake3 = "1.8.3" data-encoding = "2.11.0" futures = "0.3" futures-lite = "2.6.1" diff --git a/crates/vnidrop/src/api.rs b/crates/vnidrop/src/api.rs index 607c842..f7ad1b7 100644 --- a/crates/vnidrop/src/api.rs +++ b/crates/vnidrop/src/api.rs @@ -244,4 +244,5 @@ pub struct ReceiverRequest { pub reason: Option, pub requested_at: i64, pub responded_at: Option, + pub completed_at: Option, } diff --git a/crates/vnidrop/src/approval.rs b/crates/vnidrop/src/approval.rs index 543d337..b8ab720 100644 --- a/crates/vnidrop/src/approval.rs +++ b/crates/vnidrop/src/approval.rs @@ -8,7 +8,7 @@ use uuid::Uuid; use crate::{ access_policy::AccessPolicy, event_hub::EventHub, - handshake::{HandshakeResponse, RequestTransfer}, + handshake::{DeliveryReceipt, DeliveryReceiptResponse, HandshakeResponse, RequestTransfer}, repository::{ReceiverRequestInsert, Repository}, transfer_state::ReceiverRequestStatus, util::now_ms, @@ -35,6 +35,41 @@ pub(crate) struct ApprovalService { } impl ApprovalService { + pub(crate) async fn complete_delivery( + &self, + remote_endpoint_id: String, + receipt: DeliveryReceipt, + ) -> DeliveryReceiptResponse { + let token_hash = receipt_token_hash(&receipt.token); + match self + .repository + .complete_receiver_delivery( + &receipt.request_id, + receipt.transfer_id, + &remote_endpoint_id, + &token_hash, + ) + .await + { + Ok(()) => { + self.event_hub.emit_transfer( + receipt.transfer_id, + "send", + "delivery", + "receiver-completed", + json!({ "request_id": receipt.request_id, "remote_endpoint_id": remote_endpoint_id }), + ); + DeliveryReceiptResponse::Recorded + } + Err(error) => { + tracing::warn!(%error, "rejected receiver delivery receipt"); + DeliveryReceiptResponse::Rejected { + reason: "invalid-receipt".to_string(), + } + } + } + } + pub(crate) fn new( repository: Repository, event_hub: Arc, @@ -149,7 +184,41 @@ impl ApprovalService { remote_endpoint_id: String, request: RequestTransfer, ) -> HandshakeResponse { + let request_id = Uuid::new_v4().to_string(); + if self + .repository + .insert_receiver_request(ReceiverRequestInsert { + id: &request_id, + transfer_id: request.transfer_id, + remote_endpoint_id: &remote_endpoint_id, + transfer_name: &request.transfer_name, + receiver_name: request.receiver_name.as_deref(), + receiver_device_name: request.receiver_device_name.as_deref(), + app_version: &request.app_version, + }) + .await + .is_err() + || self + .repository + .update_receiver_request_status(&request_id, ReceiverRequestStatus::Accepted, None) + .await + .is_err() + { + return self + .deny(request.transfer_id, remote_endpoint_id, "repository-error") + .await; + } let token = Uuid::new_v4().to_string(); + if self + .repository + .set_receiver_receipt_token(&request_id, &receipt_token_hash(&token)) + .await + .is_err() + { + return self + .deny(request.transfer_id, remote_endpoint_id, "repository-error") + .await; + } let expires_at = now_ms() + APPROVAL_TTL_MS; self.event_hub.emit_transfer( request.transfer_id, @@ -161,7 +230,11 @@ impl ApprovalService { "expires_at": expires_at, }), ); - HandshakeResponse::Approved { token, expires_at } + HandshakeResponse::Approved { + request_id, + token, + expires_at, + } } async fn wait_for_sender_decision( @@ -242,7 +315,21 @@ impl ApprovalService { "expires_at": expires_at, }), ); - HandshakeResponse::Approved { token, expires_at } + if let Err(error) = self + .repository + .set_receiver_receipt_token(&decision.request_id, &receipt_token_hash(&token)) + .await + { + tracing::error!(%error, "failed to attach delivery receipt token"); + return HandshakeResponse::Denied { + reason: "repository-error".to_string(), + }; + } + HandshakeResponse::Approved { + request_id: decision.request_id, + token, + expires_at, + } } Ok(Ok(decision)) => { self.deny( @@ -290,3 +377,7 @@ impl ApprovalService { HandshakeResponse::Denied { reason } } } + +fn receipt_token_hash(token: &str) -> String { + blake3::hash(token.as_bytes()).to_hex().to_string() +} diff --git a/crates/vnidrop/src/handshake.rs b/crates/vnidrop/src/handshake.rs index 2a80d95..385a220 100644 --- a/crates/vnidrop/src/handshake.rs +++ b/crates/vnidrop/src/handshake.rs @@ -24,7 +24,7 @@ impl fmt::Debug for HandshakeService { } impl HandshakeService { - pub(crate) const ALPN: &'static [u8] = b"/vnidrop/handshake/1"; + pub(crate) const ALPN: &'static [u8] = b"/vnidrop/handshake/2"; pub(crate) fn new(approval: crate::approval::ApprovalService) -> Self { Self { approval } @@ -64,6 +64,14 @@ impl ProtocolHandler for HandshakeService { let response = self.handle_request(remote_endpoint_id.clone(), inner).await; let _ = tx.send(response).await; } + HandshakeMessage::ReportDelivery(message) => { + let WithChannels { inner, tx, .. } = message; + let response = self + .approval + .complete_delivery(remote_endpoint_id.clone(), inner) + .await; + let _ = tx.send(response).await; + } } } @@ -94,6 +102,13 @@ impl HandshakeClient { }) .await } + + pub(crate) async fn report_delivery( + &self, + receipt: DeliveryReceipt, + ) -> Result { + self.inner.rpc(receipt).await + } } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -108,8 +123,27 @@ pub(crate) struct RequestTransfer { #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub(crate) enum HandshakeResponse { - Approved { token: String, expires_at: i64 }, - Denied { reason: String }, + Approved { + request_id: String, + token: String, + expires_at: i64, + }, + Denied { + reason: String, + }, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub(crate) struct DeliveryReceipt { + pub(crate) request_id: String, + pub(crate) transfer_id: u64, + pub(crate) token: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub(crate) enum DeliveryReceiptResponse { + Recorded, + Rejected { reason: String }, } #[rpc_requests(message = HandshakeMessage)] @@ -117,4 +151,6 @@ pub(crate) enum HandshakeResponse { enum HandshakeProtocol { #[rpc(tx=oneshot::Sender)] RequestTransfer(RequestTransfer), + #[rpc(tx=oneshot::Sender)] + ReportDelivery(DeliveryReceipt), } diff --git a/crates/vnidrop/src/repository.rs b/crates/vnidrop/src/repository.rs index 01c6743..fe347be 100644 --- a/crates/vnidrop/src/repository.rs +++ b/crates/vnidrop/src/repository.rs @@ -20,7 +20,7 @@ use crate::{ util::now_ms, }; -const SCHEMA_VERSION: i64 = 3; +const SCHEMA_VERSION: i64 = 4; #[derive(Debug, Clone)] pub(crate) struct Repository { @@ -202,12 +202,34 @@ impl Repository { reason TEXT, requested_at INTEGER NOT NULL, responded_at INTEGER + ,receipt_token_hash TEXT + ,completed_at INTEGER ); "#, ) .execute(&self.pool) .await?; + let receiver_columns = sqlx::query("PRAGMA table_info(receiver_requests)") + .fetch_all(&self.pool) + .await?; + if !receiver_columns + .iter() + .any(|row| row.get::(1) == "receipt_token_hash") + { + sqlx::query("ALTER TABLE receiver_requests ADD COLUMN receipt_token_hash TEXT") + .execute(&self.pool) + .await?; + } + if !receiver_columns + .iter() + .any(|row| row.get::(1) == "completed_at") + { + sqlx::query("ALTER TABLE receiver_requests ADD COLUMN completed_at INTEGER") + .execute(&self.pool) + .await?; + } + sqlx::query( "CREATE INDEX IF NOT EXISTS idx_receiver_requests_transfer_id ON receiver_requests(transfer_id, requested_at DESC);", ) @@ -583,6 +605,70 @@ impl Repository { Ok(()) } + pub(crate) async fn set_receiver_receipt_token( + &self, + id: &str, + token_hash: &str, + ) -> Result<()> { + let result = sqlx::query( + "UPDATE receiver_requests SET receipt_token_hash = ?1 WHERE id = ?2 AND status = 'accepted'", + ) + .bind(token_hash) + .bind(id) + .execute(&self.pool) + .await?; + require_one_changed(result.rows_affected(), "attach receiver receipt token") + } + + pub(crate) async fn complete_receiver_delivery( + &self, + id: &str, + transfer_id: u64, + remote_endpoint_id: &str, + token_hash: &str, + ) -> Result<()> { + let result = sqlx::query( + r#" + UPDATE receiver_requests + SET status = 'completed', completed_at = ?1 + WHERE id = ?2 AND transfer_id = ?3 AND remote_endpoint_id = ?4 AND receipt_token_hash = ?5 + AND status = 'accepted' + "#, + ) + .bind(now_ms()) + .bind(id) + .bind(to_db_id(transfer_id)?) + .bind(remote_endpoint_id) + .bind(token_hash) + .execute(&self.pool) + .await?; + if result.rows_affected() == 1 { + return Ok(()); + } + let already_recorded = sqlx::query( + r#" + SELECT EXISTS( + SELECT 1 FROM receiver_requests + WHERE id = ?1 AND transfer_id = ?2 AND remote_endpoint_id = ?3 + AND receipt_token_hash = ?4 AND status = 'completed' + ) + "#, + ) + .bind(id) + .bind(to_db_id(transfer_id)?) + .bind(remote_endpoint_id) + .bind(token_hash) + .fetch_one(&self.pool) + .await? + .get::(0) + != 0; + if already_recorded { + Ok(()) + } else { + anyhow::bail!("delivery receipt did not match an accepted receiver request") + } + } + pub(crate) async fn expire_pending_receiver_requests(&self, reason: &str) -> Result { let result = sqlx::query( r#" @@ -606,7 +692,7 @@ impl Repository { r#" SELECT id, transfer_id, remote_endpoint_id, transfer_name, receiver_name, receiver_device_name, app_version, status, - reason, requested_at, responded_at + reason, requested_at, responded_at, completed_at FROM receiver_requests WHERE transfer_id = ?1 ORDER BY requested_at DESC @@ -652,6 +738,27 @@ impl Repository { rows.into_iter().map(row_to_transfer).collect() } + pub(crate) async fn delete_transfer(&self, transfer_id: u64) -> Result<()> { + self.maybe_fail_write()?; + let transfer_id = to_db_id(transfer_id)?; + let mut transaction = self.pool.begin().await?; + sqlx::query("DELETE FROM receiver_requests WHERE transfer_id = ?1") + .bind(transfer_id) + .execute(&mut *transaction) + .await?; + sqlx::query("DELETE FROM transfer_events WHERE transfer_id = ?1") + .bind(transfer_id) + .execute(&mut *transaction) + .await?; + let deleted = sqlx::query("DELETE FROM transfers WHERE transfer_id = ?1") + .bind(transfer_id) + .execute(&mut *transaction) + .await?; + require_one_changed(deleted.rows_affected(), "delete transfer")?; + transaction.commit().await?; + Ok(()) + } + pub(crate) async fn list_events( &self, transfer_id: Option, @@ -754,5 +861,6 @@ fn row_to_receiver_request(row: sqlx::sqlite::SqliteRow) -> ReceiverRequest { reason: row.get("reason"), requested_at: row.get("requested_at"), responded_at: row.get("responded_at"), + completed_at: row.get("completed_at"), } } diff --git a/crates/vnidrop/src/runtime.rs b/crates/vnidrop/src/runtime.rs index b748121..311126f 100644 --- a/crates/vnidrop/src/runtime.rs +++ b/crates/vnidrop/src/runtime.rs @@ -44,12 +44,12 @@ use crate::{ read_stream_from_blocking_reader, validated_relative_string, wait_for_writer, write_stream_to_blocking_writer, AtomicOutputFile, TransferImport, }, - handshake::{HandshakeResponse, HandshakeService}, + handshake::{DeliveryReceipt, DeliveryReceiptResponse, HandshakeResponse, HandshakeService}, logging::init_logging, repository::{Repository, TransferUpsert}, secret::load_or_create_secret, ticket::{parse_transfer_ticket_with_limits, ParsedTransferTicket, VnidropTicket}, - transfer_state::{TransferDirection, TransferStatus}, + transfer_state::{ReceiverRequestStatus, TransferDirection, TransferStatus}, util::{non_empty, unique_transfer_id}, }; @@ -244,6 +244,12 @@ impl VnidropCore { .map_err(VnidropError::transfer) } + pub fn delete_transfer(&self, transfer_id: u64) -> Result<(), VnidropError> { + self.runtime + .block_on(self.inner.delete_transfer(transfer_id)) + .map_err(VnidropError::transfer) + } + pub fn set_transfer_access_mode( &self, transfer_id: u64, @@ -725,20 +731,25 @@ impl CoreInner { if let ReceiveTarget::Directory(output_dir) = &target { tokio::fs::create_dir_all(output_dir).await?; } + let sender_addr = parsed.blob_ticket.addr().clone(); self.emit_transfer(transfer_id, "receive", "network", "connecting", json!({})); - if let Some(metadata) = &parsed.metadata { - self.request_transfer_approval( - transfer_id, - parsed.blob_ticket.addr().clone(), - metadata, - receiver_name.as_deref(), + let delivery_receipt = if let Some(metadata) = &parsed.metadata { + Some( + self.request_transfer_approval( + transfer_id, + sender_addr.clone(), + metadata, + receiver_name.as_deref(), + ) + .await?, ) - .await?; - } + } else { + None + }; let connection = self .endpoint - .connect(parsed.blob_ticket.addr().clone(), iroh_blobs::ALPN) + .connect(sender_addr.clone(), iroh_blobs::ALPN) .await?; self.emit_transfer(transfer_id, "receive", "network", "connected", json!({})); @@ -801,6 +812,33 @@ impl CoreInner { ) .await?; self.emit_transfer(transfer_id, "receive", "lifecycle", "done", json!({})); + if let Some(receipt) = delivery_receipt { + let sender_transfer_id = receipt.transfer_id; + let client = HandshakeService::client(self.endpoint.clone(), sender_addr); + match client.report_delivery(receipt).await { + Ok(DeliveryReceiptResponse::Recorded) => self.emit_transfer( + transfer_id, + "receive", + "delivery", + "receipt-recorded", + json!({ "sender_transfer_id": sender_transfer_id }), + ), + Ok(DeliveryReceiptResponse::Rejected { reason }) => self.emit_transfer( + transfer_id, + "receive", + "delivery", + "receipt-rejected", + json!({ "reason": reason }), + ), + Err(error) => self.emit_transfer( + transfer_id, + "receive", + "delivery", + "receipt-failed", + json!({ "reason": error.to_string() }), + ), + } + } Ok(()) } @@ -905,6 +943,58 @@ impl CoreInner { anyhow::bail!("transfer not found") } + async fn delete_transfer(&self, transfer_id: u64) -> Result<()> { + if self + .active_transfers + .lock() + .await + .contains_key(&transfer_id) + { + anyhow::bail!("an active transfer must finish or be cancelled before deletion"); + } + + let transfer = self + .repository + .list_transfers() + .await? + .into_iter() + .find(|transfer| transfer.transfer_id == transfer_id) + .ok_or_else(|| anyhow::anyhow!("transfer not found"))?; + + // Revoke a live share durably before removing its history. If the + // subsequent delete fails, a restart must never expose it again. + if transfer.status == TransferStatus::Sharing.as_str() { + self.repository + .transition_transfer_status( + transfer_id, + TransferStatus::Sharing, + TransferStatus::Stopped, + ) + .await?; + } + + for request in self.repository.list_receiver_requests(transfer_id).await? { + if request.status == ReceiverRequestStatus::Requested.as_str() { + let _ = self + .approval + .respond( + request.id, + false, + Some("transfer deleted by sender".to_string()), + ) + .await; + } + } + + self.active_shares.lock().await.remove(&transfer_id); + self.hash_to_transfer + .lock() + .await + .retain(|_, id| *id != transfer_id); + self.access_policy.remove_transfer(transfer_id).await; + self.repository.delete_transfer(transfer_id).await + } + async fn set_transfer_access_mode( &self, transfer_id: u64, @@ -948,7 +1038,7 @@ impl CoreInner { addr: iroh::EndpointAddr, metadata: &TransferMetadata, receiver_name: Option<&str>, - ) -> Result<()> { + ) -> Result { self.emit_transfer( local_transfer_id, "receive", @@ -966,7 +1056,11 @@ impl CoreInner { .await .map_err(|error| anyhow::anyhow!("handshake request failed: {error}"))? { - HandshakeResponse::Approved { expires_at, .. } => { + HandshakeResponse::Approved { + request_id, + token, + expires_at, + } => { self.emit_transfer( local_transfer_id, "receive", @@ -977,7 +1071,11 @@ impl CoreInner { "expires_at": expires_at, }), ); - Ok(()) + Ok(DeliveryReceipt { + request_id, + transfer_id: metadata.transfer_id, + token, + }) } HandshakeResponse::Denied { reason } => { anyhow::bail!("transfer request was denied by sender: {reason}") diff --git a/crates/vnidrop/src/tests/repository.rs b/crates/vnidrop/src/tests/repository.rs index 6d6b052..1272591 100644 --- a/crates/vnidrop/src/tests/repository.rs +++ b/crates/vnidrop/src/tests/repository.rs @@ -27,7 +27,7 @@ fn transfer( async fn persists_transfers_and_events_across_reopen() { let temp = tempfile::tempdir().unwrap(); let repository = Repository::open(temp.path()).await.unwrap(); - assert_eq!(repository.schema_version().await.unwrap(), 3); + assert_eq!(repository.schema_version().await.unwrap(), 4); repository .insert_transfer(transfer( 7, @@ -96,6 +96,22 @@ async fn receiver_request_can_only_be_resolved_once() { .update_receiver_request_status("request-1", ReceiverRequestStatus::Accepted, None) .await .unwrap(); + repository + .set_receiver_receipt_token("request-1", "token-hash") + .await + .unwrap(); + repository + .complete_receiver_delivery("request-1", 77, "node-a", "token-hash") + .await + .unwrap(); + repository + .complete_receiver_delivery("request-1", 77, "node-a", "token-hash") + .await + .unwrap(); + assert!(repository + .complete_receiver_delivery("request-1", 77, "node-b", "token-hash") + .await + .is_err()); assert!(repository .update_receiver_request_status("request-1", ReceiverRequestStatus::Refused, Some("late"),) @@ -108,9 +124,10 @@ async fn receiver_request_can_only_be_resolved_once() { let requests = repository.list_receiver_requests(77).await.unwrap(); assert_eq!(requests.len(), 1); - assert_eq!(requests[0].status, "accepted"); + assert_eq!(requests[0].status, "completed"); assert_eq!(requests[0].receiver_name.as_deref(), Some("receiver")); assert!(requests[0].responded_at.is_some()); + assert!(requests[0].completed_at.is_some()); } #[tokio::test] @@ -474,7 +491,7 @@ async fn migrates_schema_v2_identity_without_losing_transfer() { pool.close().await; let repository = Repository::open(temp.path()).await.unwrap(); - assert_eq!(repository.schema_version().await.unwrap(), 3); + assert_eq!(repository.schema_version().await.unwrap(), 4); let stored = repository.list_transfers().await.unwrap().remove(0); assert_eq!(stored.transfer_id, 7); assert_eq!(stored.local_id, "legacy-7-send"); @@ -511,6 +528,63 @@ async fn event_reads_respect_configured_history_limit() { assert_eq!(events[0].id, "event-2"); assert_eq!(events[1].id, "event-1"); } + +#[tokio::test] +async fn deleting_transfer_removes_related_history_transactionally() { + let temp = tempfile::tempdir().unwrap(); + let repository = Repository::open(temp.path()).await.unwrap(); + repository + .insert_transfer(transfer( + 88, + TransferDirection::Send, + TransferStatus::Stopped, + )) + .await + .unwrap(); + repository + .insert_receiver_request(ReceiverRequestInsert { + id: "request-delete", + transfer_id: 88, + 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-delete".to_string(), + timestamp: 1, + scope: "transfer".to_string(), + transfer_id: Some(88), + direction: Some("send".to_string()), + phase: "test".to_string(), + kind: "created".to_string(), + data_json: "{}".to_string(), + }, + 500, + ) + .await + .unwrap(); + + repository.delete_transfer(88).await.unwrap(); + + assert!(repository.list_transfers().await.unwrap().is_empty()); + assert!(repository + .list_events(Some(88), 500) + .await + .unwrap() + .is_empty()); + assert!(repository + .list_receiver_requests(88) + .await + .unwrap() + .is_empty()); + assert!(repository.delete_transfer(88).await.is_err()); +} use std::str::FromStr; use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions}; diff --git a/crates/vnidrop/src/transfer_state.rs b/crates/vnidrop/src/transfer_state.rs index e25f211..7952671 100644 --- a/crates/vnidrop/src/transfer_state.rs +++ b/crates/vnidrop/src/transfer_state.rs @@ -86,6 +86,7 @@ pub(crate) enum ReceiverRequestStatus { Accepted, Refused, Expired, + Completed, } impl ReceiverRequestStatus { @@ -95,6 +96,7 @@ impl ReceiverRequestStatus { Self::Accepted => "accepted", Self::Refused => "refused", Self::Expired => "expired", + Self::Completed => "completed", } } } @@ -108,6 +110,7 @@ impl TryFrom<&str> for ReceiverRequestStatus { "accepted" => Ok(Self::Accepted), "refused" => Ok(Self::Refused), "expired" => Ok(Self::Expired), + "completed" => Ok(Self::Completed), _ => bail!("unknown receiver request status: {value}"), } } diff --git a/crates/vnidrop/tests/approval.rs b/crates/vnidrop/tests/approval.rs index 2de629f..0800f68 100644 --- a/crates/vnidrop/tests/approval.rs +++ b/crates/vnidrop/tests/approval.rs @@ -47,11 +47,14 @@ fn public_share_receives_without_sender_approval() { std::fs::read(output_dir.path().join("public.txt")).unwrap(), b"public content" ); - assert!(sender + let deliveries = sender .core .list_receiver_requests(share.transfer_id) - .unwrap() - .is_empty()); + .unwrap(); + assert_eq!(deliveries.len(), 1); + assert_eq!(deliveries[0].receiver_name.as_deref(), Some("Receiver")); + assert_eq!(deliveries[0].status, "completed"); + assert!(deliveries[0].completed_at.is_some()); } #[test] @@ -93,6 +96,13 @@ fn approval_required_denies_then_allows_receiver() { std::fs::read(allowed_output.path().join("private.txt")).unwrap(), b"approved content" ); + let completed = sender + .core + .list_receiver_requests(share.transfer_id) + .unwrap(); + assert!(completed + .iter() + .any(|request| request.status == "completed")); } #[test] diff --git a/crates/vnidrop/tests/lifecycle.rs b/crates/vnidrop/tests/lifecycle.rs index 562a9fd..da4b086 100644 --- a/crates/vnidrop/tests/lifecycle.rs +++ b/crates/vnidrop/tests/lifecycle.rs @@ -54,6 +54,26 @@ fn cancelling_share_updates_status_and_events() { .any(|event| event.kind == "share-stopped")); } +#[test] +fn deleting_share_revokes_it_and_removes_persisted_history() { + let source_dir = tempfile::tempdir().unwrap(); + let core_dir = tempfile::tempdir().unwrap(); + let source_path = source_dir.path().join("delete.txt"); + std::fs::write(&source_path, b"delete me").unwrap(); + + let sender = CoreGuard::start(core_dir.path(), Arc::new(RecordingSink::default())); + let share = share_path(&sender, &source_path, 101, "delete.txt", false); + sender.delete_transfer(share.transfer_id).unwrap(); + + assert!(sender.list_transfers().unwrap().is_empty()); + assert_eq!(sender.status().active_shares, 0); + drop(sender); + + let restarted = CoreGuard::start(core_dir.path(), Arc::new(RecordingSink::default())); + assert!(restarted.list_transfers().unwrap().is_empty()); + assert_eq!(restarted.status().active_shares, 0); +} + #[test] fn persisted_share_is_recovered_and_can_be_stopped_after_restart() { let source_dir = tempfile::tempdir().unwrap();