diff --git a/crates/vnidrop/src/api.rs b/crates/vnidrop/src/api.rs index 8020a9a..68bdcb4 100644 --- a/crates/vnidrop/src/api.rs +++ b/crates/vnidrop/src/api.rs @@ -23,7 +23,7 @@ pub fn experimental_saved_device_capabilities() -> ExperimentalSavedDeviceCapabi ExperimentalSavedDeviceCapabilities { domain_contract_version: 1, relationship_protocol_version: 1, - targeted_transfer_protocol_version: 1, + targeted_transfer_protocol_version: 2, } } diff --git a/crates/vnidrop/src/runtime/facade.rs b/crates/vnidrop/src/runtime/facade.rs index abbd733..fc48025 100644 --- a/crates/vnidrop/src/runtime/facade.rs +++ b/crates/vnidrop/src/runtime/facade.rs @@ -124,6 +124,35 @@ impl VnidropCore { }) } + pub(crate) fn suppress_targeted_completion_for_test(&self, suppress: bool) { + self.inner + .suppress_targeted_completion + .store(suppress, std::sync::atomic::Ordering::SeqCst); + } + + pub(crate) fn targeted_payload_is_registered_for_test( + &self, + id: String, + ) -> Result { + self.block_on(async { + let row = self + .inner + .targeted_store() + .get_row(&id) + .await? + .ok_or_else(|| VnidropError::invalid_input(anyhow::anyhow!("unknown transfer")))?; + let hash = row + .content_hash + .parse::() + .map_err(|error| VnidropError::invalid_input(anyhow::anyhow!(error)))?; + Ok(self + .inner + .transfer_ids_for_hash(hash) + .await + .contains(&row.protocol_transfer_id)) + }) + } + pub(crate) fn initialize_with_test_secret_store_limits_and_network( app_data_dir: String, event_sink: Arc, diff --git a/crates/vnidrop/src/runtime/lifecycle.rs b/crates/vnidrop/src/runtime/lifecycle.rs index ca83922..e1ea33a 100644 --- a/crates/vnidrop/src/runtime/lifecycle.rs +++ b/crates/vnidrop/src/runtime/lifecycle.rs @@ -191,6 +191,10 @@ impl CoreInner { task.abort(); let _ = task.await; } + if let Some(task) = self.targeted_completion_task.lock().await.take() { + task.abort(); + let _ = task.await; + } if let Err(error) = self.router.shutdown().await { self.emit_endpoint( "shutdown", diff --git a/crates/vnidrop/src/runtime/mod.rs b/crates/vnidrop/src/runtime/mod.rs index fcc9914..f3a6e1b 100644 --- a/crates/vnidrop/src/runtime/mod.rs +++ b/crates/vnidrop/src/runtime/mod.rs @@ -122,10 +122,13 @@ pub(super) struct CoreInner { pub(super) provider_task: TokioMutex>>, pub(super) delivery_receipt_notify: Notify, pub(super) delivery_receipt_task: TokioMutex>>, + pub(super) targeted_completion_task: TokioMutex>>, pub(super) shutdown_started: AtomicBool, /// Test-only log of peers passed to [`Self::cancel_targeted_transfers_for_peer`]. #[cfg(test)] targeted_cancel_log: std::sync::Mutex>, + #[cfg(test)] + suppress_targeted_completion: AtomicBool, } pub(super) struct ActiveTransfer { @@ -445,9 +448,12 @@ impl CoreInner { provider_task: TokioMutex::new(None), delivery_receipt_notify: Notify::new(), delivery_receipt_task: TokioMutex::new(None), + targeted_completion_task: TokioMutex::new(None), shutdown_started: AtomicBool::new(false), #[cfg(test)] targeted_cancel_log: std::sync::Mutex::new(Vec::new()), + #[cfg(test)] + suppress_targeted_completion: AtomicBool::new(false), }); // In-flight connecting/transferring transfers become Interrupted across restart. @@ -470,6 +476,7 @@ impl CoreInner { ); inner.spawn_provider_event_task(event_rx).await; inner.spawn_delivery_receipt_task().await; + inner.spawn_targeted_completion_task().await; if let Err(error) = inner.pairing_eligibility.reconcile().await { tracing::warn!(%error, "failed to reconcile pairing eligibility"); } diff --git a/crates/vnidrop/src/runtime/receive.rs b/crates/vnidrop/src/runtime/receive.rs index d492154..4d45b59 100644 --- a/crates/vnidrop/src/runtime/receive.rs +++ b/crates/vnidrop/src/runtime/receive.rs @@ -213,8 +213,9 @@ impl CoreInner { }, ); let result = tokio::select! { - result = self.download_targeted_payload(transfer_id, blob_ticket, target) => result, + biased; _ = &mut cancelled => Err(VnidropError::cancelled("transfer cancelled").into()), + result = self.download_targeted_payload(transfer_id, blob_ticket, target) => result, }; self.active_transfers .lock() diff --git a/crates/vnidrop/src/runtime/targeted.rs b/crates/vnidrop/src/runtime/targeted.rs index dff8154..0ce3471 100644 --- a/crates/vnidrop/src/runtime/targeted.rs +++ b/crates/vnidrop/src/runtime/targeted.rs @@ -17,8 +17,9 @@ use crate::{ targeted_transfer::{ auth_secret_material, protocol::{ - map_offer_refuse_reason, CancelTargetedOffer, DeliverTargetedAuthorization, - SubmitTargetedOffer, TargetedTransferProtocol, WireOfferResponse, + map_offer_refuse_reason, CancelTargetedOffer, CompleteTargetedTransfer, + CompletionResponse, DeliverTargetedAuthorization, SubmitTargetedOffer, + TargetedTransferProtocol, WireOfferResponse, }, reconstruct_authorization, TargetedAuthorization, TargetedAuthorizationDraft, TargetedTransferRole, TargetedTransferRow, @@ -35,6 +36,71 @@ impl CoreInner { std::time::Duration::from_millis(self.limits.offer_timeout_ms) } + pub(super) async fn spawn_targeted_completion_task(self: &Arc) { + let core = self.clone(); + let task = tokio::spawn(async move { + let mut interval = tokio::time::interval(std::time::Duration::from_secs(2)); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + loop { + interval.tick().await; + core.release_completed_targeted_payloads().await; + core.retry_targeted_completions().await; + } + }); + *self.targeted_completion_task.lock().await = Some(task); + } + + async fn retry_targeted_completions(&self) { + let Ok(rows) = self.targeted_store().list_pending_completions().await else { + return; + }; + for row in rows.into_iter().take(1) { + let Ok(Some(encoded)) = self.load_stored_authorization(&row).await else { + let _ = self + .targeted_store() + .defer_pending_completion(&row.id, now_ms() + 30_000) + .await; + continue; + }; + let Ok(auth) = TargetedAuthorization::decode(&encoded) else { + let _ = self + .targeted_store() + .defer_pending_completion(&row.id, now_ms() + 30_000) + .await; + continue; + }; + if self.acknowledge_targeted_completion(&auth).await.is_ok() { + let _ = self + .targeted_store() + .clear_pending_completion(&row.id) + .await; + } else { + let _ = self + .targeted_store() + .defer_pending_completion(&row.id, now_ms() + 5_000) + .await; + } + } + } + + async fn release_completed_targeted_payloads(&self) { + let Ok(rows) = self.targeted_store().list_completed_sender_rows().await else { + return; + }; + for row in rows { + if self + .try_teardown_targeted_payload(row.protocol_transfer_id, Some(&row.id)) + .await + .is_ok() + { + let _ = self + .targeted_store() + .clear_pending_payload_release(&row.id) + .await; + } + } + } + pub(super) fn targeted_store(&self) -> crate::targeted_transfer::TargetedTransferStore { self.targeted_transfers.clone() } @@ -628,6 +694,19 @@ impl CoreInner { } }; + if let Err(error) = store + .set_state( + &transfer_uuid, + TargetedTransferState::AwaitingApproval, + TargetedTransferState::Approved, + ) + .await + { + self.teardown_targeted_payload(protocol_transfer_id, Some(&transfer_uuid)) + .await; + return Err(error); + } + let deliver = match client .deliver_authorization(DeliverTargetedAuthorization { transfer_id: transfer_uuid.clone(), @@ -649,11 +728,7 @@ impl CoreInner { }; if deliver != crate::targeted_transfer::protocol::DeliverAuthorizationResponse::Stored { let _ = store - .set_state( - &transfer_uuid, - TargetedTransferState::AwaitingApproval, - TargetedTransferState::Failed, - ) + .set_state_from_any(&transfer_uuid, TargetedTransferState::Failed) .await; self.teardown_targeted_payload(protocol_transfer_id, Some(&transfer_uuid)) .await; @@ -662,14 +737,6 @@ impl CoreInner { ))); } - store - .set_state( - &transfer_uuid, - TargetedTransferState::AwaitingApproval, - TargetedTransferState::Approved, - ) - .await?; - store .get(&transfer_uuid) .await? @@ -677,15 +744,31 @@ impl CoreInner { } async fn teardown_targeted_payload(&self, protocol_transfer_id: u64, id: Option<&str>) { + if let Err(error) = self + .try_teardown_targeted_payload(protocol_transfer_id, id) + .await + { + tracing::warn!(%error, "failed to release targeted payload"); + } + } + + async fn try_teardown_targeted_payload( + &self, + protocol_transfer_id: u64, + id: Option<&str>, + ) -> Result<(), VnidropError> { self.unregister_transfer_hashes(protocol_transfer_id).await; self.access_policy .remove_transfer(protocol_transfer_id) .await; if let Some(id) = id { - if let Err(error) = self.store.tags().delete(targeted_tag_name(id)).await { - tracing::warn!(%error, transfer_id = id, "failed to release targeted payload tag"); - } + self.store + .tags() + .delete(targeted_tag_name(id)) + .await + .map_err(VnidropError::transfer)?; } + Ok(()) } pub(super) async fn receive_targeted_transfer( @@ -693,11 +776,10 @@ impl CoreInner { transfer_id: String, output_dir: String, ) -> Result<(), VnidropError> { - self.receive_targeted_to_target( - transfer_id, - ReceiveTarget::Directory(std::path::PathBuf::from(output_dir)), - ) - .await + let output_dir = + crate::filesystem::platform_path(&output_dir).map_err(VnidropError::filesystem)?; + self.receive_targeted_to_target(transfer_id, ReceiveTarget::Directory(output_dir)) + .await } pub(super) async fn receive_targeted_transfer_with_output_sink( @@ -723,11 +805,10 @@ impl CoreInner { id: String, output_dir: String, ) -> Result<(), VnidropError> { - self.resume_targeted_to_target( - id, - ReceiveTarget::Directory(std::path::PathBuf::from(output_dir)), - ) - .await + let output_dir = + crate::filesystem::platform_path(&output_dir).map_err(VnidropError::filesystem)?; + self.resume_targeted_to_target(id, ReceiveTarget::Directory(output_dir)) + .await } pub(super) async fn resume_targeted_transfer_with_output_sink( @@ -826,16 +907,11 @@ impl CoreInner { ) .await?; } - TargetedTransferState::Connecting => { - store - .set_state( - &auth.transfer_id, - TargetedTransferState::Connecting, - TargetedTransferState::Transferring, - ) - .await?; + TargetedTransferState::Connecting | TargetedTransferState::Transferring => { + return Err(VnidropError::InvalidTransition { + reason: "targeted receive is already active".to_string(), + }); } - TargetedTransferState::Transferring => {} other => { return Err(VnidropError::InvalidTransition { reason: format!( @@ -855,17 +931,14 @@ impl CoreInner { match receive_result { Ok(()) => { - if let Ok(Some(row)) = store.get_row(&auth.transfer_id).await { - let _ = store - .set_verified_bytes(&auth.transfer_id, row.total_size) - .await; - let _ = store - .set_state( - &auth.transfer_id, - TargetedTransferState::Transferring, - TargetedTransferState::Completed, - ) - .await; + let row = store.get_row(&auth.transfer_id).await?.ok_or_else(|| { + VnidropError::internal(anyhow::anyhow!("targeted transfer missing")) + })?; + store + .complete_receiver_and_enqueue(&auth.transfer_id, row.total_size) + .await?; + if self.acknowledge_targeted_completion(auth).await.is_ok() { + store.clear_pending_completion(&auth.transfer_id).await?; } Ok(()) } @@ -888,6 +961,43 @@ impl CoreInner { } } + async fn acknowledge_targeted_completion( + &self, + auth: &TargetedAuthorization, + ) -> Result<(), VnidropError> { + #[cfg(test)] + if self + .suppress_targeted_completion + .load(std::sync::atomic::Ordering::SeqCst) + { + return Err(VnidropError::device_unavailable(anyhow::anyhow!( + "completion delivery suppressed by test" + ))); + } + let addr = self + .device_relationships + .peer_addr(&auth.sender_endpoint_id) + .await?; + let client = TargetedTransferProtocol::client(self.endpoint.clone(), addr); + let response = tokio::time::timeout( + self.connection_timeout(), + client.complete_transfer(CompleteTargetedTransfer { + transfer_id: auth.transfer_id.clone(), + verified_bytes: auth.total_size, + authorization: auth.encode()?, + }), + ) + .await + .map_err(|_| VnidropError::device_unavailable(anyhow::anyhow!("completion timed out")))? + .map_err(|error| VnidropError::network(anyhow::anyhow!(error)))?; + if response != CompletionResponse::Recorded { + return Err(VnidropError::permission(anyhow::anyhow!( + "sender rejected targeted completion" + ))); + } + Ok(()) + } + async fn persist_authorization_secret( &self, transfer_id: &str, diff --git a/crates/vnidrop/src/targeted_transfer/inbox.rs b/crates/vnidrop/src/targeted_transfer/inbox.rs index 232e88d..21f3906 100644 --- a/crates/vnidrop/src/targeted_transfer/inbox.rs +++ b/crates/vnidrop/src/targeted_transfer/inbox.rs @@ -218,6 +218,27 @@ impl TargetedOfferInbox { .map(|entry| entry.offer.clone()) } + pub(crate) async fn authorization_matches_pending( + &self, + auth: &crate::targeted_transfer::TargetedAuthorization, + ) -> bool { + self.pending + .lock() + .await + .get(&auth.transfer_id) + .is_some_and(|entry| { + let offer = &entry.offer; + offer.sender_endpoint_id == auth.sender_endpoint_id + && offer.receiver_endpoint_id == auth.receiver_endpoint_id + && offer.manifest_id == auth.manifest_id + && offer.content_hash == auth.content_hash + && offer.transfer_name == auth.transfer_name + && offer.file_count == auth.file_count + && offer.total_size == auth.total_size + && offer.protocol_version == auth.protocol_version + }) + } + pub(crate) async fn settled_authorization(&self, transfer_id: &str) -> Option { match self.settled.lock().await.get(transfer_id) { Some(SettledOfferResult::Accepted { @@ -370,6 +391,7 @@ fn offers_equivalent(left: &PendingTargetedOffer, right: &PendingTargetedOffer) && left.receiver_endpoint_id == right.receiver_endpoint_id && left.manifest_id == right.manifest_id && left.content_hash == right.content_hash + && left.transfer_name == right.transfer_name && left.file_count == right.file_count && left.total_size == right.total_size && left.protocol_version == right.protocol_version diff --git a/crates/vnidrop/src/targeted_transfer/protocol.rs b/crates/vnidrop/src/targeted_transfer/protocol.rs index adeeb9b..6e7e72a 100644 --- a/crates/vnidrop/src/targeted_transfer/protocol.rs +++ b/crates/vnidrop/src/targeted_transfer/protocol.rs @@ -18,7 +18,7 @@ use serde::{Deserialize, Serialize}; use super::{ auth::TargetedAuthorization, inbox::{TargetedOfferDecision, TargetedOfferInbox}, - state_as_str, TargetedTransferStore, + state_as_str, TargetedTransferRole, TargetedTransferStore, }; use crate::{ api::{ @@ -50,7 +50,7 @@ impl fmt::Debug for TargetedTransferProtocol { } impl TargetedTransferProtocol { - pub(crate) const ALPN: &'static [u8] = b"/vnidrop/targeted-transfer/1"; + pub(crate) const ALPN: &'static [u8] = b"/vnidrop/targeted-transfer/2"; pub(crate) fn new( relationships: std::sync::Arc, @@ -248,18 +248,45 @@ impl TargetedTransferProtocol { } if let Ok(Some(row)) = self.store.get_row(&delivery.transfer_id).await { if row.authorization_secret_handle.is_some() + && row.protocol_transfer_id == auth.protocol_transfer_id + && row.sender_endpoint_id == auth.sender_endpoint_id + && row.receiver_endpoint_id == auth.receiver_endpoint_id && row.manifest_id == auth.manifest_id && row.content_hash == auth.content_hash + && row.transfer_name == auth.transfer_name + && row.file_count == auth.file_count + && row.total_size == auth.total_size + && row.blob_ticket.as_deref() == Some(auth.blob_ticket.as_str()) + && auth.protocol_version + == experimental_saved_device_capabilities().targeted_transfer_protocol_version { return DeliverAuthorizationResponse::Stored; } + if row.authorization_secret_handle.is_some() { + return DeliverAuthorizationResponse::Rejected; + } + } + if !self.inbox.authorization_matches_pending(&auth).await { + return DeliverAuthorizationResponse::Rejected; } if self .inbox .deliver_authorization(&delivery.transfer_id, delivery.authorization) .await { - DeliverAuthorizationResponse::Stored + let deadline = tokio::time::Instant::now() + + std::time::Duration::from_millis(self.limits.offer_timeout_ms); + loop { + if let Ok(Some(row)) = self.store.get_row(&delivery.transfer_id).await { + if row.authorization_secret_handle.is_some() { + return DeliverAuthorizationResponse::Stored; + } + } + if tokio::time::Instant::now() >= deadline { + return DeliverAuthorizationResponse::Rejected; + } + tokio::time::sleep(std::time::Duration::from_millis(25)).await; + } } else { DeliverAuthorizationResponse::Rejected } @@ -285,6 +312,65 @@ impl TargetedTransferProtocol { } CancelWireOfferResponse::Cancelled } + + async fn handle_completion( + &self, + remote_endpoint_id: &str, + completion: CompleteTargetedTransfer, + ) -> CompletionResponse { + if self + .relationships + .require_saved(remote_endpoint_id) + .await + .is_err() + { + return CompletionResponse::Rejected; + } + let Ok(Some(row)) = self.store.get_row(&completion.transfer_id).await else { + return CompletionResponse::Rejected; + }; + let Ok(auth) = TargetedAuthorization::decode(&completion.authorization) else { + return CompletionResponse::Rejected; + }; + if row.role != TargetedTransferRole::Sender + || row.receiver_endpoint_id != remote_endpoint_id + || auth.receiver_endpoint_id != remote_endpoint_id + || auth.sender_endpoint_id != self.local_endpoint_id + || auth.transfer_id != row.id + || auth.protocol_transfer_id != row.protocol_transfer_id + || auth.manifest_id != row.manifest_id + || auth.content_hash != row.content_hash + || auth.transfer_name != row.transfer_name + || auth.file_count != row.file_count + || auth.total_size != row.total_size + || auth.total_size != completion.verified_bytes + || row.blob_ticket.as_deref() != Some(auth.blob_ticket.as_str()) + || auth.protocol_version + != experimental_saved_device_capabilities().targeted_transfer_protocol_version + { + return CompletionResponse::Rejected; + } + if row.state == TargetedTransferState::Completed { + return CompletionResponse::Recorded; + } + if matches!( + row.state, + TargetedTransferState::Cancelled + | TargetedTransferState::Declined + | TargetedTransferState::Failed + | TargetedTransferState::Deleted + ) { + return CompletionResponse::Rejected; + } + match self + .store + .mark_sender_completed(&completion.transfer_id) + .await + { + Ok(()) => CompletionResponse::Recorded, + Err(_) => CompletionResponse::Rejected, + } + } } impl ProtocolHandler for TargetedTransferProtocol { @@ -320,6 +406,11 @@ impl ProtocolHandler for TargetedTransferProtocol { let response = self.handle_cancel(&remote_endpoint_id, inner).await; let _ = tx.send(response).await; } + TargetedTransferMessage::CompleteTargetedTransfer(message) => { + let WithChannels { inner, tx, .. } = message; + let response = self.handle_completion(&remote_endpoint_id, inner).await; + let _ = tx.send(response).await; + } } } connection.closed().await; @@ -357,6 +448,13 @@ impl TargetedTransferClient { ) -> Result { self.inner.rpc(cancel).await } + + pub(crate) async fn complete_transfer( + &self, + completion: CompleteTargetedTransfer, + ) -> Result { + self.inner.rpc(completion).await + } } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -415,6 +513,19 @@ pub(crate) enum CancelWireOfferResponse { Rejected, } +#[derive(Debug, Clone, Serialize, Deserialize)] +pub(crate) struct CompleteTargetedTransfer { + pub(crate) transfer_id: String, + pub(crate) verified_bytes: u64, + pub(crate) authorization: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub(crate) enum CompletionResponse { + Recorded, + Rejected, +} + #[rpc_requests(message = TargetedTransferMessage)] #[derive(Debug, Serialize, Deserialize)] #[allow( @@ -430,6 +541,8 @@ enum TargetedTransferMessages { DeliverTargetedAuthorization(DeliverTargetedAuthorization), #[rpc(tx = oneshot::Sender)] CancelTargetedOffer(CancelTargetedOffer), + #[rpc(tx = oneshot::Sender)] + CompleteTargetedTransfer(CompleteTargetedTransfer), } fn parse_offer_relay_urls(values: &[String]) -> Result, ()> { diff --git a/crates/vnidrop/src/targeted_transfer/store.rs b/crates/vnidrop/src/targeted_transfer/store.rs index 2eb74ee..b455800 100644 --- a/crates/vnidrop/src/targeted_transfer/store.rs +++ b/crates/vnidrop/src/targeted_transfer/store.rs @@ -36,6 +36,42 @@ pub(crate) async fn ensure_schema(pool: &SqlitePool) -> anyhow::Result<()> { ) .execute(pool) .await?; + sqlx::query( + r#" + CREATE TABLE IF NOT EXISTS targeted_completion_outbox ( + transfer_id TEXT PRIMARY KEY, + created_at INTEGER NOT NULL, + next_attempt_at INTEGER NOT NULL, + FOREIGN KEY(transfer_id) REFERENCES targeted_transfers(id) ON DELETE CASCADE + ); + "#, + ) + .execute(pool) + .await?; + let completion_columns = sqlx::query("PRAGMA table_info(targeted_completion_outbox)") + .fetch_all(pool) + .await?; + if !completion_columns + .iter() + .any(|row| row.get::(1) == "next_attempt_at") + { + sqlx::query( + "ALTER TABLE targeted_completion_outbox ADD COLUMN next_attempt_at INTEGER NOT NULL DEFAULT 0", + ) + .execute(pool) + .await?; + } + sqlx::query( + r#" + CREATE TABLE IF NOT EXISTS targeted_payload_release_outbox ( + transfer_id TEXT PRIMARY KEY, + created_at INTEGER NOT NULL, + FOREIGN KEY(transfer_id) REFERENCES targeted_transfers(id) ON DELETE CASCADE + ); + "#, + ) + .execute(pool) + .await?; let columns = sqlx::query("PRAGMA table_info(targeted_transfers)") .fetch_all(pool) .await?; @@ -190,27 +226,156 @@ impl TargetedTransferStore { Ok(()) } - pub(crate) async fn set_verified_bytes( + pub(crate) async fn mark_sender_completed(&self, id: &str) -> Result<(), VnidropError> { + let mut transaction = self.pool.begin().await.map_err(VnidropError::repository)?; + let result = sqlx::query( + r#" + UPDATE targeted_transfers + SET state = 'completed', verified_bytes = total_size, updated_at = ?2 + WHERE id = ?1 AND role = 'sender' + AND state IN ('approved', 'connecting', 'transferring', 'interrupted') + "#, + ) + .bind(id) + .bind(now_ms()) + .execute(&mut *transaction) + .await + .map_err(VnidropError::repository)?; + if result.rows_affected() == 0 { + return Err(VnidropError::InvalidTransition { + reason: "sender transfer cannot be completed".to_string(), + }); + } + sqlx::query( + "INSERT OR IGNORE INTO targeted_payload_release_outbox (transfer_id, created_at) VALUES (?1, ?2)", + ) + .bind(id) + .bind(now_ms()) + .execute(&mut *transaction) + .await + .map_err(VnidropError::repository)?; + transaction + .commit() + .await + .map_err(VnidropError::repository)?; + Ok(()) + } + + pub(crate) async fn complete_receiver_and_enqueue( &self, id: &str, verified_bytes: u64, ) -> Result<(), VnidropError> { - sqlx::query( + let mut transaction = self.pool.begin().await.map_err(VnidropError::repository)?; + let result = sqlx::query( r#" UPDATE targeted_transfers - SET verified_bytes = ?2, updated_at = ?3 - WHERE id = ?1 + SET state = 'completed', verified_bytes = ?2, updated_at = ?3 + WHERE id = ?1 AND role = 'receiver' AND state = 'transferring' "#, ) .bind(id) .bind(verified_bytes as i64) .bind(now_ms()) + .execute(&mut *transaction) + .await + .map_err(VnidropError::repository)?; + if result.rows_affected() == 0 { + return Err(VnidropError::InvalidTransition { + reason: "receiver transfer cannot be completed".to_string(), + }); + } + sqlx::query( + "INSERT OR IGNORE INTO targeted_completion_outbox (transfer_id, created_at, next_attempt_at) VALUES (?1, ?2, ?2)", + ) + .bind(id) + .bind(now_ms()) + .execute(&mut *transaction) + .await + .map_err(VnidropError::repository)?; + transaction + .commit() + .await + .map_err(VnidropError::repository)?; + Ok(()) + } + + pub(crate) async fn list_pending_completions( + &self, + ) -> Result, VnidropError> { + let rows = sqlx::query( + r#" + SELECT t.id, t.protocol_transfer_id, t.sender_endpoint_id, t.receiver_endpoint_id, + t.manifest_id, t.content_hash, t.transfer_name, t.file_count, t.total_size, + t.verified_bytes, t.blob_ticket, t.authorization_secret_handle, t.role, + t.state, t.created_at, t.updated_at + FROM targeted_transfers t + INNER JOIN targeted_completion_outbox o ON o.transfer_id = t.id + WHERE t.role = 'receiver' AND t.state = 'completed' AND o.next_attempt_at <= ?1 + ORDER BY o.next_attempt_at, o.created_at + "#, + ) + .bind(now_ms()) + .fetch_all(&self.pool) + .await + .map_err(VnidropError::repository)?; + rows.into_iter().map(row_to_full).collect() + } + + pub(crate) async fn clear_pending_completion(&self, id: &str) -> Result<(), VnidropError> { + sqlx::query("DELETE FROM targeted_completion_outbox WHERE transfer_id = ?1") + .bind(id) + .execute(&self.pool) + .await + .map_err(VnidropError::repository)?; + Ok(()) + } + + pub(crate) async fn defer_pending_completion( + &self, + id: &str, + next_attempt_at: i64, + ) -> Result<(), VnidropError> { + sqlx::query( + "UPDATE targeted_completion_outbox SET next_attempt_at = ?2 WHERE transfer_id = ?1", + ) + .bind(id) + .bind(next_attempt_at) .execute(&self.pool) .await .map_err(VnidropError::repository)?; Ok(()) } + pub(crate) async fn list_completed_sender_rows( + &self, + ) -> Result, VnidropError> { + let rows = sqlx::query( + r#" + SELECT t.id, t.protocol_transfer_id, t.sender_endpoint_id, t.receiver_endpoint_id, + t.manifest_id, t.content_hash, t.transfer_name, t.file_count, t.total_size, + t.verified_bytes, t.blob_ticket, t.authorization_secret_handle, t.role, + t.state, t.created_at, t.updated_at + FROM targeted_transfers t + INNER JOIN targeted_payload_release_outbox o ON o.transfer_id = t.id + WHERE t.role = 'sender' AND t.state = 'completed' + "#, + ) + .fetch_all(&self.pool) + .await + .map_err(VnidropError::repository)?; + rows.into_iter().map(row_to_full).collect() + } + + pub(crate) async fn clear_pending_payload_release(&self, id: &str) -> Result<(), VnidropError> { + sqlx::query("DELETE FROM targeted_payload_release_outbox WHERE transfer_id = ?1") + .bind(id) + .execute(&self.pool) + .await + .map_err(VnidropError::repository)?; + Ok(()) + } + pub(crate) async fn store_authorization( &self, id: &str, @@ -237,6 +402,17 @@ impl TargetedTransferStore { } pub(crate) async fn clear_authorization(&self, id: &str) -> Result<(), VnidropError> { + let mut transaction = self.pool.begin().await.map_err(VnidropError::repository)?; + sqlx::query("DELETE FROM targeted_completion_outbox WHERE transfer_id = ?1") + .bind(id) + .execute(&mut *transaction) + .await + .map_err(VnidropError::repository)?; + sqlx::query("DELETE FROM targeted_payload_release_outbox WHERE transfer_id = ?1") + .bind(id) + .execute(&mut *transaction) + .await + .map_err(VnidropError::repository)?; sqlx::query( r#" UPDATE targeted_transfers @@ -249,9 +425,13 @@ impl TargetedTransferStore { ) .bind(id) .bind(now_ms()) - .execute(&self.pool) + .execute(&mut *transaction) .await .map_err(VnidropError::repository)?; + transaction + .commit() + .await + .map_err(VnidropError::repository)?; Ok(()) } diff --git a/crates/vnidrop/src/tests/api_surface.rs b/crates/vnidrop/src/tests/api_surface.rs index 43d494d..6f18867 100644 --- a/crates/vnidrop/src/tests/api_surface.rs +++ b/crates/vnidrop/src/tests/api_surface.rs @@ -69,5 +69,5 @@ fn public_api_exposes_saved_device_surface_without_prototype_contact_entry_point let caps = crate::experimental_saved_device_capabilities(); assert_eq!(caps.domain_contract_version, 1); assert_eq!(caps.relationship_protocol_version, 1); - assert_eq!(caps.targeted_transfer_protocol_version, 1); + assert_eq!(caps.targeted_transfer_protocol_version, 2); } diff --git a/crates/vnidrop/src/tests/control_plane.rs b/crates/vnidrop/src/tests/control_plane.rs index a87e4d1..0664c35 100644 --- a/crates/vnidrop/src/tests/control_plane.rs +++ b/crates/vnidrop/src/tests/control_plane.rs @@ -9,6 +9,7 @@ use crate::{ invitation::Repository, secure_secret::FaultInjectingSecretStore, targeted_transfer::inbox::{TargetedOfferDecision, TargetedOfferInbox}, + targeted_transfer::{TargetedAuthorization, TargetedAuthorizationDraft}, CoreNetworkConfig, DeviceRelationshipState, ShareMetadataInput, ShareSource, SourceKind, TransferAccessMode, VnidropCore, VnidropError, }; @@ -17,6 +18,67 @@ struct RecordingSink { events: Mutex>, } +#[tokio::test] +async fn delivered_authorization_must_match_the_approved_offer_projection() { + let (inbox, _) = inbox_with_limits(1, 60_000, 5).await; + let offer = sample_offer("bound-offer", "sender-a"); + let submit = { + let inbox = inbox.clone(); + let offer = offer.clone(); + tokio::spawn(async move { inbox.submit(offer).await }) + }; + let started = std::time::Instant::now(); + while inbox.list().await.is_empty() { + assert!(started.elapsed() < std::time::Duration::from_secs(2)); + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } + + let exact = TargetedAuthorization::issue(TargetedAuthorizationDraft { + transfer_id: offer.transfer_id.clone(), + protocol_transfer_id: 42, + sender_endpoint_id: offer.sender_endpoint_id.clone(), + receiver_endpoint_id: offer.receiver_endpoint_id.clone(), + manifest_id: offer.manifest_id.clone(), + content_hash: offer.content_hash.clone(), + file_count: offer.file_count, + total_size: offer.total_size, + protocol_version: offer.protocol_version, + transfer_name: offer.transfer_name.clone(), + blob_ticket: "blob-a".to_string(), + }) + .unwrap(); + assert!(inbox.authorization_matches_pending(&exact).await); + + let substituted = TargetedAuthorization::issue(TargetedAuthorizationDraft { + manifest_id: "replacement-manifest".to_string(), + content_hash: "replacement-hash".to_string(), + transfer_name: "replacement.txt".to_string(), + total_size: 99, + blob_ticket: "blob-b".to_string(), + ..TargetedAuthorizationDraft { + transfer_id: offer.transfer_id, + protocol_transfer_id: 42, + sender_endpoint_id: offer.sender_endpoint_id, + receiver_endpoint_id: offer.receiver_endpoint_id, + manifest_id: offer.manifest_id, + content_hash: offer.content_hash, + file_count: offer.file_count, + total_size: offer.total_size, + protocol_version: offer.protocol_version, + transfer_name: offer.transfer_name, + blob_ticket: "blob-a".to_string(), + } + }) + .unwrap(); + assert!(!inbox.authorization_matches_pending(&substituted).await); + assert!( + inbox.list().await.len() == 1, + "rejection keeps the approved offer pending" + ); + inbox.discard("bound-offer").await; + let _ = submit.await; +} + impl CoreEventSink for RecordingSink { fn on_event(&self, event: CoreEvent) { self.events.lock().unwrap().push(event); diff --git a/crates/vnidrop/src/tests/persistence.rs b/crates/vnidrop/src/tests/persistence.rs index 767f862..c1bcdc3 100644 --- a/crates/vnidrop/src/tests/persistence.rs +++ b/crates/vnidrop/src/tests/persistence.rs @@ -35,6 +35,8 @@ async fn open_all_returns_all_domain_stores_and_schemas() { "protected_secret_refs", "blocked_endpoints", "targeted_transfers", + "targeted_completion_outbox", + "targeted_payload_release_outbox", "transfers", ] { let row = sqlx::query(&format!( @@ -51,6 +53,38 @@ async fn open_all_returns_all_domain_stores_and_schemas() { } } +#[tokio::test] +async fn open_all_migrates_targeted_completion_retry_schedule() { + let temp = tempfile::tempdir().unwrap(); + let db = temp.path().join("vnidrop.sqlite3"); + let options = sqlx::sqlite::SqliteConnectOptions::new() + .filename(&db) + .create_if_missing(true); + let pool = sqlx::SqlitePool::connect_with(options).await.unwrap(); + sqlx::query( + r#" + CREATE TABLE targeted_completion_outbox ( + transfer_id TEXT PRIMARY KEY, + created_at INTEGER NOT NULL + ) + "#, + ) + .execute(&pool) + .await + .unwrap(); + drop(pool); + + let _stores = persistence::open_all(temp.path()).await.unwrap(); + let pool = open_profile_pool(temp.path()).await; + let columns = sqlx::query("PRAGMA table_info(targeted_completion_outbox)") + .fetch_all(&pool) + .await + .unwrap(); + assert!(columns + .iter() + .any(|row| row.get::("name") == "next_attempt_at")); +} + #[tokio::test] async fn open_all_migrates_name_columns_without_losing_existing_local_labels() { let temp = tempfile::tempdir().unwrap(); diff --git a/crates/vnidrop/src/tests/targeted_transfer.rs b/crates/vnidrop/src/tests/targeted_transfer.rs index 87e9e38..d94decc 100644 --- a/crates/vnidrop/src/tests/targeted_transfer.rs +++ b/crates/vnidrop/src/tests/targeted_transfer.rs @@ -412,6 +412,7 @@ fn explicit_approval_gates_content_and_binds_authorization_to_receiver() { let bob_invitation_count = bob.core().list_transfers().unwrap().len(); let alice_eligibility_count = alice.core().list_pairing_eligibilities().unwrap().len(); let bob_eligibility_count = bob.core().list_pairing_eligibilities().unwrap().len(); + let bob_artifacts_before = bob.core().list_received_artifacts().unwrap(); let source_dir = tempfile::tempdir().unwrap(); let source_path = source_dir.path().join("payload.txt"); @@ -508,6 +509,10 @@ fn explicit_approval_gates_content_and_binds_authorization_to_receiver() { bob.core().list_pairing_eligibilities().unwrap().len(), bob_eligibility_count ); + assert_eq!( + bob.core().list_received_artifacts().unwrap().len(), + bob_artifacts_before.len() + ); let charlie_output = tempfile::tempdir().unwrap(); let leaked = charlie.core().receive_targeted_transfer( @@ -788,6 +793,127 @@ fn approved_transfer_resumes_after_restart_without_reapproval() { .unwrap(); assert_eq!(completed.state, TargetedTransferState::Completed); assert_eq!(completed.verified_bytes, b"resume me please".len() as u64); + let sender_completed = alice + .core() + .get_targeted_transfer(completed.id) + .unwrap() + .unwrap(); + assert_eq!(sender_completed.state, TargetedTransferState::Completed); + let started = Instant::now(); + while alice + .core() + .targeted_payload_is_registered_for_test(sender_completed.id.clone()) + .unwrap() + { + assert!( + started.elapsed() < Duration::from_secs(10), + "completed sender payload access was not released" + ); + std::thread::sleep(Duration::from_millis(25)); + } +} + +#[test] +fn completion_retries_after_receiver_restart_without_failing_published_receive() { + let alice = ProtectedNode::new(); + let bob = ProtectedNode::new(); + establish_saved(&alice, &bob, 11_025); + let transfer = approve_one(&alice, &bob, b"eventual completion", "payload.txt"); + bob.core().suppress_targeted_completion_for_test(true); + + let output = tempfile::tempdir().unwrap(); + bob.core() + .receive_targeted_transfer( + transfer.id.clone(), + output.path().to_string_lossy().into_owned(), + ) + .unwrap(); + assert_eq!( + std::fs::read(output.path().join("payload.txt")).unwrap(), + b"eventual completion" + ); + assert_eq!( + bob.core() + .get_targeted_transfer(transfer.id.clone()) + .unwrap() + .unwrap() + .state, + TargetedTransferState::Completed + ); + assert_eq!( + alice + .core() + .get_targeted_transfer(transfer.id.clone()) + .unwrap() + .unwrap() + .state, + TargetedTransferState::Approved + ); + + let bob = bob.restart(); + let started = Instant::now(); + loop { + if alice + .core() + .get_targeted_transfer(transfer.id.clone()) + .unwrap() + .is_some_and(|entry| entry.state == TargetedTransferState::Completed) + { + break; + } + assert!( + started.elapsed() < Duration::from_secs(15), + "completion outbox was not retried after restart" + ); + std::thread::sleep(Duration::from_millis(50)); + } + assert_eq!( + bob.core() + .get_targeted_transfer(transfer.id) + .unwrap() + .unwrap() + .state, + TargetedTransferState::Completed + ); +} + +#[test] +fn targeted_path_receive_preserves_no_overwrite_and_resumes_elsewhere() { + let alice = ProtectedNode::new(); + let bob = ProtectedNode::new(); + establish_saved(&alice, &bob, 11_026); + let transfer = approve_one(&alice, &bob, b"new payload", "payload.txt"); + let occupied = tempfile::tempdir().unwrap(); + std::fs::write(occupied.path().join("payload.txt"), b"keep me").unwrap(); + + assert!(bob + .core() + .receive_targeted_transfer( + transfer.id.clone(), + occupied.path().to_string_lossy().into_owned(), + ) + .is_err()); + assert_eq!( + std::fs::read(occupied.path().join("payload.txt")).unwrap(), + b"keep me" + ); + assert_eq!( + bob.core() + .get_targeted_transfer(transfer.id.clone()) + .unwrap() + .unwrap() + .state, + TargetedTransferState::Interrupted + ); + + let clean = tempfile::tempdir().unwrap(); + bob.core() + .resume_targeted_transfer(transfer.id, clean.path().to_string_lossy().into_owned()) + .unwrap(); + assert_eq!( + std::fs::read(clean.path().join("payload.txt")).unwrap(), + b"new payload" + ); } #[test] @@ -1258,6 +1384,131 @@ struct MemoryOutputSink { files: Mutex>>, } +#[derive(Default)] +struct FailingFinishSink { + starts: std::sync::atomic::AtomicUsize, + finishes: std::sync::atomic::AtomicUsize, + aborts: std::sync::atomic::AtomicUsize, +} + +#[derive(Default)] +struct GatedSink { + gate: (Mutex, std::sync::Condvar), + entered: std::sync::atomic::AtomicBool, + finishes: std::sync::atomic::AtomicUsize, + aborts: std::sync::atomic::AtomicUsize, +} + +impl GatedSink { + fn wait_until_entered(&self) { + let started = Instant::now(); + while !self.entered.load(std::sync::atomic::Ordering::SeqCst) { + assert!(started.elapsed() < Duration::from_secs(10)); + std::thread::sleep(Duration::from_millis(10)); + } + } + + fn release(&self) { + let (lock, wake) = &self.gate; + *lock.lock().unwrap() = true; + wake.notify_all(); + } +} + +impl ReceiveOutputSink for GatedSink { + fn start_file(&self, _relative_path: String) -> Result<(), VnidropError> { + Ok(()) + } + + fn write_chunk(&self, _relative_path: String, _bytes: Vec) -> Result<(), VnidropError> { + self.entered + .store(true, std::sync::atomic::Ordering::SeqCst); + let (lock, wake) = &self.gate; + let mut released = lock.lock().unwrap(); + while !*released { + released = wake.wait(released).unwrap(); + } + Ok(()) + } + + fn finish_file(&self, _relative_path: String) -> Result<(), VnidropError> { + self.finishes + .fetch_add(1, std::sync::atomic::Ordering::SeqCst); + Ok(()) + } + + fn abort_file(&self, _relative_path: String, _reason: String) -> Result<(), VnidropError> { + self.aborts + .fetch_add(1, std::sync::atomic::Ordering::SeqCst); + Ok(()) + } +} + +impl ReceiveOutputSinkV2 for GatedSink { + fn start_file(&self, relative_path: String) -> Result<(), VnidropError> { + ReceiveOutputSink::start_file(self, relative_path) + } + + fn write_chunk(&self, relative_path: String, bytes: Vec) -> Result<(), VnidropError> { + ReceiveOutputSink::write_chunk(self, relative_path, bytes) + } + + fn finish_file(&self, relative_path: String) -> Result { + ReceiveOutputSink::finish_file(self, relative_path.clone())?; + Ok(PublishedOutput { + locator_kind: ReceivedLocatorKind::FilesystemPath, + locator: format!("memory://{relative_path}"), + }) + } + + fn abort_file(&self, relative_path: String, reason: String) -> Result<(), VnidropError> { + ReceiveOutputSink::abort_file(self, relative_path, reason) + } +} + +impl ReceiveOutputSink for FailingFinishSink { + fn start_file(&self, _relative_path: String) -> Result<(), VnidropError> { + self.starts + .fetch_add(1, std::sync::atomic::Ordering::SeqCst); + Ok(()) + } + + fn write_chunk(&self, _relative_path: String, _bytes: Vec) -> Result<(), VnidropError> { + Ok(()) + } + + fn finish_file(&self, _relative_path: String) -> Result<(), VnidropError> { + self.finishes + .fetch_add(1, std::sync::atomic::Ordering::SeqCst); + Err(VnidropError::filesystem(anyhow::anyhow!("finish failed"))) + } + + fn abort_file(&self, _relative_path: String, _reason: String) -> Result<(), VnidropError> { + self.aborts + .fetch_add(1, std::sync::atomic::Ordering::SeqCst); + Ok(()) + } +} + +impl ReceiveOutputSinkV2 for FailingFinishSink { + fn start_file(&self, relative_path: String) -> Result<(), VnidropError> { + ReceiveOutputSink::start_file(self, relative_path) + } + + fn write_chunk(&self, relative_path: String, bytes: Vec) -> Result<(), VnidropError> { + ReceiveOutputSink::write_chunk(self, relative_path, bytes) + } + + fn finish_file(&self, relative_path: String) -> Result { + ReceiveOutputSink::finish_file(self, relative_path)?; + unreachable!() + } + + fn abort_file(&self, relative_path: String, reason: String) -> Result<(), VnidropError> { + ReceiveOutputSink::abort_file(self, relative_path, reason) + } +} + impl MemoryOutputSink { fn file(&self, relative_path: &str) -> Vec { self.files.lock().unwrap()[relative_path].clone() @@ -1350,6 +1601,86 @@ fn targeted_receive_and_resume_through_output_sinks() { ); } +#[test] +fn targeted_sink_finish_failure_is_the_only_terminal_callback() { + let alice = ProtectedNode::new(); + let bob = ProtectedNode::new(); + establish_saved(&alice, &bob, 11_075); + + for use_v2 in [false, true] { + let transfer = approve_one(&alice, &bob, b"sink failure", "sink.txt"); + let sink = Arc::new(FailingFinishSink::default()); + let result = if use_v2 { + bob.core() + .receive_targeted_transfer_with_output_sink_v2(transfer.id, sink.clone()) + } else { + bob.core() + .receive_targeted_transfer_with_output_sink(transfer.id, sink.clone()) + }; + assert!(result.is_err()); + assert_eq!(sink.starts.load(std::sync::atomic::Ordering::SeqCst), 1); + assert_eq!(sink.finishes.load(std::sync::atomic::Ordering::SeqCst), 1); + assert_eq!(sink.aborts.load(std::sync::atomic::Ordering::SeqCst), 0); + } +} + +#[test] +fn targeted_receive_rejects_a_concurrent_second_pull() { + let alice = ProtectedNode::new(); + let bob = ProtectedNode::new(); + establish_saved(&alice, &bob, 11_076); + let transfer = approve_one(&alice, &bob, b"concurrent pull", "sink.txt"); + let sink = Arc::new(GatedSink::default()); + let bob_core = bob.core(); + let transfer_id = transfer.id.clone(); + let sink_for_thread = sink.clone(); + let receive = std::thread::spawn(move || { + bob_core.receive_targeted_transfer_with_output_sink(transfer_id, sink_for_thread) + }); + sink.wait_until_entered(); + let second = bob.core().resume_targeted_transfer( + transfer.id, + tempfile::tempdir() + .unwrap() + .path() + .to_string_lossy() + .into_owned(), + ); + assert!(matches!( + second, + Err(VnidropError::InvalidTransition { .. }) + )); + sink.release(); + receive.join().unwrap().unwrap(); +} + +#[test] +fn targeted_cancel_aborts_each_sink_exactly_once() { + for use_v2 in [false, true] { + let alice = ProtectedNode::new(); + let bob = ProtectedNode::new(); + establish_saved(&alice, &bob, if use_v2 { 11_078 } else { 11_077 }); + let transfer = approve_one(&alice, &bob, b"cancel sink", "sink.txt"); + let sink = Arc::new(GatedSink::default()); + let bob_core = bob.core(); + let transfer_id = transfer.id.clone(); + let sink_for_thread = sink.clone(); + let receive = std::thread::spawn(move || { + if use_v2 { + bob_core.receive_targeted_transfer_with_output_sink_v2(transfer_id, sink_for_thread) + } else { + bob_core.receive_targeted_transfer_with_output_sink(transfer_id, sink_for_thread) + } + }); + sink.wait_until_entered(); + bob.core().cancel_targeted_transfer(transfer.id).unwrap(); + sink.release(); + assert!(receive.join().unwrap().is_err()); + assert_eq!(sink.finishes.load(std::sync::atomic::Ordering::SeqCst), 0); + assert_eq!(sink.aborts.load(std::sync::atomic::Ordering::SeqCst), 1); + } +} + #[test] fn decline_returns_typed_declined_outcome() { let alice = ProtectedNode::new(); diff --git a/crates/vnidrop/tests/experimental_domain.rs b/crates/vnidrop/tests/experimental_domain.rs index 5625e73..4148bdb 100644 --- a/crates/vnidrop/tests/experimental_domain.rs +++ b/crates/vnidrop/tests/experimental_domain.rs @@ -14,7 +14,7 @@ fn saved_device_protocols_are_explicitly_experimental_and_versioned() { ExperimentalSavedDeviceCapabilities { domain_contract_version: 1, relationship_protocol_version: 1, - targeted_transfer_protocol_version: 1, + targeted_transfer_protocol_version: 2, } ); }