diff --git a/DESIGN-DEVICE-HISTORY.md b/DESIGN-DEVICE-HISTORY.md index 1dda29c..675ab77 100644 --- a/DESIGN-DEVICE-HISTORY.md +++ b/DESIGN-DEVICE-HISTORY.md @@ -451,8 +451,15 @@ wake-up: refresh durable state via list/get APIs. Mid-transfer progress polish | Kind | Meaning | |---|---| | `offer-received` | A pre-approval offer is pending local approve/decline. | -| `offer-accepted` | Local approval completed; authorization is in core custody. | +| `approved` | Local approval completed; authorization is in core custody. | | `offer-declined` | Local decline completed. | +| `created`, `offering`, `awaiting-approval` | Sender-side durable setup and offer lifecycle changed. | +| `connecting`, `transferring`, `progress`, `interrupted` | Receiver-side pull lifecycle or verified payload progress changed. | +| `completed`, `cancelled`, `failed`, `deleted` | A durable targeted-transfer terminal snapshot changed. | + +Lifecycle payloads use `targeted_transfer_id`; consumers refresh the corresponding +snapshot after receiving the wake-up. Progress payloads remain advisory and the +durable snapshot is authoritative. Failures remain typed where callers can act differently, including: diff --git a/crates/vnidrop/CORE_FLOW.md b/crates/vnidrop/CORE_FLOW.md index 23dde58..8911c1e 100644 --- a/crates/vnidrop/CORE_FLOW.md +++ b/crates/vnidrop/CORE_FLOW.md @@ -56,7 +56,10 @@ Mid-transfer progress polish may follow. `relationship-changed`, `relationship-grant-rotated`, `saved-device-forgotten`, `device-blocked`. -**`targeted_transfer`:** `offer-received`, `offer-accepted`, `offer-declined`. +**`targeted_transfer`:** `offer-received`, `approved`, `offer-declined`, +`created`, `offering`, `awaiting-approval`, `connecting`, `transferring`, +`progress`, `interrupted`, `completed`, `cancelled`, `failed`, `deleted`. +Lifecycle payloads identify the durable row with `targeted_transfer_id`. ## Platform File Rules diff --git a/crates/vnidrop/src/api.rs b/crates/vnidrop/src/api.rs index 68bdcb4..4f216c6 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: 2, + targeted_transfer_protocol_version: 3, } } @@ -95,6 +95,7 @@ pub struct TargetedTransfer { pub sender_endpoint_id: String, pub receiver_endpoint_id: String, pub manifest_id: String, + pub transfer_name: String, pub file_count: u64, pub total_size: u64, /// Bytes verified so far; survives interruption for resume. diff --git a/crates/vnidrop/src/runtime/facade.rs b/crates/vnidrop/src/runtime/facade.rs index fc48025..77c5398 100644 --- a/crates/vnidrop/src/runtime/facade.rs +++ b/crates/vnidrop/src/runtime/facade.rs @@ -681,11 +681,7 @@ impl VnidropCore { /// /// Stops active streaming synchronously before asynchronous cleanup. pub fn cancel_targeted_transfer(&self, id: String) -> Result<(), VnidropError> { - if let Ok(Some(row)) = self.block_on(self.inner.targeted_store().get_row(&id)) { - let _ = self - .inner - .signal_targeted_transfer_cancel(row.protocol_transfer_id); - } + let _ = self.inner.signal_targeted_transfer_cancel_by_id(&id); self.block_on(self.inner.cancel_targeted_transfer(id)) } @@ -693,11 +689,7 @@ impl VnidropCore { /// /// Local denial is mandatory even when remote cleanup fails. pub fn delete_targeted_transfer(&self, id: String) -> Result<(), VnidropError> { - if let Ok(Some(row)) = self.block_on(self.inner.targeted_store().get_row(&id)) { - let _ = self - .inner - .signal_targeted_transfer_cancel(row.protocol_transfer_id); - } + let _ = self.inner.signal_targeted_transfer_cancel_by_id(&id); self.block_on(self.inner.delete_targeted_transfer(id)) } diff --git a/crates/vnidrop/src/runtime/mod.rs b/crates/vnidrop/src/runtime/mod.rs index f3a6e1b..d5edb14 100644 --- a/crates/vnidrop/src/runtime/mod.rs +++ b/crates/vnidrop/src/runtime/mod.rs @@ -112,12 +112,13 @@ pub(super) struct CoreInner { pub(super) access_policy: Arc, /// Sync mutex so cancel can remove + signal without awaiting (and without /// holding a Tokio lock across repository I/O). - pub(super) active_transfers: std::sync::Mutex>, + pub(super) active_transfers: Arc>>, + pub(super) active_targeted_transfers: Arc>>, // Active shares are protected by persistent Iroh tags. pub(super) active_shares: TokioMutex>, /// Content hash → active share transfer ids (root and collection members). /// Multiple transfers can share the same content-addressed hash. - pub(super) hash_to_transfer: TokioMutex>>, + pub(super) hash_to_transfer: Arc>>>, pub(super) connection_endpoints: TokioMutex>, pub(super) provider_task: TokioMutex>>, pub(super) delivery_receipt_notify: Notify, @@ -400,6 +401,57 @@ impl CoreInner { limits.max_saved_devices, limits.pairing_timeout_ms, )); + let active_transfers = + Arc::new(std::sync::Mutex::new(HashMap::::new())); + let active_targeted_transfers = Arc::new(std::sync::Mutex::new(HashMap::< + String, + ActiveTransfer, + >::new())); + let hash_to_transfer = Arc::new(TokioMutex::new(restored_hashes)); + let cleanup_targeted = active_targeted_transfers.clone(); + let cleanup_hashes = hash_to_transfer.clone(); + let cleanup_store = store.clone(); + let cleanup_custody = secret_custody.clone(); + let cleanup_targeted_store = targeted_transfers.clone(); + let targeted_cleanup = + Arc::new(move |row: crate::targeted_transfer::TargetedTransferRow| { + let targeted = cleanup_targeted.clone(); + let hashes = cleanup_hashes.clone(); + let blobs = cleanup_store.clone(); + let custody = cleanup_custody.clone(); + let transfers = cleanup_targeted_store.clone(); + Box::pin(async move { + let active_targeted = targeted + .lock() + .expect("active_targeted_transfers") + .remove(&row.id); + if let Some(active_targeted) = active_targeted { + let _ = active_targeted.cancel.send(()); + } + { + let mut hashes = hashes.lock().await; + hashes.retain(|_, owners| { + owners.remove(&row.protocol_transfer_id); + !owners.is_empty() + }); + } + if row.role == crate::targeted_transfer::TargetedTransferRole::Sender { + blobs + .tags() + .delete(targeted_tag_name(&row.id)) + .await + .map_err(crate::error::VnidropError::transfer)?; + } else if let Some(handle) = row.authorization_secret_handle { + if let Some(custody) = custody { + custody + .remove(&crate::secure_secret::SecretHandle::from_stored(handle)) + .await?; + } + transfers.clear_authorization(&row.id).await?; + } + Ok(()) + }) as crate::targeted_transfer::protocol::TargetedCleanupFuture + }); let router = Router::builder(endpoint.clone()) .accept(iroh_blobs::ALPN, blobs) .accept(HandshakeService::ALPN, handshake) @@ -417,6 +469,9 @@ impl CoreInner { endpoint.id().to_string(), relay_mode, relay_urls.clone(), + event_hub.clone(), + access_policy.clone(), + targeted_cleanup, ), ) .spawn(); @@ -441,9 +496,10 @@ impl CoreInner { transfer_slots: Semaphore::new(limits.max_concurrent_transfers as usize), limits, access_policy, - active_transfers: std::sync::Mutex::new(HashMap::new()), + active_transfers, + active_targeted_transfers, active_shares: TokioMutex::new(restored_active_shares), - hash_to_transfer: TokioMutex::new(restored_hashes), + hash_to_transfer, connection_endpoints: TokioMutex::new(HashMap::new()), provider_task: TokioMutex::new(None), delivery_receipt_notify: Notify::new(), @@ -457,8 +513,15 @@ impl CoreInner { }); // In-flight connecting/transferring transfers become Interrupted across restart. - if let Err(error) = inner.targeted_store().mark_interrupted_in_flight().await { - tracing::warn!(%error, "failed to mark in-flight targeted transfers interrupted"); + match inner.targeted_store().mark_interrupted_in_flight().await { + Ok(ids) => { + for id in ids { + inner.emit_targeted_lifecycle(&id, "interrupted"); + } + } + Err(error) => { + tracing::warn!(%error, "failed to mark in-flight targeted transfers interrupted"); + } } // Restore permanent receiver ACLs for approved targeted shares. if let Err(error) = inner.restore_targeted_transfer_access().await { diff --git a/crates/vnidrop/src/runtime/receive.rs b/crates/vnidrop/src/runtime/receive.rs index 4d45b59..69b0461 100644 --- a/crates/vnidrop/src/runtime/receive.rs +++ b/crates/vnidrop/src/runtime/receive.rs @@ -184,7 +184,10 @@ impl CoreInner { pub(super) async fn receive_targeted_payload( self: &Arc, + targeted_transfer_id: &str, transfer_id: u64, + expected_file_count: u64, + expected_payload_size: u64, mut blob_ticket: BlobTicket, target: ReceiveTarget, ) -> Result<()> { @@ -202,11 +205,11 @@ impl CoreInner { .map_err(VnidropError::network)?; blob_ticket = BlobTicket::new(sender_addr, blob_ticket.hash(), blob_ticket.format()); let (cancel, mut cancelled) = oneshot::channel(); - self.active_transfers + self.active_targeted_transfers .lock() - .expect("active_transfers") + .expect("active_targeted_transfers") .insert( - transfer_id, + targeted_transfer_id.to_string(), ActiveTransfer { direction: TransferDirection::Receive, cancel, @@ -215,18 +218,28 @@ impl CoreInner { let result = tokio::select! { biased; _ = &mut cancelled => Err(VnidropError::cancelled("transfer cancelled").into()), - result = self.download_targeted_payload(transfer_id, blob_ticket, target) => result, + result = self.download_targeted_payload( + targeted_transfer_id, + transfer_id, + expected_file_count, + expected_payload_size, + blob_ticket, + target, + ) => result, }; - self.active_transfers + self.active_targeted_transfers .lock() - .expect("active_transfers") - .remove(&transfer_id); + .expect("active_targeted_transfers") + .remove(targeted_transfer_id); result } async fn download_targeted_payload( &self, + targeted_transfer_id: &str, transfer_id: u64, + expected_file_count: u64, + expected_payload_size: u64, blob_ticket: BlobTicket, target: ReceiveTarget, ) -> Result<()> { @@ -241,25 +254,42 @@ impl CoreInner { .await .map_err(VnidropError::network)?; let hash_and_format = blob_ticket.hash_and_format(); - let (_hash_seq, sizes) = + let (hash_seq, sizes) = get_hash_seq_and_sizes(&connection, &hash_and_format.hash, 1024 * 1024 * 32, None) .await .context("failed to get targeted payload sizes") .map_err(VnidropError::network)?; - let total_size = sizes + let remote_size = sizes .iter() .try_fold(0u64, |total, size| total.checked_add(*size)) .context("remote collection size overflow")?; let total_files = sizes.len().saturating_sub(1) as u64; + if total_files != expected_file_count { + anyhow::bail!( + "targeted payload file count {total_files} does not match authorized count {expected_file_count}" + ); + } + let hash_sequence_bytes = hash_seq.len() as u64 * 32; + let collection_metadata_bytes = sizes.first().copied().unwrap_or(0); + let payload_size = sizes + .iter() + .skip(1) + .try_fold(0u64, |total, size| total.checked_add(*size)) + .context("remote targeted payload size overflow")?; + if payload_size != expected_payload_size { + anyhow::bail!( + "targeted payload size {payload_size} does not match authorized size {expected_payload_size}" + ); + } if total_files > self.limits.max_collection_files { anyhow::bail!( "remote collection has {total_files} files, limit is {}", self.limits.max_collection_files ); } - if total_size > self.limits.max_total_bytes { + if remote_size > self.limits.max_total_bytes { anyhow::bail!( - "remote collection size {total_size} exceeds limit {}", + "remote collection size {remote_size} exceeds limit {}", self.limits.max_total_bytes ); } @@ -271,14 +301,62 @@ impl CoreInner { anyhow::bail!("targeted download ended without completion"); }; match item { - GetProgressItem::Progress(downloaded) => self.emit_transfer( - transfer_id, - "receive", - "download", - "progress", - json!({ "downloaded": downloaded, "total_size": total_size }), - ), - GetProgressItem::Done(_) => break, + GetProgressItem::Progress(_) => { + let verified = self + .store + .remote() + .local(hash_and_format) + .await? + .local_bytes() + .saturating_sub(hash_sequence_bytes) + .saturating_sub(collection_metadata_bytes) + .min(payload_size); + if self + .targeted_store() + .advance_verified_bytes(targeted_transfer_id, verified) + .await? + { + self.emit_transfer( + transfer_id, + "receive", + "download", + "progress", + json!({ "downloaded": verified, "total_size": payload_size }), + ); + self.emit_targeted_lifecycle(targeted_transfer_id, "progress"); + } + } + GetProgressItem::Done(_) => { + let verified = self + .store + .remote() + .local(hash_and_format) + .await? + .local_bytes() + .saturating_sub(hash_sequence_bytes) + .saturating_sub(collection_metadata_bytes) + .min(payload_size); + if verified != payload_size { + anyhow::bail!( + "targeted payload completed with {verified} verified bytes, expected {payload_size}" + ); + } + if self + .targeted_store() + .advance_verified_bytes(targeted_transfer_id, verified) + .await? + { + self.emit_transfer( + transfer_id, + "receive", + "download", + "progress", + json!({ "downloaded": verified, "total_size": payload_size }), + ); + self.emit_targeted_lifecycle(targeted_transfer_id, "progress"); + } + break; + } GetProgressItem::Error(error) => { return Err(VnidropError::network(anyhow::anyhow!( "targeted download failed: {error}" diff --git a/crates/vnidrop/src/runtime/targeted.rs b/crates/vnidrop/src/runtime/targeted.rs index 0ce3471..8fd9065 100644 --- a/crates/vnidrop/src/runtime/targeted.rs +++ b/crates/vnidrop/src/runtime/targeted.rs @@ -28,6 +28,14 @@ use crate::{ }; impl CoreInner { + pub(crate) fn emit_targeted_lifecycle(&self, transfer_id: &str, kind: &str) { + self.emit_endpoint( + "targeted_transfer", + kind, + serde_json::json!({ "targeted_transfer_id": transfer_id }), + ); + } + fn connection_timeout(&self) -> std::time::Duration { std::time::Duration::from_millis(self.limits.connection_timeout_ms) } @@ -197,9 +205,16 @@ impl CoreInner { .targeted_store() .sender_payloads_for_peer(peer_endpoint_id) .await?; - // Signal active transfers synchronously before awaiting share teardown. - for protocol_transfer_id in &protocol_ids { - let _ = self.take_active_transfer(*protocol_transfer_id); + let active_ids = self.targeted_store().ids_for_peer(peer_endpoint_id).await?; + for id in active_ids { + let _ = self.signal_targeted_transfer_cancel_by_id(&id); + } + let cancelled = self + .targeted_store() + .cancel_by_peer(peer_endpoint_id) + .await?; + for id in &cancelled { + self.emit_targeted_lifecycle(id, "cancelled"); } for protocol_transfer_id in &protocol_ids { self.teardown_targeted_payload(*protocol_transfer_id, None) @@ -209,12 +224,15 @@ impl CoreInner { self.teardown_targeted_payload(protocol_transfer_id, Some(&id)) .await; } - self.targeted_store().cancel_by_peer(peer_endpoint_id).await + Ok(cancelled.len() as u64) } - /// Synchronously stop streaming for one transfer (facade calls this first). - pub(super) fn signal_targeted_transfer_cancel(&self, protocol_transfer_id: u64) -> bool { - self.take_active_transfer(protocol_transfer_id).is_some() + pub(super) fn signal_targeted_transfer_cancel_by_id(&self, id: &str) -> bool { + self.active_targeted_transfers + .lock() + .expect("active_targeted_transfers") + .remove(id) + .is_some_and(|active| active.cancel.send(()).is_ok()) } pub(super) async fn cancel_targeted_transfer(&self, id: String) -> Result<(), VnidropError> { @@ -224,42 +242,41 @@ impl CoreInner { self.targeted_offers.discard(&id).await; return Ok(()); }; - let _ = self.take_active_transfer(row.protocol_transfer_id); self.targeted_offers.discard(&id).await; + let changed = store + .transition_terminal(&id, TargetedTransferState::Cancelled, false) + .await?; + if changed { + self.emit_targeted_lifecycle(&id, "cancelled"); + } self.access_policy .remove_transfer(row.protocol_transfer_id) .await; self.teardown_targeted_payload(row.protocol_transfer_id, Some(&row.id)) .await; - if !matches!( - row.state, - TargetedTransferState::Completed - | TargetedTransferState::Declined - | TargetedTransferState::Cancelled - | TargetedTransferState::Failed - | TargetedTransferState::Deleted - ) { - let _ = store - .set_state_from_any(&id, TargetedTransferState::Cancelled) - .await; + if row.role == TargetedTransferRole::Receiver { + if let Some(handle) = &row.authorization_secret_handle { + if let Some(custody) = &self.secret_custody { + custody + .remove(&SecretHandle::from_stored(handle.clone())) + .await?; + } + store.clear_authorization(&id).await?; + } } - // Best-effort remote withdraw of an unapproved live offer. - if row.role == TargetedTransferRole::Sender - && matches!( - row.state, - TargetedTransferState::Offering | TargetedTransferState::AwaitingApproval - ) - { - if let Ok(addr) = self - .device_relationships - .peer_addr(&row.receiver_endpoint_id) - .await - { + // Best-effort idempotent peer teardown for both pre- and post-approval work. + let peer_id = match row.role { + TargetedTransferRole::Sender => &row.receiver_endpoint_id, + TargetedTransferRole::Receiver => &row.sender_endpoint_id, + }; + if row.state != TargetedTransferState::Deleted { + if let Ok(addr) = self.device_relationships.peer_addr(peer_id).await { let client = TargetedTransferProtocol::client(self.endpoint.clone(), addr); let _ = tokio::time::timeout( self.connection_timeout(), client.cancel_offer(CancelTargetedOffer { transfer_id: id.clone(), + terminal: Some(TargetedTransferState::Cancelled), }), ) .await; @@ -277,8 +294,27 @@ impl CoreInner { self.targeted_offers.discard(&id).await; return Ok(()); }; - // Durable local denial first — remote cleanup is best-effort. - let _ = self.take_active_transfer(row.protocol_transfer_id); + let already_deleted = row.state == TargetedTransferState::Deleted; + if already_deleted && row.authorization_secret_handle.is_none() && row.blob_ticket.is_none() + { + self.targeted_offers.discard(&id).await; + return Ok(()); + } + // Signal first, then commit denial before any fallible remote or blob cleanup. + let peer_id = match row.role { + TargetedTransferRole::Sender => &row.receiver_endpoint_id, + TargetedTransferRole::Receiver => &row.sender_endpoint_id, + }; + let changed = if already_deleted { + false + } else { + store + .transition_terminal(&id, TargetedTransferState::Deleted, false) + .await? + }; + if changed { + self.emit_targeted_lifecycle(&id, "deleted"); + } self.targeted_offers.discard(&id).await; self.access_policy .remove_transfer(row.protocol_transfer_id) @@ -287,27 +323,26 @@ impl CoreInner { .await; if let Some(handle) = &row.authorization_secret_handle { if let Some(custody) = &self.secret_custody { - let _ = custody + custody .remove(&SecretHandle::from_stored(handle.clone())) - .await; + .await?; + store.clear_authorization(&id).await?; } + } else { + store.clear_authorization(&id).await?; } - store.clear_authorization(&id).await?; - if row.state != TargetedTransferState::Deleted { - if !matches!( - row.state, - TargetedTransferState::Completed - | TargetedTransferState::Declined - | TargetedTransferState::Cancelled - | TargetedTransferState::Failed - ) { - let _ = store - .set_state_from_any(&id, TargetedTransferState::Cancelled) - .await; - } - let _ = store - .set_state_from_any(&id, TargetedTransferState::Deleted) + if !already_deleted { + if let Ok(addr) = self.device_relationships.peer_addr(peer_id).await { + let client = TargetedTransferProtocol::client(self.endpoint.clone(), addr); + let _ = tokio::time::timeout( + self.connection_timeout(), + client.cancel_offer(CancelTargetedOffer { + transfer_id: id.clone(), + terminal: Some(TargetedTransferState::Deleted), + }), + ) .await; + } } Ok(()) } @@ -340,10 +375,23 @@ impl CoreInner { } match self.targeted_offers.respond(&transfer_id, accepted).await { - Ok(Some(auth)) => { - self.persist_receiver_authorization(&auth).await?; - Ok(TargetedOfferResponse::Approved { transfer_id }) - } + Ok(Some(auth)) => match self.persist_receiver_authorization(&auth).await { + Ok(()) => { + self.emit_targeted_lifecycle(&transfer_id, "approved"); + Ok(TargetedOfferResponse::Approved { transfer_id }) + } + Err(error) => { + if self + .targeted_store() + .set_state_from_any(&transfer_id, TargetedTransferState::Failed) + .await + .is_ok() + { + self.emit_targeted_lifecycle(&transfer_id, "failed"); + } + Err(error) + } + }, Ok(None) => Ok(TargetedOfferResponse::Declined), Err(crate::targeted_transfer::RespondError::Unknown) => Err( VnidropError::invalid_input(anyhow::anyhow!("unknown targeted offer")), @@ -459,6 +507,7 @@ impl CoreInner { .await; return Err(error); } + self.emit_targeted_lifecycle(&transfer_uuid, "created"); if let Err(error) = store .set_state( &transfer_uuid, @@ -471,6 +520,7 @@ impl CoreInner { .await; return Err(error); } + self.emit_targeted_lifecycle(&transfer_uuid, "offering"); let addr = match self .device_relationships @@ -479,13 +529,17 @@ impl CoreInner { { Ok(addr) => addr, Err(error) => { - let _ = store + if store .set_state( &transfer_uuid, TargetedTransferState::Offering, TargetedTransferState::Failed, ) - .await; + .await + .is_ok() + { + self.emit_targeted_lifecycle(&transfer_uuid, "failed"); + } self.teardown_targeted_payload(protocol_transfer_id, Some(&transfer_uuid)) .await; return Err(error); @@ -497,25 +551,33 @@ impl CoreInner { { Ok(Ok(challenge)) => challenge, Ok(Err(error)) => { - let _ = store + if store .set_state( &transfer_uuid, TargetedTransferState::Offering, TargetedTransferState::Failed, ) - .await; + .await + .is_ok() + { + self.emit_targeted_lifecycle(&transfer_uuid, "failed"); + } self.teardown_targeted_payload(protocol_transfer_id, Some(&transfer_uuid)) .await; return Err(map_connect_failure(error)); } Err(_) => { - let _ = store + if store .set_state( &transfer_uuid, TargetedTransferState::Offering, TargetedTransferState::Failed, ) - .await; + .await + .is_ok() + { + self.emit_targeted_lifecycle(&transfer_uuid, "failed"); + } self.teardown_targeted_payload(protocol_transfer_id, Some(&transfer_uuid)) .await; return Err(VnidropError::device_unavailable(anyhow::anyhow!( @@ -531,13 +593,17 @@ impl CoreInner { { Ok(proof) => proof, Err(error) => { - let _ = store + if store .set_state( &transfer_uuid, TargetedTransferState::Offering, TargetedTransferState::Failed, ) - .await; + .await + .is_ok() + { + self.emit_targeted_lifecycle(&transfer_uuid, "failed"); + } self.teardown_targeted_payload(protocol_transfer_id, Some(&transfer_uuid)) .await; return Err(error); @@ -558,6 +624,7 @@ impl CoreInner { .await; return Err(error); } + self.emit_targeted_lifecycle(&transfer_uuid, "awaiting-approval"); let response = match tokio::time::timeout( self.connection_timeout() + self.offer_wait_timeout(), @@ -586,25 +653,33 @@ impl CoreInner { { Ok(Ok(response)) => response, Ok(Err(error)) => { - let _ = store + if store .set_state( &transfer_uuid, TargetedTransferState::AwaitingApproval, TargetedTransferState::Failed, ) - .await; + .await + .is_ok() + { + self.emit_targeted_lifecycle(&transfer_uuid, "failed"); + } self.teardown_targeted_payload(protocol_transfer_id, Some(&transfer_uuid)) .await; return Err(map_connect_failure(error)); } Err(_) => { - let _ = store + if store .set_state( &transfer_uuid, TargetedTransferState::AwaitingApproval, TargetedTransferState::Failed, ) - .await; + .await + .is_ok() + { + self.emit_targeted_lifecycle(&transfer_uuid, "failed"); + } self.teardown_targeted_payload(protocol_transfer_id, Some(&transfer_uuid)) .await; return Err(VnidropError::offer_timeout(anyhow::anyhow!( @@ -616,13 +691,17 @@ impl CoreInner { match response { WireOfferResponse::Accepted => {} WireOfferResponse::Declined { reason } => { - let _ = store + if store .set_state( &transfer_uuid, TargetedTransferState::AwaitingApproval, TargetedTransferState::Declined, ) - .await; + .await + .is_ok() + { + self.emit_targeted_lifecycle(&transfer_uuid, "offer-declined"); + } self.teardown_targeted_payload(protocol_transfer_id, Some(&transfer_uuid)) .await; return Err(VnidropError::permission(anyhow::anyhow!( @@ -630,13 +709,17 @@ impl CoreInner { ))); } WireOfferResponse::Refused { reason } => { - let _ = store + if store .set_state( &transfer_uuid, TargetedTransferState::AwaitingApproval, TargetedTransferState::Failed, ) - .await; + .await + .is_ok() + { + self.emit_targeted_lifecycle(&transfer_uuid, "failed"); + } self.teardown_targeted_payload(protocol_transfer_id, Some(&transfer_uuid)) .await; return Err(map_offer_refuse_reason(&reason)); @@ -663,9 +746,13 @@ impl CoreInner { }) { Ok(authorization) => authorization, Err(error) => { - let _ = store + if store .set_state_from_any(&transfer_uuid, TargetedTransferState::Failed) - .await; + .await + .is_ok() + { + self.emit_targeted_lifecycle(&transfer_uuid, "failed"); + } self.teardown_targeted_payload(protocol_transfer_id, Some(&transfer_uuid)) .await; return Err(error); @@ -675,9 +762,13 @@ impl CoreInner { .persist_authorization_secret(&transfer_uuid, &authorization) .await { - let _ = store + if store .set_state_from_any(&transfer_uuid, TargetedTransferState::Failed) - .await; + .await + .is_ok() + { + self.emit_targeted_lifecycle(&transfer_uuid, "failed"); + } self.teardown_targeted_payload(protocol_transfer_id, Some(&transfer_uuid)) .await; return Err(error); @@ -685,9 +776,13 @@ impl CoreInner { let encoded = match authorization.encode() { Ok(encoded) => encoded, Err(error) => { - let _ = store + if store .set_state_from_any(&transfer_uuid, TargetedTransferState::Failed) - .await; + .await + .is_ok() + { + self.emit_targeted_lifecycle(&transfer_uuid, "failed"); + } self.teardown_targeted_payload(protocol_transfer_id, Some(&transfer_uuid)) .await; return Err(error); @@ -706,6 +801,7 @@ impl CoreInner { .await; return Err(error); } + self.emit_targeted_lifecycle(&transfer_uuid, "approved"); let deliver = match client .deliver_authorization(DeliverTargetedAuthorization { @@ -718,18 +814,26 @@ impl CoreInner { { Ok(deliver) => deliver, Err(error) => { - let _ = store + if store .set_state_from_any(&transfer_uuid, TargetedTransferState::Failed) - .await; + .await + .is_ok() + { + self.emit_targeted_lifecycle(&transfer_uuid, "failed"); + } self.teardown_targeted_payload(protocol_transfer_id, Some(&transfer_uuid)) .await; return Err(error); } }; if deliver != crate::targeted_transfer::protocol::DeliverAuthorizationResponse::Stored { - let _ = store + if store .set_state_from_any(&transfer_uuid, TargetedTransferState::Failed) - .await; + .await + .is_ok() + { + self.emit_targeted_lifecycle(&transfer_uuid, "failed"); + } self.teardown_targeted_payload(protocol_transfer_id, Some(&transfer_uuid)) .await; return Err(VnidropError::network(anyhow::anyhow!( @@ -899,6 +1003,7 @@ impl CoreInner { TargetedTransferState::Connecting, ) .await?; + self.emit_targeted_lifecycle(&auth.transfer_id, "connecting"); store .set_state( &auth.transfer_id, @@ -906,6 +1011,7 @@ impl CoreInner { TargetedTransferState::Transferring, ) .await?; + self.emit_targeted_lifecycle(&auth.transfer_id, "transferring"); } TargetedTransferState::Connecting | TargetedTransferState::Transferring => { return Err(VnidropError::InvalidTransition { @@ -926,7 +1032,14 @@ impl CoreInner { let blob_ticket = BlobTicket::from_str_compat(&auth.blob_ticket) .map_err(|error| VnidropError::ticket(anyhow::anyhow!(error)))?; let receive_result = self - .receive_targeted_payload(auth.protocol_transfer_id, blob_ticket, target) + .receive_targeted_payload( + &auth.transfer_id, + auth.protocol_transfer_id, + auth.file_count, + auth.total_size, + blob_ticket, + target, + ) .await; match receive_result { @@ -937,6 +1050,7 @@ impl CoreInner { store .complete_receiver_and_enqueue(&auth.transfer_id, row.total_size) .await?; + self.emit_targeted_lifecycle(&auth.transfer_id, "completed"); if self.acknowledge_targeted_completion(auth).await.is_ok() { store.clear_pending_completion(&auth.transfer_id).await?; } @@ -947,13 +1061,12 @@ impl CoreInner { if matches!( row.state, TargetedTransferState::Connecting | TargetedTransferState::Transferring - ) { - let _ = store - .set_state_from_any( - &auth.transfer_id, - TargetedTransferState::Interrupted, - ) - .await; + ) && store + .set_state_from_any(&auth.transfer_id, TargetedTransferState::Interrupted) + .await + .is_ok() + { + self.emit_targeted_lifecycle(&auth.transfer_id, "interrupted"); } } Err(VnidropError::transfer(error)) @@ -1013,15 +1126,39 @@ impl CoreInner { let handle = custody .protect(SecretKind::TargetedAuthorization, material, None) .await?; - self.targeted_store() + if let Err(error) = self + .targeted_store() .store_authorization(transfer_id, &authorization.blob_ticket, handle.as_str()) .await + { + if let Err(cleanup_error) = custody.remove(&handle).await { + tracing::warn!(%cleanup_error, "failed to roll back targeted authorization secret"); + } + return Err(error); + } + Ok(()) } async fn persist_receiver_authorization(&self, encoded: &str) -> Result<(), VnidropError> { let auth = TargetedAuthorization::decode(encoded)?; let store = self.targeted_store(); if store.get_row(&auth.transfer_id).await?.is_none() { + let invitation_collision = self + .repository + .list_transfers() + .await + .map_err(VnidropError::repository)? + .into_iter() + .any(|transfer| transfer.transfer_id == auth.protocol_transfer_id); + if invitation_collision + || store + .contains_protocol_id(auth.protocol_transfer_id) + .await? + { + return Err(VnidropError::invalid_input(anyhow::anyhow!( + "targeted transfer protocol id collides with local work" + ))); + } let now = now_ms(); store .insert(&TargetedTransferRow { diff --git a/crates/vnidrop/src/targeted_transfer/inbox.rs b/crates/vnidrop/src/targeted_transfer/inbox.rs index 21f3906..b5a3129 100644 --- a/crates/vnidrop/src/targeted_transfer/inbox.rs +++ b/crates/vnidrop/src/targeted_transfer/inbox.rs @@ -148,7 +148,7 @@ impl TargetedOfferInbox { "targeted_transfer", "offer-received", json!({ - "transfer_id": offer.transfer_id, + "targeted_transfer_id": offer.transfer_id, "sender_endpoint_id": offer.sender_endpoint_id, "file_count": offer.file_count, "total_size": offer.total_size, @@ -293,7 +293,7 @@ impl TargetedOfferInbox { self.event_hub.emit_endpoint( "targeted_transfer", "offer-declined", - json!({ "transfer_id": transfer_id }), + json!({ "targeted_transfer_id": transfer_id }), ); return Ok(None); } @@ -330,11 +330,6 @@ impl TargetedOfferInbox { authorization: Some(auth.clone()), }, ); - self.event_hub.emit_endpoint( - "targeted_transfer", - "offer-accepted", - json!({ "transfer_id": transfer_id }), - ); Ok(Some(auth)) } Ok(Err(())) | Err(_) => { diff --git a/crates/vnidrop/src/targeted_transfer/protocol.rs b/crates/vnidrop/src/targeted_transfer/protocol.rs index 6e7e72a..96d9a5e 100644 --- a/crates/vnidrop/src/targeted_transfer/protocol.rs +++ b/crates/vnidrop/src/targeted_transfer/protocol.rs @@ -4,6 +4,7 @@ //! summary and relationship proof only — never a reusable share ticket. use std::fmt; +use std::{future::Future, pin::Pin}; use anyhow::Result; use iroh::{ @@ -41,8 +42,15 @@ pub(crate) struct TargetedTransferProtocol { local_endpoint_id: String, relay_mode: CoreRelayMode, custom_relay_urls: Vec, + event_hub: std::sync::Arc, + access_policy: std::sync::Arc, + cleanup: + std::sync::Arc TargetedCleanupFuture + Send + Sync>, } +pub(crate) type TargetedCleanupFuture = + Pin> + Send>>; + impl fmt::Debug for TargetedTransferProtocol { fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { formatter.write_str("TargetedTransferProtocol") @@ -50,8 +58,9 @@ impl fmt::Debug for TargetedTransferProtocol { } impl TargetedTransferProtocol { - pub(crate) const ALPN: &'static [u8] = b"/vnidrop/targeted-transfer/2"; + pub(crate) const ALPN: &'static [u8] = b"/vnidrop/targeted-transfer/3"; + #[allow(clippy::too_many_arguments)] pub(crate) fn new( relationships: std::sync::Arc, inbox: TargetedOfferInbox, @@ -60,6 +69,11 @@ impl TargetedTransferProtocol { local_endpoint_id: String, relay_mode: CoreRelayMode, custom_relay_urls: Vec, + event_hub: std::sync::Arc, + access_policy: std::sync::Arc, + cleanup: std::sync::Arc< + dyn Fn(super::TargetedTransferRow) -> TargetedCleanupFuture + Send + Sync, + >, ) -> Self { Self { relationships, @@ -69,6 +83,9 @@ impl TargetedTransferProtocol { local_endpoint_id, relay_mode, custom_relay_urls, + event_hub, + access_policy, + cleanup, } } @@ -281,6 +298,14 @@ impl TargetedTransferProtocol { if row.authorization_secret_handle.is_some() { return DeliverAuthorizationResponse::Stored; } + if matches!( + row.state, + TargetedTransferState::Failed + | TargetedTransferState::Cancelled + | TargetedTransferState::Deleted + ) { + return DeliverAuthorizationResponse::Rejected; + } } if tokio::time::Instant::now() >= deadline { return DeliverAuthorizationResponse::Rejected; @@ -305,9 +330,59 @@ impl TargetedTransferProtocol { return CancelWireOfferResponse::Cancelled; } if let Ok(Some(row)) = self.store.get_row(&cancel.transfer_id).await { - if row.sender_endpoint_id != remote_endpoint_id { + let remote_is_peer = match row.role { + TargetedTransferRole::Sender => row.receiver_endpoint_id == remote_endpoint_id, + TargetedTransferRole::Receiver => row.sender_endpoint_id == remote_endpoint_id, + }; + if !remote_is_peer { return CancelWireOfferResponse::Rejected; } + if let Some(terminal) = cancel.terminal { + if matches!( + terminal, + TargetedTransferState::Cancelled | TargetedTransferState::Deleted + ) { + if row.role == TargetedTransferRole::Sender { + self.access_policy + .remove_transfer(row.protocol_transfer_id) + .await; + } + let changed = if row.state == TargetedTransferState::Cancelled { + false + } else if matches!( + row.state, + TargetedTransferState::Completed + | TargetedTransferState::Declined + | TargetedTransferState::Failed + | TargetedTransferState::Deleted + ) { + return CancelWireOfferResponse::Cancelled; + } else { + let Ok(changed) = self + .store + .transition_terminal( + &cancel.transfer_id, + TargetedTransferState::Cancelled, + false, + ) + .await + else { + return CancelWireOfferResponse::Rejected; + }; + changed + }; + if changed { + self.event_hub.emit_endpoint( + "targeted_transfer", + "cancelled", + serde_json::json!({ "targeted_transfer_id": cancel.transfer_id }), + ); + } + if (self.cleanup)(row.clone()).await.is_err() { + return CancelWireOfferResponse::Rejected; + } + } + } return CancelWireOfferResponse::Cancelled; } CancelWireOfferResponse::Cancelled @@ -367,7 +442,16 @@ impl TargetedTransferProtocol { .mark_sender_completed(&completion.transfer_id) .await { - Ok(()) => CompletionResponse::Recorded, + Ok(changed) => { + if changed { + self.event_hub.emit_endpoint( + "targeted_transfer", + "completed", + serde_json::json!({ "targeted_transfer_id": completion.transfer_id }), + ); + } + CompletionResponse::Recorded + } Err(_) => CompletionResponse::Rejected, } } @@ -505,6 +589,7 @@ pub(crate) enum DeliverAuthorizationResponse { #[derive(Debug, Clone, Serialize, Deserialize)] pub(crate) struct CancelTargetedOffer { pub(crate) transfer_id: String, + pub(crate) terminal: Option, } #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] diff --git a/crates/vnidrop/src/targeted_transfer/state.rs b/crates/vnidrop/src/targeted_transfer/state.rs index de8c318..95f0908 100644 --- a/crates/vnidrop/src/targeted_transfer/state.rs +++ b/crates/vnidrop/src/targeted_transfer/state.rs @@ -4,6 +4,9 @@ impl TargetedTransferState { /// Validates a durable state change without exposing foreign state mutation. pub fn validate_transition_to(self, next: Self) -> Result<(), VnidropError> { let allowed = matches!( + (self, next), + (state, Self::Deleted) if state != Self::Deleted + ) || matches!( (self, next), ( Self::Preparing, @@ -16,7 +19,7 @@ impl TargetedTransferState { Self::Approved | Self::Declined | Self::Cancelled | Self::Failed ) | ( Self::Approved, - Self::Connecting | Self::Cancelled | Self::Failed | Self::Deleted + Self::Connecting | Self::Cancelled | Self::Failed ) | ( Self::Connecting, Self::Transferring | Self::Interrupted | Self::Cancelled | Self::Failed @@ -25,10 +28,7 @@ impl TargetedTransferState { Self::Completed | Self::Interrupted | Self::Cancelled | Self::Failed ) | ( Self::Interrupted, - Self::Connecting | Self::Cancelled | Self::Failed | Self::Deleted - ) | ( - Self::Completed | Self::Declined | Self::Cancelled | Self::Failed, - Self::Deleted + Self::Connecting | Self::Cancelled | Self::Failed ) ); if allowed { diff --git a/crates/vnidrop/src/targeted_transfer/store.rs b/crates/vnidrop/src/targeted_transfer/store.rs index b455800..81bfcd5 100644 --- a/crates/vnidrop/src/targeted_transfer/store.rs +++ b/crates/vnidrop/src/targeted_transfer/store.rs @@ -226,7 +226,7 @@ impl TargetedTransferStore { Ok(()) } - pub(crate) async fn mark_sender_completed(&self, id: &str) -> Result<(), VnidropError> { + pub(crate) async fn mark_sender_completed(&self, id: &str) -> Result { let mut transaction = self.pool.begin().await.map_err(VnidropError::repository)?; let result = sqlx::query( r#" @@ -242,9 +242,26 @@ impl TargetedTransferStore { .await .map_err(VnidropError::repository)?; if result.rows_affected() == 0 { - return Err(VnidropError::InvalidTransition { - reason: "sender transfer cannot be completed".to_string(), - }); + let completed = sqlx::query( + "SELECT EXISTS(SELECT 1 FROM targeted_transfers WHERE id = ?1 AND role = 'sender' AND state = 'completed')", + ) + .bind(id) + .fetch_one(&mut *transaction) + .await + .map_err(VnidropError::repository)? + .get::(0) + != 0; + transaction + .commit() + .await + .map_err(VnidropError::repository)?; + return if completed { + Ok(false) + } else { + 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)", @@ -258,7 +275,7 @@ impl TargetedTransferStore { .commit() .await .map_err(VnidropError::repository)?; - Ok(()) + Ok(true) } pub(crate) async fn complete_receiver_and_enqueue( @@ -300,6 +317,28 @@ impl TargetedTransferStore { Ok(()) } + pub(crate) async fn advance_verified_bytes( + &self, + id: &str, + verified_bytes: u64, + ) -> Result { + let result = sqlx::query( + r#" + UPDATE targeted_transfers + SET verified_bytes = MIN(total_size, MAX(verified_bytes, ?2)), updated_at = ?3 + WHERE id = ?1 AND role = 'receiver' AND state = 'transferring' + AND ?2 > verified_bytes + "#, + ) + .bind(id) + .bind(verified_bytes as i64) + .bind(now_ms()) + .execute(&self.pool) + .await + .map_err(VnidropError::repository)?; + Ok(result.rows_affected() == 1) + } + pub(crate) async fn list_pending_completions( &self, ) -> Result, VnidropError> { @@ -417,14 +456,11 @@ impl TargetedTransferStore { r#" UPDATE targeted_transfers SET blob_ticket = NULL, - authorization_secret_handle = NULL, - verified_bytes = 0, - updated_at = ?2 + authorization_secret_handle = NULL WHERE id = ?1 "#, ) .bind(id) - .bind(now_ms()) .execute(&mut *transaction) .await .map_err(VnidropError::repository)?; @@ -435,10 +471,67 @@ impl TargetedTransferStore { Ok(()) } + /// Commit a local terminal denial and optional secret cleanup atomically. + pub(crate) async fn transition_terminal( + &self, + id: &str, + state: TargetedTransferState, + clear_authorization: bool, + ) -> Result { + let state = match state { + TargetedTransferState::Cancelled => "cancelled", + TargetedTransferState::Deleted => "deleted", + _ => { + return Err(VnidropError::invalid_input(anyhow::anyhow!( + "terminal transition requires cancelled or deleted" + ))) + } + }; + let mut transaction = self.pool.begin().await.map_err(VnidropError::repository)?; + if clear_authorization { + 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)?; + } + let result = sqlx::query( + r#" + UPDATE targeted_transfers + SET state = ?2, + blob_ticket = CASE WHEN ?3 THEN NULL ELSE blob_ticket END, + authorization_secret_handle = CASE WHEN ?3 THEN NULL ELSE authorization_secret_handle END, + updated_at = ?4 + WHERE id = ?1 AND state != ?2 + AND ( + (?2 = 'deleted' AND state != 'deleted') + OR (?2 = 'cancelled' AND state NOT IN ('completed', 'declined', 'cancelled', 'failed', 'deleted')) + ) + "#, + ) + .bind(id) + .bind(state) + .bind(clear_authorization) + .bind(now_ms()) + .execute(&mut *transaction) + .await + .map_err(VnidropError::repository)?; + transaction + .commit() + .await + .map_err(VnidropError::repository)?; + Ok(result.rows_affected() == 1) + } + pub(crate) async fn get(&self, id: &str) -> Result, VnidropError> { let row = sqlx::query( r#" - SELECT id, sender_endpoint_id, receiver_endpoint_id, manifest_id, + SELECT id, sender_endpoint_id, receiver_endpoint_id, manifest_id, transfer_name, file_count, total_size, verified_bytes, state, created_at, updated_at FROM targeted_transfers WHERE id = ?1 "#, @@ -473,7 +566,7 @@ impl TargetedTransferStore { pub(crate) async fn list(&self) -> Result, VnidropError> { let rows = sqlx::query( r#" - SELECT id, sender_endpoint_id, receiver_endpoint_id, manifest_id, + SELECT id, sender_endpoint_id, receiver_endpoint_id, manifest_id, transfer_name, file_count, total_size, verified_bytes, state, created_at, updated_at FROM targeted_transfers ORDER BY updated_at DESC @@ -505,22 +598,26 @@ impl TargetedTransferStore { rows.into_iter().map(row_to_full).collect() } - pub(crate) async fn cancel_by_peer(&self, peer_endpoint_id: &str) -> Result { + pub(crate) async fn cancel_by_peer( + &self, + peer_endpoint_id: &str, + ) -> Result, VnidropError> { let now = now_ms(); - let result = sqlx::query( + let rows = sqlx::query( r#" UPDATE targeted_transfers SET state = 'cancelled', updated_at = ?2 WHERE (sender_endpoint_id = ?1 OR receiver_endpoint_id = ?1) AND state NOT IN ('completed', 'declined', 'cancelled', 'failed', 'deleted') + RETURNING id "#, ) .bind(peer_endpoint_id) .bind(now) - .execute(&self.pool) + .fetch_all(&self.pool) .await .map_err(VnidropError::repository)?; - Ok(result.rows_affected()) + Ok(rows.into_iter().map(|row| row.get("id")).collect()) } pub(crate) async fn protocol_ids_for_peer( @@ -543,6 +640,23 @@ impl TargetedTransferStore { .collect()) } + pub(crate) async fn ids_for_peer( + &self, + peer_endpoint_id: &str, + ) -> Result, VnidropError> { + let rows = sqlx::query( + r#" + SELECT id FROM targeted_transfers + WHERE sender_endpoint_id = ?1 OR receiver_endpoint_id = ?1 + "#, + ) + .bind(peer_endpoint_id) + .fetch_all(&self.pool) + .await + .map_err(VnidropError::repository)?; + Ok(rows.into_iter().map(|row| row.get("id")).collect()) + } + pub(crate) async fn sender_payloads_for_peer( &self, peer_endpoint_id: &str, @@ -569,20 +683,21 @@ impl TargetedTransferStore { .collect()) } - pub(crate) async fn mark_interrupted_in_flight(&self) -> Result { + pub(crate) async fn mark_interrupted_in_flight(&self) -> Result, VnidropError> { let now = now_ms(); - let result = sqlx::query( + let rows = sqlx::query( r#" UPDATE targeted_transfers SET state = 'interrupted', updated_at = ?1 WHERE state IN ('connecting', 'transferring') + RETURNING id "#, ) .bind(now) - .execute(&self.pool) + .fetch_all(&self.pool) .await .map_err(VnidropError::repository)?; - Ok(result.rows_affected()) + Ok(rows.into_iter().map(|row| row.get("id")).collect()) } } @@ -618,6 +733,7 @@ fn row_to_transfer(row: sqlx::sqlite::SqliteRow) -> Result("file_count") as u64, total_size: row.get::("total_size") as u64, verified_bytes: row.try_get::("verified_bytes").unwrap_or(0) as u64, diff --git a/crates/vnidrop/src/tests.rs b/crates/vnidrop/src/tests.rs index 42acc9e..971747b 100644 --- a/crates/vnidrop/src/tests.rs +++ b/crates/vnidrop/src/tests.rs @@ -51,6 +51,8 @@ mod secure_secret_tests; #[cfg(target_os = "windows")] #[path = "tests/secure_secret_windows.rs"] mod secure_secret_windows_tests; +#[path = "tests/targeted_lifecycle.rs"] +mod targeted_lifecycle_tests; #[path = "tests/targeted_transfer.rs"] mod targeted_transfer_tests; #[path = "tests/ticket.rs"] diff --git a/crates/vnidrop/src/tests/api_surface.rs b/crates/vnidrop/src/tests/api_surface.rs index 6f18867..ff59ba9 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, 2); + assert_eq!(caps.targeted_transfer_protocol_version, 3); } diff --git a/crates/vnidrop/src/tests/targeted_lifecycle.rs b/crates/vnidrop/src/tests/targeted_lifecycle.rs new file mode 100644 index 0000000..f2358fa --- /dev/null +++ b/crates/vnidrop/src/tests/targeted_lifecycle.rs @@ -0,0 +1,462 @@ +use std::{ + path::Path, + str::FromStr, + sync::{mpsc, Arc, Mutex}, + time::{Duration, Instant}, +}; + +use iroh::{endpoint::presets, Endpoint}; +use iroh_blobs::{get::request::get_hash_seq_and_sizes, ticket::BlobTicket}; + +use crate::{ + secure_secret::FaultInjectingSecretStore, CoreEvent, CoreEventSink, DeviceRelationshipState, + PendingTargetedOffer, ShareMetadataInput, ShareSource, SourceKind, TargetedTransferState, + TransferAccessMode, VnidropCore, +}; + +struct RecordingSink { + events: Mutex>, +} + +impl CoreEventSink for RecordingSink { + fn on_event(&self, event: CoreEvent) { + self.events.lock().unwrap().push(event); + } +} + +struct EventGateSink { + kind: &'static str, + observed: mpsc::SyncSender, + release: Mutex>, +} + +impl CoreEventSink for EventGateSink { + fn on_event(&self, event: CoreEvent) { + if event.phase == "targeted_transfer" && event.kind == self.kind { + self.observed.send(event).unwrap(); + self.release.lock().unwrap().recv().unwrap(); + } + } +} + +struct ProtectedNode { + data_dir: tempfile::TempDir, + secret_store: Arc, + core: Option>, +} + +impl ProtectedNode { + fn new() -> Self { + Self::with_sink(Arc::new(RecordingSink { + events: Mutex::new(Vec::new()), + })) + } + + fn with_sink(sink: Arc) -> Self { + let data_dir = tempfile::tempdir().unwrap(); + let secret_store = Arc::new(FaultInjectingSecretStore::default()); + let core = VnidropCore::initialize_with_test_secret_store( + data_dir.path().to_string_lossy().into_owned(), + sink, + secret_store.clone(), + ) + .unwrap(); + Self { + data_dir, + secret_store, + core: Some(core), + } + } + + fn core(&self) -> Arc { + self.core.as_ref().unwrap().clone() + } + + fn restart(mut self) -> Self { + self.core.take().unwrap().shutdown(); + let core = VnidropCore::initialize_with_test_secret_store( + self.data_dir.path().to_string_lossy().into_owned(), + Arc::new(RecordingSink { + events: Mutex::new(Vec::new()), + }), + self.secret_store.clone(), + ) + .unwrap(); + self.core = Some(core); + self + } +} + +impl Drop for ProtectedNode { + fn drop(&mut self) { + if let Some(core) = self.core.take() { + core.shutdown(); + } + } +} + +fn source(path: &Path) -> ShareSource { + ShareSource { + kind: SourceKind::Path, + value: path.to_string_lossy().into_owned(), + display_name: Some(path.file_name().unwrap().to_string_lossy().into_owned()), + is_directory: false, + } +} + +fn wait_for_request(core: &VnidropCore, transfer_id: u64) -> crate::ReceiverRequest { + let started = Instant::now(); + loop { + if let Some(request) = core + .list_receiver_requests(transfer_id) + .unwrap() + .into_iter() + .find(|request| request.status == "requested") + { + return request; + } + assert!(started.elapsed() < Duration::from_secs(15)); + std::thread::sleep(Duration::from_millis(25)); + } +} + +fn establish_saved(sender: &ProtectedNode, receiver: &ProtectedNode, transfer_id: u64) { + let source_dir = tempfile::tempdir().unwrap(); + let output_dir = tempfile::tempdir().unwrap(); + let path = source_dir.path().join("pair.txt"); + std::fs::write(&path, b"pair").unwrap(); + let share = sender + .core() + .share_files( + vec![source(&path)], + ShareMetadataInput { + transfer_id, + transfer_name: Some("pair.txt".to_string()), + sender_name: Some("sender".to_string()), + access_mode: TransferAccessMode::ApprovalRequired, + }, + ) + .unwrap(); + let receiver_core = receiver.core(); + let output = output_dir.path().to_string_lossy().into_owned(); + let receive = std::thread::spawn(move || { + receiver_core.receive(share.ticket, output, Some("receiver".to_string())) + }); + let request = wait_for_request(&sender.core(), transfer_id); + sender + .core() + .respond_receiver_request(request.id, true, None) + .unwrap(); + receive.join().unwrap().unwrap(); + + let sender_id = sender.core().status().endpoint_id; + let receiver_id = receiver.core().status().endpoint_id; + let started = Instant::now(); + while !sender + .core() + .list_pairing_eligibilities() + .unwrap() + .iter() + .any(|entry| entry.peer_endpoint_id == receiver_id) + { + assert!(started.elapsed() < Duration::from_secs(10)); + std::thread::sleep(Duration::from_millis(25)); + } + assert!(sender + .core() + .request_saved_device_pairing(receiver_id.clone()) + .unwrap()); + let started = Instant::now(); + while !receiver + .core() + .list_device_relationships() + .unwrap() + .iter() + .any(|row| { + row.remote_endpoint_id == sender_id + && row.state == DeviceRelationshipState::PendingIncoming + }) + { + assert!(started.elapsed() < Duration::from_secs(15)); + std::thread::sleep(Duration::from_millis(25)); + } + receiver + .core() + .respond_to_device_pairing(sender_id.clone(), true) + .unwrap(); + for (node, peer) in [(sender, receiver_id), (receiver, sender_id)] { + let started = Instant::now(); + while !node + .core() + .list_saved_devices() + .unwrap() + .iter() + .any(|device| device.endpoint_id == peer) + { + assert!(started.elapsed() < Duration::from_secs(15)); + std::thread::sleep(Duration::from_millis(25)); + } + } +} + +fn wait_for_offer(core: &VnidropCore) -> PendingTargetedOffer { + let started = Instant::now(); + loop { + if let Some(offer) = core.list_pending_targeted_offers().into_iter().next() { + return offer; + } + assert!(started.elapsed() < Duration::from_secs(20)); + std::thread::sleep(Duration::from_millis(25)); + } +} + +fn approve( + sender: &ProtectedNode, + receiver: &ProtectedNode, + name: &str, + payload: &[u8], +) -> crate::TargetedTransfer { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join(name); + std::fs::write(&path, payload).unwrap(); + let receiver_core = receiver.core(); + let accept = std::thread::spawn(move || { + let offer = wait_for_offer(&receiver_core); + receiver_core.respond_to_targeted_offer(offer.transfer_id, true) + }); + let transfer = sender + .core() + .create_targeted_transfer( + receiver.core().status().endpoint_id, + vec![source(&path)], + Some(name.to_string()), + ) + .unwrap(); + accept.join().unwrap().unwrap(); + transfer +} + +#[test] +fn snapshot_name_round_trips_through_create_get_list_and_restart() { + let sender = ProtectedNode::new(); + let receiver = ProtectedNode::new(); + establish_saved(&sender, &receiver, 21_001); + let transfer = approve(&sender, &receiver, "Quarterly report.pdf", b"report"); + assert_eq!(transfer.transfer_name, "Quarterly report.pdf"); + assert_eq!( + sender + .core() + .get_targeted_transfer(transfer.id.clone()) + .unwrap() + .unwrap() + .transfer_name, + "Quarterly report.pdf" + ); + assert_eq!( + sender + .core() + .list_targeted_transfers() + .unwrap() + .into_iter() + .find(|row| row.id == transfer.id) + .unwrap() + .transfer_name, + "Quarterly report.pdf" + ); + let sender = sender.restart(); + assert_eq!( + sender + .core() + .get_targeted_transfer(transfer.id) + .unwrap() + .unwrap() + .transfer_name, + "Quarterly report.pdf" + ); +} + +#[test] +fn accepted_event_is_emitted_only_after_receiver_snapshot_is_durable() { + let (observed_tx, observed_rx) = mpsc::sync_channel(1); + let (release_tx, release_rx) = mpsc::sync_channel(1); + let sender = ProtectedNode::new(); + let receiver = ProtectedNode::with_sink(Arc::new(EventGateSink { + kind: "approved", + observed: observed_tx, + release: Mutex::new(release_rx), + })); + establish_saved(&sender, &receiver, 21_002); + + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("event.txt"); + std::fs::write(&path, b"event").unwrap(); + let receiver_core = receiver.core(); + let accept = std::thread::spawn(move || { + let offer = wait_for_offer(&receiver_core); + receiver_core.respond_to_targeted_offer(offer.transfer_id, true) + }); + let sender_core = sender.core(); + let receiver_id = receiver.core().status().endpoint_id; + let create = std::thread::spawn(move || { + sender_core.create_targeted_transfer( + receiver_id, + vec![source(&path)], + Some("event.txt".to_string()), + ) + }); + let event = observed_rx.recv_timeout(Duration::from_secs(20)).unwrap(); + let id = serde_json::from_str::(&event.data_json).unwrap() + ["targeted_transfer_id"] + .as_str() + .unwrap() + .to_string(); + let durable = receiver.core().get_targeted_transfer(id.clone()).unwrap(); + release_tx.send(()).unwrap(); + accept.join().unwrap().unwrap(); + create.join().unwrap().unwrap(); + assert_eq!(durable.unwrap().state, TargetedTransferState::Approved); +} + +#[test] +fn terminal_events_have_ordered_revisions_and_no_later_progress() { + let sender = ProtectedNode::new(); + let receiver = ProtectedNode::new(); + establish_saved(&sender, &receiver, 21_003); + let transfer = approve(&sender, &receiver, "events.bin", &vec![7; 128 * 1024]); + let output = tempfile::tempdir().unwrap(); + receiver + .core() + .receive_targeted_transfer( + transfer.id.clone(), + output.path().to_string_lossy().into_owned(), + ) + .unwrap(); + let mut events = receiver + .core() + .list_events(None) + .unwrap() + .into_iter() + .filter(|event| { + event.phase == "targeted_transfer" && event.data_json.contains(&transfer.id) + }) + .collect::>(); + events.sort_by_key(|event| event.revision); + assert!(events + .windows(2) + .all(|pair| pair[0].revision < pair[1].revision)); + let completed = events + .iter() + .position(|event| event.kind == "completed") + .expect("completed wake-up"); + assert!(events[completed + 1..] + .iter() + .all(|event| event.kind != "progress")); + assert_eq!( + receiver + .core() + .get_targeted_transfer(transfer.id) + .unwrap() + .unwrap() + .state, + TargetedTransferState::Completed + ); +} + +#[test] +fn delete_preserves_verified_progress_and_is_idempotent() { + let sender = ProtectedNode::new(); + let receiver = ProtectedNode::new(); + establish_saved(&sender, &receiver, 21_004); + let transfer = approve(&sender, &receiver, "delete.bin", b"verified payload"); + let output = tempfile::tempdir().unwrap(); + receiver + .core() + .receive_targeted_transfer( + transfer.id.clone(), + output.path().to_string_lossy().into_owned(), + ) + .unwrap(); + let completed = receiver + .core() + .get_targeted_transfer(transfer.id.clone()) + .unwrap() + .unwrap(); + receiver + .core() + .delete_targeted_transfer(transfer.id.clone()) + .unwrap(); + let deleted = receiver + .core() + .get_targeted_transfer(transfer.id.clone()) + .unwrap() + .unwrap(); + assert_eq!(deleted.verified_bytes, completed.verified_bytes); + receiver + .core() + .delete_targeted_transfer(transfer.id.clone()) + .unwrap(); + assert_eq!( + receiver + .core() + .get_targeted_transfer(transfer.id) + .unwrap() + .unwrap() + .updated_at, + deleted.updated_at + ); +} + +#[test] +fn restart_restores_bound_receiver_but_not_third_peer_access() { + let sender = ProtectedNode::new(); + let receiver = ProtectedNode::new(); + let stranger = ProtectedNode::new(); + establish_saved(&sender, &receiver, 21_005); + let transfer = approve(&sender, &receiver, "restart.txt", b"restart payload"); + let sender = sender.restart(); + let receiver = receiver.restart(); + + let stranger_result = stranger.core().receive_targeted_transfer( + transfer.id.clone(), + tempfile::tempdir() + .unwrap() + .path() + .to_string_lossy() + .into_owned(), + ); + assert!(stranger_result.is_err()); + let (_, leaked_ticket) = sender + .core() + .targeted_blob_ticket_for_test(transfer.id.clone()) + .unwrap(); + let leaked_ticket = BlobTicket::from_str(&leaked_ticket).unwrap(); + tokio::runtime::Builder::new_multi_thread() + .enable_all() + .build() + .unwrap() + .block_on(async move { + let attacker = Endpoint::builder(presets::Minimal).bind().await.unwrap(); + if let Ok(connection) = attacker + .connect(leaked_ticket.addr().clone(), iroh_blobs::ALPN) + .await + { + assert!(get_hash_seq_and_sizes( + &connection, + &leaked_ticket.hash_and_format().hash, + 1024 * 1024 * 32, + None, + ) + .await + .is_err()); + } + }); + let output = tempfile::tempdir().unwrap(); + receiver + .core() + .resume_targeted_transfer(transfer.id, output.path().to_string_lossy().into_owned()) + .unwrap(); + assert_eq!( + std::fs::read(output.path().join("restart.txt")).unwrap(), + b"restart payload" + ); +} diff --git a/crates/vnidrop/src/tests/targeted_transfer.rs b/crates/vnidrop/src/tests/targeted_transfer.rs index d94decc..b887a67 100644 --- a/crates/vnidrop/src/tests/targeted_transfer.rs +++ b/crates/vnidrop/src/tests/targeted_transfer.rs @@ -9,10 +9,11 @@ use iroh::{endpoint::presets, Endpoint}; use iroh_blobs::{get::request::get_hash_seq_and_sizes, ticket::BlobTicket}; use crate::{ - secure_secret::FaultInjectingSecretStore, CoreEvent, CoreEventSink, CoreNetworkConfig, - CoreRelayMode, DeviceRelationshipState, PendingTargetedOffer, PublishedOutput, - ReceiveOutputSink, ReceiveOutputSinkV2, ReceivedLocatorKind, ShareMetadataInput, ShareSource, - SourceKind, TargetedTransferState, TransferAccessMode, VnidropCore, VnidropError, + secure_secret::{FaultInjectingSecretStore, ReferenceStoreFailure}, + CoreEvent, CoreEventSink, CoreNetworkConfig, CoreRelayMode, DeviceRelationshipState, + PendingTargetedOffer, PublishedOutput, ReceiveOutputSink, ReceiveOutputSinkV2, + ReceivedLocatorKind, ShareMetadataInput, ShareSource, SourceKind, TargetedTransferState, + TransferAccessMode, VnidropCore, VnidropError, }; struct RecordingSink { @@ -30,6 +31,7 @@ struct ProtectedNode { secret_store: Arc, network_config: CoreNetworkConfig, core: Option>, + sink: Arc, } impl ProtectedNode { @@ -45,7 +47,7 @@ impl ProtectedNode { let store = Arc::new(FaultInjectingSecretStore::default()); let core = VnidropCore::initialize_with_test_secret_store_and_network( data_dir.path().to_string_lossy().into_owned(), - sink, + sink.clone(), store.clone(), network_config.clone(), ) @@ -55,6 +57,7 @@ impl ProtectedNode { secret_store: store, network_config, core: Some(core), + sink, } } @@ -71,12 +74,13 @@ impl ProtectedNode { }); let core = VnidropCore::initialize_with_test_secret_store_and_network( self.data_dir.path().to_string_lossy().into_owned(), - sink, + sink.clone(), self.secret_store.clone(), self.network_config.clone(), ) .expect("restarted protected test core"); self.core = Some(core); + self.sink = sink; self } } @@ -309,6 +313,7 @@ fn create_targeted_transfer_is_immutable_and_saved_only() { assert_eq!(transfer.receiver_endpoint_id, bob_id); assert_eq!(transfer.file_count, 1); assert_eq!(transfer.total_size, b"immutable payload".len() as u64); + assert_eq!(transfer.transfer_name, "payload.txt"); assert!(!transfer.id.is_empty()); assert!(!transfer.manifest_id.is_empty()); assert!(matches!( @@ -328,6 +333,7 @@ fn create_targeted_transfer_is_immutable_and_saved_only() { assert_eq!(listed.sender_endpoint_id, transfer.sender_endpoint_id); assert_eq!(listed.receiver_endpoint_id, transfer.receiver_endpoint_id); assert_eq!(listed.manifest_id, transfer.manifest_id); + assert_eq!(listed.transfer_name, transfer.transfer_name); assert_eq!(listed.file_count, transfer.file_count); assert_eq!(listed.total_size, transfer.total_size); } @@ -905,7 +911,20 @@ fn targeted_path_receive_preserves_no_overwrite_and_resumes_elsewhere() { .state, TargetedTransferState::Interrupted ); + let interrupted = bob + .core() + .get_targeted_transfer(transfer.id.clone()) + .unwrap() + .unwrap(); + assert_eq!(interrupted.verified_bytes, interrupted.total_size); + let bob = bob.restart(); + let restored = bob + .core() + .get_targeted_transfer(transfer.id.clone()) + .unwrap() + .unwrap(); + assert_eq!(restored.verified_bytes, interrupted.verified_bytes); let clean = tempfile::tempdir().unwrap(); bob.core() .resume_targeted_transfer(transfer.id, clean.path().to_string_lossy().into_owned()) @@ -1008,6 +1027,52 @@ fn delete_removes_authorization_and_resumable_state() { assert_eq!(sender_deleted.state, TargetedTransferState::Deleted); } +#[test] +fn delete_keeps_durable_denial_and_retries_secure_secret_cleanup() { + let alice = ProtectedNode::new(); + let bob = ProtectedNode::new(); + establish_saved(&alice, &bob, 11_051); + let transfer = approve_one(&alice, &bob, b"delete retry", "payload.txt"); + + bob.secret_store + .fail_with(Some(ReferenceStoreFailure::Unavailable)); + assert!(bob + .core() + .delete_targeted_transfer(transfer.id.clone()) + .is_err()); + let denied = bob + .core() + .get_targeted_transfer(transfer.id.clone()) + .unwrap() + .unwrap(); + assert_eq!(denied.state, TargetedTransferState::Deleted); + + bob.secret_store.fail_with(None); + bob.core() + .delete_targeted_transfer(transfer.id.clone()) + .unwrap(); + let retried = bob + .core() + .get_targeted_transfer(transfer.id.clone()) + .unwrap() + .unwrap(); + assert_eq!( + retried, denied, + "private cleanup must not mutate the public snapshot" + ); + assert!(bob + .core() + .resume_targeted_transfer( + transfer.id, + tempfile::tempdir() + .unwrap() + .path() + .to_string_lossy() + .into_owned(), + ) + .is_err()); +} + #[test] fn sender_delete_revokes_receiver_bound_payload_access() { let alice = ProtectedNode::new(); @@ -1043,18 +1108,37 @@ fn concurrent_independent_transfers_between_same_devices_are_isolated() { establish_saved(&alice, &bob, 11_060); // Design allows only one unresolved offer per sender; approve sequentially, - // then prove independent approved transfers do not corrupt each other. + // then overlap the independent pulls. let first = approve_one(&alice, &bob, b"alpha", "one.txt"); let second = approve_one(&alice, &bob, b"beta-payload", "two.txt"); assert_ne!(first.id, second.id); - alice - .core() + let first_sink = Arc::new(GatedSink::default()); + let second_sink = Arc::new(GatedSink::default()); + let first_core = bob.core(); + let first_id = first.id.clone(); + let first_thread_sink = first_sink.clone(); + let first_receive = std::thread::spawn(move || { + first_core.receive_targeted_transfer_with_output_sink(first_id, first_thread_sink) + }); + let second_core = bob.core(); + let second_id = second.id.clone(); + let second_thread_sink = second_sink.clone(); + let second_receive = std::thread::spawn(move || { + second_core.receive_targeted_transfer_with_output_sink(second_id, second_thread_sink) + }); + first_sink.wait_until_entered(); + second_sink.wait_until_entered(); + + bob.core() .cancel_targeted_transfer(first.id.clone()) .unwrap(); + first_sink.release(); + second_sink.release(); + assert!(first_receive.join().unwrap().is_err()); + second_receive.join().unwrap().unwrap(); assert_eq!( - alice - .core() + bob.core() .get_targeted_transfer(first.id.clone()) .unwrap() .unwrap() @@ -1062,35 +1146,23 @@ fn concurrent_independent_transfers_between_same_devices_are_isolated() { TargetedTransferState::Cancelled ); assert_eq!( - alice - .core() + bob.core() .get_targeted_transfer(second.id.clone()) .unwrap() .unwrap() .state, - TargetedTransferState::Approved + TargetedTransferState::Completed ); - - let output = tempfile::tempdir().unwrap(); - bob.core() - .receive_targeted_transfer( - second.id.clone(), - output.path().to_string_lossy().into_owned(), - ) - .unwrap(); assert_eq!( - std::fs::read(output.path().join("two.txt")).unwrap(), - b"beta-payload" + first_sink.aborts.load(std::sync::atomic::Ordering::SeqCst), + 1 + ); + assert_eq!( + second_sink + .finishes + .load(std::sync::atomic::Ordering::SeqCst), + 1 ); - - let cancelled_output = tempfile::tempdir().unwrap(); - assert!(bob - .core() - .receive_targeted_transfer( - first.id, - cancelled_output.path().to_string_lossy().into_owned(), - ) - .is_err()); } fn complete_targeted_roundtrip(alice: &ProtectedNode, bob: &ProtectedNode, transfer_id: u64) { @@ -1681,6 +1753,71 @@ fn targeted_cancel_aborts_each_sink_exactly_once() { } } +#[test] +fn forgetting_saved_sender_aborts_active_targeted_receive() { + let alice = ProtectedNode::new(); + let bob = ProtectedNode::new(); + establish_saved(&alice, &bob, 11_079); + let transfer = approve_one(&alice, &bob, b"forget during receive", "sink.txt"); + let sink = Arc::new(GatedSink::default()); + let bob_core = bob.core(); + let transfer_id = transfer.id.clone(); + let receive_sink = sink.clone(); + let receive = std::thread::spawn(move || { + bob_core.receive_targeted_transfer_with_output_sink(transfer_id, receive_sink) + }); + sink.wait_until_entered(); + + bob.core() + .forget_saved_device(alice.core().status().endpoint_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); + assert_eq!( + bob.core() + .get_targeted_transfer(transfer.id) + .unwrap() + .unwrap() + .state, + TargetedTransferState::Cancelled + ); +} + +#[test] +fn sender_cancel_stops_online_receiver_before_publish() { + let alice = ProtectedNode::new(); + let bob = ProtectedNode::new(); + establish_saved(&alice, &bob, 11_080); + let transfer = approve_one(&alice, &bob, b"sender cancel", "sink.txt"); + let sink = Arc::new(GatedSink::default()); + let bob_core = bob.core(); + let transfer_id = transfer.id.clone(); + let receive_sink = sink.clone(); + let receive = std::thread::spawn(move || { + bob_core.receive_targeted_transfer_with_output_sink(transfer_id, receive_sink) + }); + sink.wait_until_entered(); + + alice + .core() + .cancel_targeted_transfer(transfer.id.clone()) + .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); + assert_eq!( + bob.core() + .get_targeted_transfer(transfer.id) + .unwrap() + .unwrap() + .state, + TargetedTransferState::Cancelled + ); +} + #[test] fn decline_returns_typed_declined_outcome() { let alice = ProtectedNode::new(); @@ -1713,3 +1850,45 @@ fn decline_returns_typed_declined_outcome() { crate::TargetedOfferResponse::Declined ); } + +#[test] +fn approval_secret_failure_records_failed_receiver_snapshot_and_wakeup() { + let alice = ProtectedNode::new(); + let bob = ProtectedNode::new(); + establish_saved(&alice, &bob, 11_081); + let source_dir = tempfile::tempdir().unwrap(); + let source_path = source_dir.path().join("secret-failure.txt"); + std::fs::write(&source_path, b"secret failure").unwrap(); + let sender = alice.core(); + let receiver_id = bob.core().status().endpoint_id; + let create = std::thread::spawn(move || { + sender.create_targeted_transfer( + receiver_id, + vec![targeted_source(&source_path)], + Some("secret-failure.txt".to_string()), + ) + }); + let offer = wait_for_pending_offer(&bob.core()); + bob.secret_store + .fail_with(Some(ReferenceStoreFailure::Unavailable)); + assert!(bob + .core() + .respond_to_targeted_offer(offer.transfer_id.clone(), true) + .is_err()); + bob.secret_store.fail_with(None); + let failed = bob + .core() + .get_targeted_transfer(offer.transfer_id.clone()) + .unwrap() + .unwrap(); + assert_eq!(failed.state, TargetedTransferState::Failed); + assert!(bob + .core() + .list_events(None) + .unwrap() + .iter() + .any(|event| event.phase == "targeted_transfer" + && event.kind == "failed" + && event.data_json.contains(&offer.transfer_id))); + assert!(create.join().unwrap().is_err()); +} diff --git a/crates/vnidrop/tests/experimental_domain.rs b/crates/vnidrop/tests/experimental_domain.rs index 4148bdb..56d7b81 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: 2, + targeted_transfer_protocol_version: 3, } ); } @@ -41,6 +41,7 @@ fn saved_devices_relationships_and_targeted_transfers_are_distinct_contracts() { sender_endpoint_id: "sender-endpoint".to_string(), receiver_endpoint_id: device.endpoint_id.clone(), manifest_id: "immutable-manifest-id".to_string(), + transfer_name: "Holiday photos".to_string(), file_count: 2, total_size: 42, verified_bytes: 0, diff --git a/shared/docs/saved-devices-ui-contract.md b/shared/docs/saved-devices-ui-contract.md index 112a03f..19e6e44 100644 --- a/shared/docs/saved-devices-ui-contract.md +++ b/shared/docs/saved-devices-ui-contract.md @@ -21,7 +21,10 @@ Use **saved device**, **device relationship**, **targeted transfer**, **invitati Relevant kinds: - **pairing:** `eligibility-available`, `eligibility-removed`, `relationship-changed`, `relationship-grant-rotated`, `saved-device-forgotten`, `device-blocked` -- **targeted_transfer:** `offer-received`, `offer-accepted`, `offer-declined` +- **targeted_transfer:** `offer-received`, `approved`, `offer-declined`, `created`, + `offering`, `awaiting-approval`, `connecting`, `transferring`, `progress`, + `interrupted`, `completed`, `cancelled`, `failed`, `deleted`. Durable-row + wake-ups include `targeted_transfer_id`; refresh list/get after each event. ## Pairing