cdricms/vnidrop
1
1
Fork 0
mirror of https://github.com/sudosylabs/vnidrop.git synced 2026-08-13 05:49:57 +02:00
Code Issues Packages Projects Releases Wiki Activity

feat(core): expose durable targeted transfer lifecycle

Browse Source
This commit is contained in:
Hammed Abass
2026-08-12 16:11:53 +02:00
parent 32d69b9771
commit 67a557af7c
17 changed files with 1321 additions and 197 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
crates/vnidrop/CORE_FLOW.md
View File

@@ -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

3
crates/vnidrop/src/api.rs
View File

@@ -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.

12
crates/vnidrop/src/runtime/facade.rs
View File

@@ -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))
}

75
crates/vnidrop/src/runtime/mod.rs
View File

@@ -112,12 +112,13 @@ pub(super) struct CoreInner {
pub(super) access_policy: Arc<AccessPolicy>,
/// 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<HashMap<u64, ActiveTransfer>>,
pub(super) active_transfers: Arc<std::sync::Mutex<HashMap<u64, ActiveTransfer>>>,
pub(super) active_targeted_transfers: Arc<std::sync::Mutex<HashMap<String, ActiveTransfer>>>,
// Active shares are protected by persistent Iroh tags.
pub(super) active_shares: TokioMutex<HashMap<u64, ()>>,
/// 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<HashMap<String, HashSet<u64>>>,
pub(super) hash_to_transfer: Arc<TokioMutex<HashMap<String, HashSet<u64>>>>,
pub(super) connection_endpoints: TokioMutex<HashMap<u64, String>>,
pub(super) provider_task: TokioMutex<Option<JoinHandle<()>>>,
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::<u64, ActiveTransfer>::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 {

116
crates/vnidrop/src/runtime/receive.rs
View File

@@ -184,7 +184,10 @@ impl CoreInner {
pub(super) async fn receive_targeted_payload(
self: &Arc<Self>,
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}"

315
crates/vnidrop/src/runtime/targeted.rs
View File

@@ -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 {

9
crates/vnidrop/src/targeted_transfer/inbox.rs
View File

@@ -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(_) => {

91
crates/vnidrop/src/targeted_transfer/protocol.rs
View File

@@ -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<RelayUrl>,
event_hub: std::sync::Arc<crate::event_hub::EventHub>,
access_policy: std::sync::Arc<crate::access_policy::AccessPolicy>,
cleanup:
std::sync::Arc<dyn Fn(super::TargetedTransferRow) -> TargetedCleanupFuture + Send + Sync>,
}
pub(crate) type TargetedCleanupFuture =
Pin<Box<dyn Future<Output = Result<(), VnidropError>> + 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<DeviceRelationshipService>,
inbox: TargetedOfferInbox,
@@ -60,6 +69,11 @@ impl TargetedTransferProtocol {
local_endpoint_id: String,
relay_mode: CoreRelayMode,
custom_relay_urls: Vec<RelayUrl>,
event_hub: std::sync::Arc<crate::event_hub::EventHub>,
access_policy: std::sync::Arc<crate::access_policy::AccessPolicy>,
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<TargetedTransferState>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]

10
crates/vnidrop/src/targeted_transfer/state.rs
View File

@@ -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 {

154
crates/vnidrop/src/targeted_transfer/store.rs
View File

@@ -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<bool, VnidropError> {
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::<i64, _>(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<bool, VnidropError> {
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<Vec<TargetedTransferRow>, 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<bool, VnidropError> {
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<Option<TargetedTransfer>, 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<Vec<TargetedTransfer>, 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<u64, VnidropError> {
pub(crate) async fn cancel_by_peer(
&self,
peer_endpoint_id: &str,
) -> Result<Vec<String>, 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<Vec<String>, 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<u64, VnidropError> {
pub(crate) async fn mark_interrupted_in_flight(&self) -> Result<Vec<String>, 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<TargetedTransfer, Vni
sender_endpoint_id: row.get("sender_endpoint_id"),
receiver_endpoint_id: row.get("receiver_endpoint_id"),
manifest_id: row.get("manifest_id"),
transfer_name: row.get("transfer_name"),
file_count: row.get::<i64, _>("file_count") as u64,
total_size: row.get::<i64, _>("total_size") as u64,
verified_bytes: row.try_get::<i64, _>("verified_bytes").unwrap_or(0) as u64,

2
crates/vnidrop/src/tests.rs
View File

@@ -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"]

2
crates/vnidrop/src/tests/api_surface.rs
View File

@@ -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);
}

462
crates/vnidrop/src/tests/targeted_lifecycle.rs Normal file
View File

@@ -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<Vec<CoreEvent>>,
}
impl CoreEventSink for RecordingSink {
fn on_event(&self, event: CoreEvent) {
self.events.lock().unwrap().push(event);
}
}
struct EventGateSink {
kind: &'static str,
observed: mpsc::SyncSender<CoreEvent>,
release: Mutex<mpsc::Receiver<()>>,
}
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<FaultInjectingSecretStore>,
core: Option<Arc<VnidropCore>>,
}
impl ProtectedNode {
fn new() -> Self {
Self::with_sink(Arc::new(RecordingSink {
events: Mutex::new(Vec::new()),
}))
}
fn with_sink(sink: Arc<dyn CoreEventSink>) -> 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<VnidropCore> {
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::<serde_json::Value>(&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::<Vec<_>>();
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"
);
}

245
crates/vnidrop/src/tests/targeted_transfer.rs
View File

@@ -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<FaultInjectingSecretStore>,
network_config: CoreNetworkConfig,
core: Option<Arc<VnidropCore>>,
sink: Arc<RecordingSink>,
}
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());
}

3
crates/vnidrop/tests/experimental_domain.rs
View File

@@ -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,