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

fix(core): harden targeted transfer recovery

Browse Source
This commit is contained in:
Hammed Abass
2026-08-12 18:48:13 +02:00
parent 4f09e474c5
commit 2ac9166b34
25 changed files with 3164 additions and 1509 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
crates/vnidrop/AGENTS.md
View File

@@ -55,7 +55,11 @@ src/
lifecycle.rs # cancel share, delete, status, access mode, shutdown
provider.rs # provider messages, per-peer transfer progress
saved_devices.rs # saved-device pairing, forget, block
targeted.rs # saved-device targeted transfers
targeted.rs # targeted lifecycle and restart restoration
targeted_create.rs # targeted import, offer, and sender approval
targeted_receive.rs # targeted download, resume, and completion
targeted_payload.rs # targeted blob fetch and export bridge
targeted_reconciliation.rs # durable delivery and cleanup retries
persistence.rs # AppDataStores / persistence open (domain stores)
invitation/ # invitation-transfer domain store (type name: Repository)
pairing_eligibility/ # eligibility service + store

3
crates/vnidrop/CORE_FLOW.md
View File

@@ -79,7 +79,8 @@ bytes through Kotlin memory.
### Pairing and targeted-transfer event catalog
Treat every event as a wake-up: refresh durable state via list/get APIs.
Mid-transfer progress polish may follow.
Targeted progress updates persist monotonic `verified_bytes`; event payloads are
advisory and the durable targeted-transfer snapshot is authoritative.
**`pairing`:** `eligibility-available`, `eligibility-removed`,
`relationship-changed`, `relationship-grant-rotated`, `saved-device-forgotten`,

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

@@ -186,6 +186,135 @@ impl VnidropCore {
.store(suppress, std::sync::atomic::Ordering::SeqCst);
}
pub(crate) fn suppress_targeted_authorization_delivery_for_test(&self, suppress: bool) {
self.inner
.suppress_targeted_authorization_delivery
.store(suppress, std::sync::atomic::Ordering::SeqCst);
}
pub(crate) fn accept_targeted_offer_without_waiting_for_test(
&self,
id: String,
) -> Result<(), VnidropError> {
self.block_on(async {
let offer = self
.inner
.targeted_offers
.pending_for_acceptance(&id)
.await
.ok_or_else(|| {
VnidropError::invalid_input(anyhow::anyhow!("unknown targeted offer"))
})?;
self.inner
.targeted_store()
.persist_accepted_offer_intent(&offer)
.await?;
self.inner
.targeted_offers
.accept_live(&id)
.await
.map_err(|_| {
VnidropError::device_unavailable(anyhow::anyhow!(
"sender disconnected before acceptance"
))
})
})
}
pub(crate) fn persist_block_without_cleanup_for_test(
&self,
endpoint_id: String,
) -> Result<(), VnidropError> {
self.block_on(async {
self.inner
.blocked_devices
.block_endpoint(&endpoint_id, crate::util::now_ms())
.await
.map_err(VnidropError::repository)
})
}
pub(crate) fn corrupt_targeted_content_hash_for_test(
&self,
id: String,
) -> Result<(), VnidropError> {
self.block_on(
self.inner
.targeted_store()
.corrupt_content_hash_for_test(&id),
)
}
pub(crate) fn create_orphaned_targeted_authorization_for_test(
&self,
) -> Result<(), VnidropError> {
self.block_on(async {
let custody = self.inner.secret_custody.as_ref().ok_or_else(|| {
VnidropError::SecureStorageUnavailable {
reason: "test custody unavailable".to_string(),
}
})?;
custody
.create_orphaned_targeted_authorization_for_test()
.await
})
}
pub(crate) fn targeted_authorization_handle_count_for_test(
&self,
) -> Result<usize, VnidropError> {
self.block_on(async {
let custody = self.inner.secret_custody.as_ref().ok_or_else(|| {
VnidropError::SecureStorageUnavailable {
reason: "test custody unavailable".to_string(),
}
})?;
Ok(custody
.list_active_handles(crate::secure_secret::SecretKind::TargetedAuthorization)
.await?
.len())
})
}
pub(crate) fn redeliver_targeted_authorization_for_test(
&self,
id: String,
) -> Result<bool, VnidropError> {
self.block_on(async {
let row = self
.inner
.targeted_store()
.get_row(&id)
.await?
.ok_or_else(|| VnidropError::invalid_input(anyhow::anyhow!("unknown transfer")))?;
self.inner.deliver_stored_targeted_authorization(&row).await
})
}
pub(crate) fn targeted_authorization_delivery_attempts_for_test(&self) -> u64 {
self.inner
.targeted_authorization_delivery_attempts
.load(std::sync::atomic::Ordering::SeqCst)
}
pub(crate) fn hold_all_transfer_slots_for_test(&self) -> tokio::sync::oneshot::Sender<()> {
let inner = self.inner.clone();
let permits = inner.limits.max_concurrent_transfers as u32;
let (ready_tx, ready_rx) = std::sync::mpsc::sync_channel(1);
let (release_tx, release_rx) = tokio::sync::oneshot::channel();
self.runtime.handle().spawn(async move {
let _permits = inner
.transfer_slots
.acquire_many(permits)
.await
.expect("transfer limiter open");
ready_tx.send(()).expect("slot holder ready");
let _ = release_rx.await;
});
ready_rx.recv().expect("slot holder started");
release_tx
}
pub(crate) fn targeted_payload_is_registered_for_test(
&self,
id: String,
@@ -639,7 +768,7 @@ impl VnidropCore {
))
}
/// Offline-only pending offers awaiting explicit local approval.
/// Ticket-free pending offers awaiting explicit local approval.
pub fn list_pending_targeted_offers(&self) -> Vec<crate::api::PendingTargetedOffer> {
self.block_on(self.inner.list_pending_targeted_offers())
}

2
crates/vnidrop/src/runtime/lifecycle.rs
View File

@@ -191,7 +191,7 @@ impl CoreInner {
task.abort();
let _ = task.await;
}
if let Some(task) = self.targeted_completion_task.lock().await.take() {
if let Some(task) = self.targeted_reconciliation_task.lock().await.take() {
task.abort();
let _ = task.await;
}

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

@@ -18,6 +18,10 @@ mod saved_devices;
mod share;
mod storage;
mod targeted;
mod targeted_create;
mod targeted_payload;
mod targeted_receive;
mod targeted_reconciliation;
pub use facade::VnidropCore;
#[cfg(test)]
@@ -122,13 +126,17 @@ pub(super) struct CoreInner {
pub(super) provider_task: TokioMutex<Option<JoinHandle<()>>>,
pub(super) delivery_receipt_notify: Notify,
pub(super) delivery_receipt_task: TokioMutex<Option<JoinHandle<()>>>,
pub(super) targeted_completion_task: TokioMutex<Option<JoinHandle<()>>>,
pub(super) targeted_reconciliation_task: TokioMutex<Option<JoinHandle<()>>>,
pub(super) shutdown_started: AtomicBool,
/// Test-only log of peers passed to [`Self::cancel_targeted_transfers_for_peer`].
#[cfg(test)]
targeted_cancel_log: std::sync::Mutex<Vec<String>>,
#[cfg(test)]
suppress_targeted_completion: AtomicBool,
#[cfg(test)]
suppress_targeted_authorization_delivery: AtomicBool,
#[cfg(test)]
targeted_authorization_delivery_attempts: std::sync::atomic::AtomicU64,
}
pub(super) struct ActiveTransfer {
@@ -398,6 +406,15 @@ impl CoreInner {
limits.max_saved_devices,
limits.pairing_timeout_ms,
));
if let Some(custody) = &secret_custody {
let referenced = targeted_transfers.authorization_secret_handles().await?;
let removed = custody
.remove_orphaned_targeted_authorizations(&referenced)
.await?;
if removed > 0 {
tracing::warn!(removed, "removed orphaned targeted authorization secrets");
}
}
let active_transfers =
Arc::new(std::sync::Mutex::new(HashMap::<u64, ActiveTransfer>::new()));
let active_targeted_transfers = Arc::new(std::sync::Mutex::new(HashMap::<
@@ -449,6 +466,125 @@ impl CoreInner {
Ok(())
}) as crate::targeted_transfer::protocol::TargetedCleanupFuture
});
let authorization_custody = secret_custody.clone();
let authorization_store = targeted_transfers.clone();
let authorization_repository = repository.clone();
let authorization_events = event_hub.clone();
let persist_targeted_authorization = Arc::new(
move |authorization: crate::targeted_transfer::TargetedAuthorization| {
let custody = authorization_custody.clone();
let transfers = authorization_store.clone();
let repository = authorization_repository.clone();
let events = authorization_events.clone();
Box::pin(async move {
let custody = custody.ok_or_else(|| {
crate::error::VnidropError::SecureStorageUnavailable {
reason: "targeted authorization requires protected custody".to_string(),
}
})?;
if let Some(row) = transfers.get_row(&authorization.transfer_id).await? {
let exact = row.role
== crate::targeted_transfer::TargetedTransferRole::Receiver
&& matches!(
row.state,
crate::api::TargetedTransferState::Approved
| crate::api::TargetedTransferState::Connecting
| crate::api::TargetedTransferState::Transferring
| crate::api::TargetedTransferState::Interrupted
| crate::api::TargetedTransferState::Completed
)
&& row.protocol_transfer_id == authorization.protocol_transfer_id
&& row.sender_endpoint_id == authorization.sender_endpoint_id
&& row.receiver_endpoint_id == authorization.receiver_endpoint_id
&& row.manifest_id == authorization.manifest_id
&& row.content_hash == authorization.content_hash
&& row.transfer_name == authorization.transfer_name
&& row.file_count == authorization.file_count
&& row.total_size == authorization.total_size
&& row.blob_ticket.as_deref()
== Some(authorization.blob_ticket.as_str());
let Some(handle) = row.authorization_secret_handle else {
return Err(crate::error::VnidropError::SecureStorageMissing {
reason: "receiver authorization handle is missing".to_string(),
});
};
if !exact {
return Err(crate::error::VnidropError::permission(anyhow::anyhow!(
"targeted authorization conflicts with receiver state"
)));
}
let material = custody
.load(&crate::secure_secret::SecretHandle::from_stored(handle))
.await?;
let rebuilt = crate::targeted_transfer::reconstruct_authorization(
crate::targeted_transfer::TargetedAuthorizationDraft {
transfer_id: row.id,
protocol_transfer_id: row.protocol_transfer_id,
sender_endpoint_id: row.sender_endpoint_id,
receiver_endpoint_id: row.receiver_endpoint_id,
manifest_id: row.manifest_id,
content_hash: row.content_hash,
file_count: row.file_count,
total_size: row.total_size,
protocol_version: authorization.protocol_version,
transfer_name: row.transfer_name,
blob_ticket: row.blob_ticket.expect("checked blob ticket"),
},
&material,
)?;
if rebuilt.encode()? != authorization.encode()? {
return Err(crate::error::VnidropError::permission(anyhow::anyhow!(
"protected receiver authorization does not match delivery"
)));
}
return Ok(false);
}
let invitation_collision = repository
.list_transfers()
.await
.map_err(crate::error::VnidropError::repository)?
.into_iter()
.any(|transfer| transfer.transfer_id == authorization.protocol_transfer_id);
if invitation_collision {
return Err(crate::error::VnidropError::invalid_input(anyhow::anyhow!(
"targeted transfer protocol id collides with invitation work"
)));
}
let handle = custody
.protect(
crate::secure_secret::SecretKind::TargetedAuthorization,
crate::targeted_transfer::auth_secret_material(&authorization)?,
None,
)
.await?;
let created = match transfers
.persist_receiver_authorization_and_consume_intent(
&authorization,
handle.as_str(),
)
.await
{
Ok(created) => created,
Err(error) => {
if let Err(cleanup_error) = custody.remove(&handle).await {
tracing::warn!(%cleanup_error, "failed to roll back receiver authorization secret");
}
return Err(error);
}
};
if created {
// Protocol callbacks are durable wake-ups, including restart recovery.
events.emit_endpoint(
"targeted_transfer",
"approved",
serde_json::json!({ "targeted_transfer_id": authorization.transfer_id }),
);
}
Ok(created)
})
as crate::targeted_transfer::protocol::TargetedAuthorizationPersistFuture
},
);
let router = Router::builder(endpoint.clone())
.accept(iroh_blobs::ALPN, blobs)
.accept(HandshakeService::ALPN, handshake)
@@ -469,6 +605,7 @@ impl CoreInner {
event_hub.clone(),
access_policy.clone(),
targeted_cleanup,
persist_targeted_authorization,
),
)
.spawn();
@@ -501,12 +638,16 @@ impl CoreInner {
provider_task: TokioMutex::new(None),
delivery_receipt_notify: Notify::new(),
delivery_receipt_task: TokioMutex::new(None),
targeted_completion_task: TokioMutex::new(None),
targeted_reconciliation_task: TokioMutex::new(None),
shutdown_started: AtomicBool::new(false),
#[cfg(test)]
targeted_cancel_log: std::sync::Mutex::new(Vec::new()),
#[cfg(test)]
suppress_targeted_completion: AtomicBool::new(false),
#[cfg(test)]
suppress_targeted_authorization_delivery: AtomicBool::new(false),
#[cfg(test)]
targeted_authorization_delivery_attempts: std::sync::atomic::AtomicU64::new(0),
});
// In-flight connecting/transferring transfers become Interrupted across restart.
@@ -536,7 +677,7 @@ impl CoreInner {
);
inner.spawn_provider_event_task(event_rx).await;
inner.spawn_delivery_receipt_task().await;
inner.spawn_targeted_completion_task().await;
inner.spawn_targeted_reconciliation_task().await;
if let Err(error) = inner.pairing_eligibility.reconcile().await {
tracing::warn!(%error, "failed to reconcile pairing eligibility");
}

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

@@ -182,196 +182,10 @@ impl CoreInner {
.await
}
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<()> {
let _permit = self
.transfer_slots
.acquire()
.await
.context("transfer limiter is closed")
.map_err(VnidropError::internal)?;
let sender_addr = filter_peer_addr_for_relay_mode(
blob_ticket.addr(),
self.relay_mode,
&self.custom_relay_urls,
)
.map_err(VnidropError::network)?;
blob_ticket = BlobTicket::new(sender_addr, blob_ticket.hash(), blob_ticket.format());
let (cancel, mut cancelled) = oneshot::channel();
self.active_targeted_transfers
.lock()
.expect("active_targeted_transfers")
.insert(
targeted_transfer_id.to_string(),
ActiveTransfer {
direction: TransferDirection::Receive,
cancel,
},
);
let result = tokio::select! {
biased;
_ = &mut cancelled => Err(VnidropError::cancelled("transfer cancelled").into()),
result = self.download_targeted_payload(
targeted_transfer_id,
transfer_id,
expected_file_count,
expected_payload_size,
blob_ticket,
target,
) => result,
};
self.active_targeted_transfers
.lock()
.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<()> {
if let ReceiveTarget::Directory(output_dir) = &target {
tokio::fs::create_dir_all(output_dir)
.await
.map_err(VnidropError::filesystem)?;
}
let connection = self
.endpoint
.connect(blob_ticket.addr().clone(), iroh_blobs::ALPN)
.await
.map_err(VnidropError::network)?;
let hash_and_format = blob_ticket.hash_and_format();
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 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 remote_size > self.limits.max_total_bytes {
anyhow::bail!(
"remote collection size {remote_size} exceeds limit {}",
self.limits.max_total_bytes
);
}
let download_tag = self.store.tags().temp_tag(hash_and_format).await?;
let get = self.store.remote().fetch(connection, hash_and_format);
let mut stream = get.stream();
loop {
let Some(item) = stream.next().await else {
anyhow::bail!("targeted download ended without completion");
};
match item {
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}"
))
.into());
}
}
}
let collection = Collection::load(hash_and_format.hash, self.store.as_ref()).await?;
self.export_collection_untracked(transfer_id, total_files, target, collection)
.await?;
drop(download_tag);
Ok(())
}
#[allow(
clippy::too_many_arguments,
reason = "targeted receive binds immutable authorization fields plus its cancel token"
)]
pub(super) async fn receive_to_target(
self: &Arc<Self>,
ticket: String,
@@ -783,7 +597,7 @@ impl CoreInner {
Ok(())
}
async fn export_collection_untracked(
pub(super) async fn export_collection_untracked(
&self,
transfer_id: u64,
total_files: u64,

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

File diff suppressed because it is too large Load Diff

400
crates/vnidrop/src/runtime/targeted_create.rs Normal file
View File

@@ -0,0 +1,400 @@
//! Sender-side targeted-transfer creation and offer approval.
use std::sync::Arc;
use iroh_blobs::{ticket::BlobTicket, BlobFormat};
use uuid::Uuid;
use super::{
targeted::{allocate_protocol_transfer_id, map_connect_failure},
targeted_tag_name, CoreInner,
};
use crate::{
api::{
saved_device_capabilities, ShareSource, TargetedTransfer, TargetedTransferState,
TransferAccessMode,
},
error::VnidropError,
targeted_transfer::{
protocol::{
map_offer_refuse_reason, SubmitTargetedOffer, TargetedTransferProtocol,
WireOfferResponse,
},
TargetedAuthorization, TargetedAuthorizationDraft, TargetedTransferRole,
TargetedTransferRow,
},
util::{non_empty, now_ms},
};
impl CoreInner {
pub(super) async fn create_targeted_transfer(
self: &Arc<Self>,
receiver_endpoint_id: String,
sources: Vec<ShareSource>,
transfer_name: Option<String>,
) -> Result<TargetedTransfer, VnidropError> {
self.device_relationships
.require_saved(&receiver_endpoint_id)
.await?;
let (transfer_uuid, protocol_transfer_id) = loop {
let transfer_uuid = Uuid::new_v4().to_string();
let protocol_transfer_id = allocate_protocol_transfer_id(&transfer_uuid);
let invitation_collision = self
.repository
.list_transfers()
.await
.map_err(VnidropError::repository)?
.into_iter()
.any(|transfer| transfer.transfer_id == protocol_transfer_id);
if !invitation_collision
&& !self
.targeted_store()
.contains_protocol_id(protocol_transfer_id)
.await?
{
break (transfer_uuid, protocol_transfer_id);
}
};
let sender_endpoint_id = self.endpoint.id().to_string();
let now = now_ms();
if sources.is_empty() {
return Err(VnidropError::invalid_input(anyhow::anyhow!(
"at least one source is required"
)));
}
if sources.len() as u64 > self.limits.max_sources {
return Err(VnidropError::invalid_input(anyhow::anyhow!(
"source count {} exceeds limit {}",
sources.len(),
self.limits.max_sources
)));
}
self.limits
.validate_metadata_text("transfer name", transfer_name.as_deref())
.map_err(VnidropError::invalid_input)?;
let import = self
.import_sources(protocol_transfer_id, sources)
.await
.map_err(VnidropError::transfer)?;
let payload_name = transfer_name
.and_then(non_empty)
.unwrap_or_else(|| import.default_name.clone());
let blob_ticket =
BlobTicket::new(self.endpoint.addr(), import.root_hash, BlobFormat::HashSeq);
self.store
.tags()
.set(
targeted_tag_name(&transfer_uuid),
(import.root_hash, BlobFormat::HashSeq),
)
.await
.map_err(VnidropError::transfer)?;
self.register_share_hashes(
protocol_transfer_id,
std::iter::once(import.root_hash).chain(import.member_hashes.iter().copied()),
)
.await;
self.access_policy
.set_mode(protocol_transfer_id, TransferAccessMode::ApprovalRequired)
.await;
drop(import.tag);
let store = self.targeted_store();
let row = TargetedTransferRow {
id: transfer_uuid.clone(),
protocol_transfer_id,
sender_endpoint_id: sender_endpoint_id.clone(),
receiver_endpoint_id: receiver_endpoint_id.clone(),
manifest_id: blob_ticket.hash().to_string(),
content_hash: blob_ticket.hash().to_string(),
transfer_name: payload_name.clone(),
file_count: import.file_count,
total_size: import.total_size,
verified_bytes: 0,
blob_ticket: None,
authorization_secret_handle: None,
role: TargetedTransferRole::Sender,
state: TargetedTransferState::Preparing,
created_at: now,
updated_at: now,
};
if let Err(error) = store.insert(&row).await {
self.teardown_targeted_payload(protocol_transfer_id, Some(&transfer_uuid))
.await;
return Err(error);
}
self.emit_targeted_lifecycle(&transfer_uuid, "created");
if let Err(error) = store
.set_state(
&transfer_uuid,
TargetedTransferState::Preparing,
TargetedTransferState::Offering,
)
.await
{
self.teardown_targeted_payload(protocol_transfer_id, Some(&transfer_uuid))
.await;
return Err(error);
}
self.emit_targeted_lifecycle(&transfer_uuid, "offering");
let addr = match self
.device_relationships
.peer_addr(&receiver_endpoint_id)
.await
{
Ok(addr) => addr,
Err(error) => {
if store
.set_state(
&transfer_uuid,
TargetedTransferState::Offering,
TargetedTransferState::Failed,
)
.await
.is_ok()
{
self.emit_targeted_lifecycle(&transfer_uuid, "failed");
}
self.teardown_targeted_payload(protocol_transfer_id, Some(&transfer_uuid))
.await;
return Err(error);
}
};
let client = TargetedTransferProtocol::client(self.endpoint.clone(), addr);
let challenge =
match tokio::time::timeout(self.connection_timeout(), client.request_challenge()).await
{
Ok(Ok(challenge)) => challenge,
Ok(Err(error)) => {
if store
.set_state(
&transfer_uuid,
TargetedTransferState::Offering,
TargetedTransferState::Failed,
)
.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(_) => {
if store
.set_state(
&transfer_uuid,
TargetedTransferState::Offering,
TargetedTransferState::Failed,
)
.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!(
"device did not answer in time"
)));
}
};
let (proof, generation, relationship_protocol_version) = match self
.device_relationships
.prove_saved_possession(&receiver_endpoint_id, &challenge)
.await
{
Ok(proof) => proof,
Err(error) => {
if store
.set_state(
&transfer_uuid,
TargetedTransferState::Offering,
TargetedTransferState::Failed,
)
.await
.is_ok()
{
self.emit_targeted_lifecycle(&transfer_uuid, "failed");
}
self.teardown_targeted_payload(protocol_transfer_id, Some(&transfer_uuid))
.await;
return Err(error);
}
};
let protocol_version = saved_device_capabilities().targeted_transfer_protocol_version;
if let Err(error) = store
.set_state(
&transfer_uuid,
TargetedTransferState::Offering,
TargetedTransferState::AwaitingApproval,
)
.await
{
self.teardown_targeted_payload(protocol_transfer_id, Some(&transfer_uuid))
.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(),
client.submit_offer(SubmitTargetedOffer {
proof,
generation,
relationship_protocol_version,
protocol_version,
transfer_id: transfer_uuid.clone(),
sender_endpoint_id: sender_endpoint_id.clone(),
receiver_endpoint_id: receiver_endpoint_id.clone(),
manifest_id: blob_ticket.hash().to_string(),
content_hash: blob_ticket.hash().to_string(),
transfer_name: payload_name.clone(),
file_count: import.file_count,
total_size: import.total_size,
relay_mode: self.relay_mode,
relay_urls: self
.custom_relay_urls
.iter()
.map(ToString::to_string)
.collect(),
}),
)
.await
{
Ok(Ok(response)) => response,
Ok(Err(error)) => {
if store
.set_state(
&transfer_uuid,
TargetedTransferState::AwaitingApproval,
TargetedTransferState::Failed,
)
.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(_) => {
if store
.set_state(
&transfer_uuid,
TargetedTransferState::AwaitingApproval,
TargetedTransferState::Failed,
)
.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!(
"offer timed out"
)));
}
};
match response {
WireOfferResponse::Accepted => {}
WireOfferResponse::Declined { reason } => {
if store
.set_state(
&transfer_uuid,
TargetedTransferState::AwaitingApproval,
TargetedTransferState::Declined,
)
.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!(
"targeted offer declined: {reason}"
)));
}
WireOfferResponse::Refused { reason } => {
if store
.set_state(
&transfer_uuid,
TargetedTransferState::AwaitingApproval,
TargetedTransferState::Failed,
)
.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));
}
}
// Permanent until cancel/delete — approved targeted transfers must resume.
self.access_policy
.approve_endpoint_until(protocol_transfer_id, receiver_endpoint_id.clone(), None)
.await;
let authorization = match TargetedAuthorization::issue(TargetedAuthorizationDraft {
transfer_id: transfer_uuid.clone(),
protocol_transfer_id,
sender_endpoint_id,
receiver_endpoint_id,
manifest_id: blob_ticket.hash().to_string(),
content_hash: blob_ticket.hash().to_string(),
file_count: import.file_count,
total_size: import.total_size,
protocol_version,
transfer_name: payload_name,
blob_ticket: blob_ticket.to_string(),
}) {
Ok(authorization) => authorization,
Err(error) => {
if store
.set_state_from_any(&transfer_uuid, TargetedTransferState::Failed)
.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 let Err(error) = self
.persist_sender_authorization_and_approve(&authorization)
.await
{
if store
.set_state_from_any(&transfer_uuid, TargetedTransferState::Failed)
.await
.is_ok()
{
self.emit_targeted_lifecycle(&transfer_uuid, "failed");
}
self.teardown_targeted_payload(protocol_transfer_id, Some(&transfer_uuid))
.await;
return Err(error);
}
self.emit_targeted_lifecycle(&transfer_uuid, "approved");
store
.get(&transfer_uuid)
.await?
.ok_or_else(|| VnidropError::internal(anyhow::anyhow!("targeted transfer missing")))
}
}

197
crates/vnidrop/src/runtime/targeted_payload.rs Normal file
View File

@@ -0,0 +1,197 @@
//! Shared blob download and export path for targeted transfers.
use std::sync::Arc;
use anyhow::{Context, Result};
use futures_lite::StreamExt;
use iroh_blobs::{
api::remote::GetProgressItem, format::collection::Collection,
get::request::get_hash_seq_and_sizes, ticket::BlobTicket,
};
use serde_json::json;
use tokio::sync::oneshot;
use super::{receive::ReceiveTarget, CoreInner};
use crate::{error::VnidropError, ticket::filter_peer_addr_for_relay_mode};
impl CoreInner {
#[allow(
clippy::too_many_arguments,
reason = "targeted receive binds immutable authorization fields plus its cancel token"
)]
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,
mut cancelled: oneshot::Receiver<()>,
) -> Result<()> {
let sender_addr = filter_peer_addr_for_relay_mode(
blob_ticket.addr(),
self.relay_mode,
&self.custom_relay_urls,
)
.map_err(VnidropError::network)?;
blob_ticket = BlobTicket::new(sender_addr, blob_ticket.hash(), blob_ticket.format());
let _permit = tokio::select! {
biased;
_ = &mut cancelled => return Err(VnidropError::cancelled("transfer cancelled").into()),
permit = self.transfer_slots.acquire() => permit
.context("transfer limiter is closed")
.map_err(VnidropError::internal)?,
};
tokio::select! {
biased;
_ = &mut cancelled => Err(VnidropError::cancelled("transfer cancelled").into()),
result = self.download_targeted_payload(
targeted_transfer_id,
transfer_id,
expected_file_count,
expected_payload_size,
blob_ticket,
target,
) => 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<()> {
if let ReceiveTarget::Directory(output_dir) = &target {
tokio::fs::create_dir_all(output_dir)
.await
.map_err(VnidropError::filesystem)?;
}
let connection = self
.endpoint
.connect(blob_ticket.addr().clone(), iroh_blobs::ALPN)
.await
.map_err(VnidropError::network)?;
let hash_and_format = blob_ticket.hash_and_format();
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 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 remote_size > self.limits.max_total_bytes {
anyhow::bail!(
"remote collection size {remote_size} exceeds limit {}",
self.limits.max_total_bytes
);
}
let download_tag = self.store.tags().temp_tag(hash_and_format).await?;
let get = self.store.remote().fetch(connection, hash_and_format);
let mut stream = get.stream();
loop {
let Some(item) = stream.next().await else {
anyhow::bail!("targeted download ended without completion");
};
match item {
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}"
))
.into());
}
}
}
let collection = Collection::load(hash_and_format.hash, self.store.as_ref()).await?;
self.export_collection_untracked(transfer_id, total_files, target, collection)
.await?;
drop(download_tag);
Ok(())
}
}

381
crates/vnidrop/src/runtime/targeted_receive.rs Normal file
View File

@@ -0,0 +1,381 @@
//! Receiver-side targeted-transfer execution and completion.
use std::sync::Arc;
use anyhow::Result;
use iroh_blobs::ticket::BlobTicket;
use super::{receive::ReceiveTarget, targeted::BlobTicketParse, CoreInner};
use crate::{
api::{saved_device_capabilities, TargetedTransferState},
error::VnidropError,
secure_secret::{SecretHandle, SecretKind},
targeted_transfer::{
auth_secret_material,
protocol::{CompleteTargetedTransfer, CompletionResponse, TargetedTransferProtocol},
reconstruct_authorization, TargetedAuthorization, TargetedAuthorizationDraft,
TargetedTransferRow,
},
};
impl CoreInner {
pub(super) async fn receive_targeted_transfer(
self: &Arc<Self>,
transfer_id: String,
output_dir: String,
) -> Result<(), VnidropError> {
let output_dir =
crate::filesystem::platform_path(&output_dir).map_err(VnidropError::filesystem)?;
self.receive_targeted_to_target(transfer_id, ReceiveTarget::Directory(output_dir))
.await
}
pub(super) async fn receive_targeted_transfer_with_output_sink(
self: &Arc<Self>,
transfer_id: String,
output_sink: Arc<dyn crate::ReceiveOutputSink>,
) -> Result<(), VnidropError> {
self.receive_targeted_to_target(transfer_id, ReceiveTarget::OutputSink(output_sink))
.await
}
pub(super) async fn receive_targeted_transfer_with_output_sink_v2(
self: &Arc<Self>,
transfer_id: String,
output_sink: Arc<dyn crate::ReceiveOutputSinkV2>,
) -> Result<(), VnidropError> {
self.receive_targeted_to_target(transfer_id, ReceiveTarget::OutputSinkV2(output_sink))
.await
}
pub(super) async fn resume_targeted_transfer(
self: &Arc<Self>,
id: String,
output_dir: String,
) -> Result<(), VnidropError> {
let output_dir =
crate::filesystem::platform_path(&output_dir).map_err(VnidropError::filesystem)?;
self.resume_targeted_to_target(id, ReceiveTarget::Directory(output_dir))
.await
}
pub(super) async fn resume_targeted_transfer_with_output_sink(
self: &Arc<Self>,
id: String,
output_sink: Arc<dyn crate::ReceiveOutputSink>,
) -> Result<(), VnidropError> {
self.resume_targeted_to_target(id, ReceiveTarget::OutputSink(output_sink))
.await
}
pub(super) async fn resume_targeted_transfer_with_output_sink_v2(
self: &Arc<Self>,
id: String,
output_sink: Arc<dyn crate::ReceiveOutputSinkV2>,
) -> Result<(), VnidropError> {
self.resume_targeted_to_target(id, ReceiveTarget::OutputSinkV2(output_sink))
.await
}
async fn receive_targeted_to_target(
self: &Arc<Self>,
transfer_id: String,
target: ReceiveTarget,
) -> Result<(), VnidropError> {
let auth = self.load_receiver_authorization(&transfer_id).await?;
self.run_targeted_receive(&auth, target).await
}
async fn resume_targeted_to_target(
self: &Arc<Self>,
id: String,
target: ReceiveTarget,
) -> Result<(), VnidropError> {
let store = self.targeted_store();
let row = store.get_row(&id).await?.ok_or_else(|| {
VnidropError::invalid_input(anyhow::anyhow!("unknown targeted transfer"))
})?;
if !matches!(
row.state,
TargetedTransferState::Approved
| TargetedTransferState::Connecting
| TargetedTransferState::Transferring
| TargetedTransferState::Interrupted
) {
return Err(VnidropError::InvalidTransition {
reason: format!(
"cannot resume from {}",
crate::targeted_transfer::state_as_str(row.state)
),
});
}
let auth = self.load_receiver_authorization(&id).await?;
self.run_targeted_receive(&auth, target).await
}
async fn load_receiver_authorization(
&self,
transfer_id: &str,
) -> Result<TargetedAuthorization, VnidropError> {
let store = self.targeted_store();
let row = store.get_row(transfer_id).await?.ok_or_else(|| {
VnidropError::invalid_input(anyhow::anyhow!("unknown targeted transfer"))
})?;
let encoded = self.load_stored_authorization(&row).await?.ok_or_else(|| {
VnidropError::invalid_input(anyhow::anyhow!(
"targeted transfer has no durable authorization"
))
})?;
let auth = TargetedAuthorization::decode(&encoded)?;
auth.verify_for_receiver(&self.endpoint.id().to_string())?;
Ok(auth)
}
async fn run_targeted_receive(
self: &Arc<Self>,
auth: &TargetedAuthorization,
target: ReceiveTarget,
) -> Result<(), VnidropError> {
let (cancel, cancelled) = tokio::sync::oneshot::channel();
{
let mut active = self
.active_targeted_transfers
.lock()
.expect("active_targeted_transfers");
match active.entry(auth.transfer_id.clone()) {
std::collections::hash_map::Entry::Vacant(entry) => {
entry.insert(crate::runtime::ActiveTransfer {
direction: crate::transfer_state::TransferDirection::Receive,
cancel,
});
}
std::collections::hash_map::Entry::Occupied(_) => {
return Err(VnidropError::InvalidTransition {
reason: "targeted receive is already active".to_string(),
});
}
}
}
let result = self
.run_registered_targeted_receive(auth, target, cancelled)
.await;
self.active_targeted_transfers
.lock()
.expect("active_targeted_transfers")
.remove(&auth.transfer_id);
result
}
async fn run_registered_targeted_receive(
self: &Arc<Self>,
auth: &TargetedAuthorization,
target: ReceiveTarget,
cancelled: tokio::sync::oneshot::Receiver<()>,
) -> Result<(), VnidropError> {
let store = self.targeted_store();
if let Some(row) = store.get_row(&auth.transfer_id).await? {
match row.state {
TargetedTransferState::Approved | TargetedTransferState::Interrupted => {
store
.set_state(
&auth.transfer_id,
row.state,
TargetedTransferState::Connecting,
)
.await?;
self.emit_targeted_lifecycle(&auth.transfer_id, "connecting");
store
.set_state(
&auth.transfer_id,
TargetedTransferState::Connecting,
TargetedTransferState::Transferring,
)
.await?;
self.emit_targeted_lifecycle(&auth.transfer_id, "transferring");
}
TargetedTransferState::Connecting | TargetedTransferState::Transferring => {
return Err(VnidropError::InvalidTransition {
reason: "targeted receive is already active".to_string(),
});
}
other => {
return Err(VnidropError::InvalidTransition {
reason: format!(
"cannot receive from {}",
crate::targeted_transfer::state_as_str(other)
),
});
}
}
}
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.transfer_id,
auth.protocol_transfer_id,
auth.file_count,
auth.total_size,
blob_ticket,
target,
cancelled,
)
.await;
match receive_result {
Ok(()) => {
let row = store.get_row(&auth.transfer_id).await?.ok_or_else(|| {
VnidropError::internal(anyhow::anyhow!("targeted transfer missing"))
})?;
store
.complete_receiver_and_enqueue(&auth.transfer_id, row.total_size)
.await?;
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?;
}
Ok(())
}
Err(error) => {
match store.get_row(&auth.transfer_id).await {
Ok(Some(row))
if matches!(
row.state,
TargetedTransferState::Connecting | TargetedTransferState::Transferring
) =>
{
match store
.set_state_from_any(
&auth.transfer_id,
TargetedTransferState::Interrupted,
)
.await
{
Ok(()) => {
self.emit_targeted_lifecycle(&auth.transfer_id, "interrupted")
}
Err(state_error) => {
tracing::warn!(transfer_id = %auth.transfer_id, %state_error, "failed to persist targeted receive interruption")
}
}
}
Ok(_) => {}
Err(state_error) => {
tracing::warn!(transfer_id = %auth.transfer_id, %state_error, "failed to load targeted receive state after error")
}
}
Err(VnidropError::transfer(error))
}
}
}
pub(super) async fn acknowledge_targeted_completion(
&self,
auth: &TargetedAuthorization,
) -> Result<(), VnidropError> {
#[cfg(test)]
if self
.suppress_targeted_completion
.load(std::sync::atomic::Ordering::SeqCst)
{
return Err(VnidropError::device_unavailable(anyhow::anyhow!(
"completion delivery suppressed by test"
)));
}
let addr = self
.device_relationships
.peer_addr(&auth.sender_endpoint_id)
.await?;
let client = TargetedTransferProtocol::client(self.endpoint.clone(), addr);
let response = tokio::time::timeout(
self.connection_timeout(),
client.complete_transfer(CompleteTargetedTransfer {
transfer_id: auth.transfer_id.clone(),
verified_bytes: auth.total_size,
authorization: auth.encode()?,
}),
)
.await
.map_err(|_| VnidropError::device_unavailable(anyhow::anyhow!("completion timed out")))?
.map_err(|error| VnidropError::network(anyhow::anyhow!(error)))?;
if response != CompletionResponse::Recorded {
return Err(VnidropError::permission(anyhow::anyhow!(
"sender rejected targeted completion"
)));
}
Ok(())
}
pub(super) async fn persist_sender_authorization_and_approve(
&self,
authorization: &TargetedAuthorization,
) -> Result<(), VnidropError> {
let custody =
self.secret_custody
.as_ref()
.ok_or_else(|| VnidropError::SecureStorageUnavailable {
reason: "targeted authorization requires protected custody".to_string(),
})?;
let handle = custody
.protect(
SecretKind::TargetedAuthorization,
auth_secret_material(authorization)?,
None,
)
.await?;
if let Err(error) = self
.targeted_store()
.finalize_sender_authorization_and_enqueue(
&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 sender authorization secret");
}
return Err(error);
}
Ok(())
}
pub(crate) async fn load_stored_authorization(
&self,
row: &TargetedTransferRow,
) -> Result<Option<String>, VnidropError> {
let (Some(handle), Some(blob_ticket)) = (
row.authorization_secret_handle.as_ref(),
row.blob_ticket.as_ref(),
) else {
return Ok(None);
};
let custody =
self.secret_custody
.as_ref()
.ok_or_else(|| VnidropError::SecureStorageUnavailable {
reason: "targeted authorization requires protected custody".to_string(),
})?;
let material = custody
.load(&SecretHandle::from_stored(handle.clone()))
.await?;
let auth = reconstruct_authorization(
TargetedAuthorizationDraft {
transfer_id: row.id.clone(),
protocol_transfer_id: row.protocol_transfer_id,
sender_endpoint_id: row.sender_endpoint_id.clone(),
receiver_endpoint_id: row.receiver_endpoint_id.clone(),
manifest_id: row.manifest_id.clone(),
content_hash: row.content_hash.clone(),
file_count: row.file_count,
total_size: row.total_size,
protocol_version: saved_device_capabilities().targeted_transfer_protocol_version,
transfer_name: row.transfer_name.clone(),
blob_ticket: blob_ticket.clone(),
},
&material,
)?;
Ok(Some(auth.encode()?))
}
}

214
crates/vnidrop/src/runtime/targeted_reconciliation.rs Normal file
View File

@@ -0,0 +1,214 @@
//! Durable targeted-transfer background reconciliation.
use std::sync::Arc;
use anyhow::Result;
use super::{targeted::map_connect_failure, CoreInner};
use crate::{
error::VnidropError,
targeted_transfer::{
protocol::{DeliverTargetedAuthorization, TargetedTransferProtocol},
TargetedAuthorization, TargetedTransferRow,
},
util::now_ms,
};
impl CoreInner {
pub(super) async fn spawn_targeted_reconciliation_task(self: &Arc<Self>) {
let core = self.clone();
let task = tokio::spawn(async move {
let mut interval = tokio::time::interval(std::time::Duration::from_secs(2));
interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
loop {
interval.tick().await;
core.release_completed_targeted_payloads().await;
core.retry_targeted_authorization_deliveries().await;
core.retry_targeted_completions().await;
}
});
*self.targeted_reconciliation_task.lock().await = Some(task);
}
async fn retry_targeted_completions(&self) {
let rows = match self.targeted_store().list_pending_completions().await {
Ok(rows) => rows,
Err(error) => {
tracing::warn!(%error, "failed to load pending targeted completions");
return;
}
};
for row in rows.into_iter().take(1) {
let encoded = match self.load_stored_authorization(&row).await {
Ok(Some(encoded)) => encoded,
Ok(None) => {
tracing::warn!(transfer_id = %row.id, "pending targeted completion has no authorization");
self.defer_targeted_completion_with_log(&row.id, now_ms() + 30_000)
.await;
continue;
}
Err(error) => {
tracing::warn!(transfer_id = %row.id, %error, "failed to load pending targeted completion authorization");
self.defer_targeted_completion_with_log(&row.id, now_ms() + 30_000)
.await;
continue;
}
};
let auth = match TargetedAuthorization::decode(&encoded) {
Ok(auth) => auth,
Err(error) => {
tracing::warn!(transfer_id = %row.id, %error, "failed to decode pending targeted completion authorization");
self.defer_targeted_completion_with_log(&row.id, now_ms() + 30_000)
.await;
continue;
}
};
match self.acknowledge_targeted_completion(&auth).await {
Ok(()) => {
if let Err(error) = self
.targeted_store()
.clear_pending_completion(&row.id)
.await
{
tracing::warn!(transfer_id = %row.id, %error, "failed to settle targeted completion");
}
}
Err(error) => {
tracing::warn!(transfer_id = %row.id, %error, "targeted completion retry failed");
self.defer_targeted_completion_with_log(&row.id, now_ms() + 5_000)
.await;
}
}
}
}
async fn defer_targeted_completion_with_log(&self, id: &str, next_attempt_at: i64) {
if let Err(error) = self
.targeted_store()
.defer_pending_completion(id, next_attempt_at)
.await
{
tracing::warn!(transfer_id = %id, %error, "failed to defer targeted completion");
}
}
async fn retry_targeted_authorization_deliveries(&self) {
let rows = match self
.targeted_store()
.list_pending_authorization_deliveries()
.await
{
Ok(rows) => rows,
Err(error) => {
tracing::warn!(%error, "failed to load pending targeted authorization deliveries");
return;
}
};
for row in rows.into_iter().take(1) {
match self.deliver_stored_targeted_authorization(&row).await {
Ok(true) => {
if let Err(error) = self
.targeted_store()
.clear_pending_authorization_delivery(&row.id)
.await
{
tracing::warn!(transfer_id = %row.id, %error, "failed to settle targeted authorization delivery");
}
}
Ok(false) => {
tracing::warn!(transfer_id = %row.id, "receiver rejected targeted authorization delivery");
if let Err(error) = self
.targeted_store()
.defer_pending_authorization_delivery(&row.id, now_ms() + 5_000)
.await
{
tracing::warn!(transfer_id = %row.id, %error, "failed to defer rejected targeted authorization delivery");
}
}
Err(error) => {
tracing::warn!(transfer_id = %row.id, %error, "targeted authorization delivery retry failed");
if let Err(error) = self
.targeted_store()
.defer_pending_authorization_delivery(&row.id, now_ms() + 5_000)
.await
{
tracing::warn!(transfer_id = %row.id, %error, "failed to defer targeted authorization delivery");
}
}
}
}
}
pub(super) async fn deliver_stored_targeted_authorization(
&self,
row: &TargetedTransferRow,
) -> Result<bool, VnidropError> {
#[cfg(test)]
self.targeted_authorization_delivery_attempts
.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
#[cfg(test)]
if self
.suppress_targeted_authorization_delivery
.load(std::sync::atomic::Ordering::SeqCst)
{
return Err(VnidropError::device_unavailable(anyhow::anyhow!(
"targeted authorization delivery suppressed by test"
)));
}
let encoded = self.load_stored_authorization(row).await?.ok_or_else(|| {
VnidropError::SecureStorageMissing {
reason: "targeted authorization is missing".to_string(),
}
})?;
let addr = self
.device_relationships
.peer_addr(&row.receiver_endpoint_id)
.await?;
let response = tokio::time::timeout(
self.connection_timeout(),
TargetedTransferProtocol::client(self.endpoint.clone(), addr).deliver_authorization(
DeliverTargetedAuthorization {
transfer_id: row.id.clone(),
authorization: encoded,
},
),
)
.await
.map_err(|_| {
VnidropError::device_unavailable(anyhow::anyhow!(
"targeted authorization delivery timed out"
))
})?
.map_err(map_connect_failure)?;
Ok(response == crate::targeted_transfer::protocol::DeliverAuthorizationResponse::Stored)
}
async fn release_completed_targeted_payloads(&self) {
let rows = match self.targeted_store().list_completed_sender_rows().await {
Ok(rows) => rows,
Err(error) => {
tracing::warn!(%error, "failed to load pending targeted payload releases");
return;
}
};
for row in rows {
match self
.try_teardown_targeted_payload(row.protocol_transfer_id, Some(&row.id))
.await
{
Ok(()) => {
if let Err(error) = self
.targeted_store()
.clear_pending_payload_release(&row.id)
.await
{
tracing::warn!(transfer_id = %row.id, %error, "failed to settle targeted payload release");
}
}
Err(error) => {
tracing::warn!(transfer_id = %row.id, %error, "targeted payload release retry failed")
}
}
}
}
}

35
crates/vnidrop/src/secure_secret.rs
View File

@@ -419,6 +419,19 @@ impl SecretCustody {
Ok(handle)
}
#[cfg(test)]
pub(crate) async fn create_orphaned_targeted_authorization_for_test(
&self,
) -> Result<(), VnidropError> {
self.protect(
SecretKind::TargetedAuthorization,
SecretMaterial::new(vec![7; 32])?,
None,
)
.await?;
Ok(())
}
pub(crate) async fn load(&self, handle: &SecretHandle) -> Result<SecretMaterial, VnidropError> {
let metadata = self.metadata.find(handle).await?.ok_or_else(|| {
VnidropError::SecureStorageMissing {
@@ -629,6 +642,24 @@ impl SecretCustody {
Ok(summary)
}
pub(crate) async fn remove_orphaned_targeted_authorizations(
&self,
referenced_handles: &HashSet<String>,
) -> Result<u64, VnidropError> {
let entries = self.metadata.list().await?;
let mut removed = 0;
for entry in entries {
if entry.kind == SecretKind::TargetedAuthorization
&& entry.state == SecretMetadataState::Active
&& !referenced_handles.contains(entry.handle.as_str())
{
self.remove(&entry.handle).await?;
removed += 1;
}
}
Ok(removed)
}
async fn delete_if_present(&self, handle: &SecretHandle) -> Result<(), VnidropError> {
match self.store_delete_raw(handle.clone()).await? {
Ok(()) | Err(SecureSecretStoreError::Missing) => Ok(()),
@@ -825,6 +856,10 @@ impl FaultInjectingSecretStore {
self.corrupted.lock().unwrap().push(handle.clone());
}
pub(crate) fn stored_value_count_for_test(&self) -> usize {
self.values.lock().unwrap().len()
}
pub(crate) fn only_handle_for_test(&self) -> SecretHandle {
let handles = self
.values

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

@@ -218,6 +218,7 @@ impl TargetedOfferInbox {
.map(|entry| entry.offer.clone())
}
#[cfg(test)]
pub(crate) async fn authorization_matches_pending(
&self,
auth: &crate::targeted_transfer::TargetedAuthorization,
@@ -271,16 +272,14 @@ impl TargetedOfferInbox {
let Some(sender_endpoint_id) = sender_endpoint_id else {
return Err(RespondError::Unknown);
};
let waiter = {
let decisions = self.decisions.lock().await;
decisions
.get(transfer_id)
.map(|entry| entry.decision.clone())
};
let Some(decision_tx) = waiter else {
return Err(RespondError::Unknown);
};
if !accepted {
let decision_tx = {
let decisions = self.decisions.lock().await;
decisions
.get(transfer_id)
.map(|entry| entry.decision.clone())
.ok_or(RespondError::Unknown)?
};
let _ = decision_tx.send(Some(false));
self.discard(transfer_id).await;
self.cooldown.record_decline(&sender_endpoint_id);
@@ -298,17 +297,52 @@ impl TargetedOfferInbox {
return Ok(None);
}
let (auth_tx, mut auth_rx) = watch::channel(None);
self.accept_live(transfer_id).await?;
self.wait_for_authorization(transfer_id).await
}
pub(crate) async fn pending_for_acceptance(
&self,
transfer_id: &str,
) -> Option<PendingTargetedOffer> {
self.get_pending(transfer_id).await
}
pub(crate) async fn accept_live(&self, transfer_id: &str) -> Result<(), RespondError> {
let decision_tx = {
let decisions = self.decisions.lock().await;
decisions
.get(transfer_id)
.map(|entry| entry.decision.clone())
.ok_or(RespondError::Unknown)?
};
let (auth_tx, auth_rx) = watch::channel(None);
self.auths
.lock()
.await
.insert(transfer_id.to_string(), AuthWaiter { auth: auth_tx });
if decision_tx.send(Some(true)).is_err() {
self.auths.lock().await.remove(transfer_id);
self.discard(transfer_id).await;
return Err(RespondError::SenderGone);
}
drop(auth_rx);
Ok(())
}
pub(crate) async fn wait_for_authorization(
&self,
transfer_id: &str,
) -> Result<Option<String>, RespondError> {
if let Some(auth) = self.settled_authorization(transfer_id).await {
return Ok(Some(auth));
}
let mut auth_rx = {
let auths = self.auths.lock().await;
auths
.get(transfer_id)
.map(|entry| entry.auth.subscribe())
.ok_or(RespondError::Unknown)?
};
let wait_auth = async {
loop {
if let Some(auth) = auth_rx.borrow_and_update().clone() {
@@ -333,7 +367,7 @@ impl TargetedOfferInbox {
Ok(Some(auth))
}
Ok(Err(())) | Err(_) => {
self.discard(transfer_id).await;
self.auths.lock().await.remove(transfer_id);
Err(RespondError::AuthorizationTimeout)
}
}

5
crates/vnidrop/src/targeted_transfer/mod.rs
View File

@@ -6,8 +6,10 @@
mod auth;
pub(crate) mod inbox;
pub(crate) mod protocol;
mod schema;
mod state;
mod store;
mod store_outbox;
pub(crate) use auth::{
auth_secret_material, reconstruct_authorization, TargetedAuthorization,
@@ -15,6 +17,7 @@ pub(crate) use auth::{
};
pub(crate) use inbox::{RespondError, TargetedOfferInbox};
pub(crate) use protocol::TargetedTransferProtocol;
pub(crate) use schema::ensure_schema;
pub(crate) use store::{
ensure_schema, state_as_str, TargetedTransferRole, TargetedTransferRow, TargetedTransferStore,
state_as_str, TargetedTransferRole, TargetedTransferRow, TargetedTransferStore,
};

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

@@ -43,10 +43,15 @@ pub(crate) struct TargetedTransferProtocol {
access_policy: std::sync::Arc<crate::access_policy::AccessPolicy>,
cleanup:
std::sync::Arc<dyn Fn(super::TargetedTransferRow) -> TargetedCleanupFuture + Send + Sync>,
persist_authorization: std::sync::Arc<
dyn Fn(TargetedAuthorization) -> TargetedAuthorizationPersistFuture + Send + Sync,
>,
}
pub(crate) type TargetedCleanupFuture =
Pin<Box<dyn Future<Output = Result<(), VnidropError>> + Send>>;
pub(crate) type TargetedAuthorizationPersistFuture =
Pin<Box<dyn Future<Output = Result<bool, VnidropError>> + Send>>;
impl fmt::Debug for TargetedTransferProtocol {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
@@ -71,6 +76,9 @@ impl TargetedTransferProtocol {
cleanup: std::sync::Arc<
dyn Fn(super::TargetedTransferRow) -> TargetedCleanupFuture + Send + Sync,
>,
persist_authorization: std::sync::Arc<
dyn Fn(TargetedAuthorization) -> TargetedAuthorizationPersistFuture + Send + Sync,
>,
) -> Self {
Self {
relationships,
@@ -83,6 +91,7 @@ impl TargetedTransferProtocol {
event_hub,
access_policy,
cleanup,
persist_authorization,
}
}
@@ -251,6 +260,14 @@ impl TargetedTransferProtocol {
remote_endpoint_id: &str,
delivery: DeliverTargetedAuthorization,
) -> DeliverAuthorizationResponse {
if self
.relationships
.require_saved(remote_endpoint_id)
.await
.is_err()
{
return DeliverAuthorizationResponse::Rejected;
}
let Ok(auth) = TargetedAuthorization::decode(&delivery.authorization) else {
return DeliverAuthorizationResponse::Rejected;
};
@@ -260,57 +277,23 @@ impl TargetedTransferProtocol {
{
return DeliverAuthorizationResponse::Rejected;
}
if let Ok(Some(row)) = self.store.get_row(&delivery.transfer_id).await {
if row.authorization_secret_handle.is_some()
&& row.protocol_transfer_id == auth.protocol_transfer_id
&& row.sender_endpoint_id == auth.sender_endpoint_id
&& row.receiver_endpoint_id == auth.receiver_endpoint_id
&& row.manifest_id == auth.manifest_id
&& row.content_hash == auth.content_hash
&& row.transfer_name == auth.transfer_name
&& row.file_count == auth.file_count
&& row.total_size == auth.total_size
&& row.blob_ticket.as_deref() == Some(auth.blob_ticket.as_str())
&& auth.protocol_version
== saved_device_capabilities().targeted_transfer_protocol_version
{
return DeliverAuthorizationResponse::Stored;
}
if row.authorization_secret_handle.is_some() {
return DeliverAuthorizationResponse::Rejected;
}
}
if !self.inbox.authorization_matches_pending(&auth).await {
if auth.protocol_version != saved_device_capabilities().targeted_transfer_protocol_version {
return DeliverAuthorizationResponse::Rejected;
}
if self
.inbox
.deliver_authorization(&delivery.transfer_id, delivery.authorization)
.await
{
let deadline = tokio::time::Instant::now()
+ std::time::Duration::from_millis(self.limits.offer_timeout_ms);
loop {
if let Ok(Some(row)) = self.store.get_row(&delivery.transfer_id).await {
if row.authorization_secret_handle.is_some() {
return DeliverAuthorizationResponse::Stored;
}
if matches!(
row.state,
TargetedTransferState::Failed
| TargetedTransferState::Cancelled
| TargetedTransferState::Deleted
) {
return DeliverAuthorizationResponse::Rejected;
}
}
if tokio::time::Instant::now() >= deadline {
return DeliverAuthorizationResponse::Rejected;
}
tokio::time::sleep(std::time::Duration::from_millis(25)).await;
}
} else {
let Ok(ticket) = auth.blob_ticket.parse::<iroh_blobs::ticket::BlobTicket>() else {
return DeliverAuthorizationResponse::Rejected;
};
if ticket.hash().to_string() != auth.manifest_id || auth.manifest_id != auth.content_hash {
return DeliverAuthorizationResponse::Rejected;
}
if (self.persist_authorization)(auth).await.is_err() {
DeliverAuthorizationResponse::Rejected
} else {
let _ = self
.inbox
.deliver_authorization(&delivery.transfer_id, delivery.authorization)
.await;
DeliverAuthorizationResponse::Stored
}
}
@@ -324,6 +307,14 @@ impl TargetedTransferProtocol {
return CancelWireOfferResponse::Rejected;
}
self.inbox.discard(&cancel.transfer_id).await;
if self
.store
.clear_accepted_offer_intent(&cancel.transfer_id)
.await
.is_err()
{
return CancelWireOfferResponse::Rejected;
}
return CancelWireOfferResponse::Cancelled;
}
if let Ok(Some(row)) = self.store.get_row(&cancel.transfer_id).await {
@@ -382,6 +373,14 @@ impl TargetedTransferProtocol {
}
return CancelWireOfferResponse::Cancelled;
}
if self
.store
.clear_accepted_intent_if_sender(&cancel.transfer_id, remote_endpoint_id)
.await
.is_err()
{
return CancelWireOfferResponse::Rejected;
}
CancelWireOfferResponse::Cancelled
}

167
crates/vnidrop/src/targeted_transfer/schema.rs Normal file
View File

@@ -0,0 +1,167 @@
//! Targeted-transfer SQLite schema and one-time migrations.
use sqlx::{Row, SqlitePool};
use crate::util::now_ms;
pub(crate) async fn ensure_schema(pool: &SqlitePool) -> anyhow::Result<()> {
sqlx::query(
r#"
CREATE TABLE IF NOT EXISTS targeted_transfers (
id TEXT PRIMARY KEY,
protocol_transfer_id INTEGER NOT NULL UNIQUE,
sender_endpoint_id TEXT NOT NULL,
receiver_endpoint_id TEXT NOT NULL,
manifest_id TEXT NOT NULL,
content_hash TEXT NOT NULL,
transfer_name TEXT NOT NULL,
file_count INTEGER NOT NULL,
total_size INTEGER NOT NULL,
verified_bytes INTEGER NOT NULL DEFAULT 0,
blob_ticket TEXT,
authorization_secret_handle TEXT,
role TEXT NOT NULL DEFAULT 'sender',
state TEXT NOT NULL,
created_at INTEGER NOT NULL,
updated_at INTEGER NOT NULL
);
"#,
)
.execute(pool)
.await?;
sqlx::query(
r#"
CREATE TABLE IF NOT EXISTS targeted_accepted_offer_intents (
transfer_id TEXT PRIMARY KEY,
sender_endpoint_id TEXT NOT NULL,
receiver_endpoint_id TEXT NOT NULL,
manifest_id TEXT NOT NULL,
content_hash TEXT NOT NULL,
transfer_name TEXT NOT NULL,
file_count INTEGER NOT NULL,
total_size INTEGER NOT NULL,
protocol_version INTEGER NOT NULL,
accepted_at INTEGER NOT NULL
);
"#,
)
.execute(pool)
.await?;
sqlx::query(
r#"
CREATE TABLE IF NOT EXISTS targeted_authorization_delivery_outbox (
transfer_id TEXT PRIMARY KEY,
created_at INTEGER NOT NULL,
next_attempt_at INTEGER NOT NULL,
FOREIGN KEY(transfer_id) REFERENCES targeted_transfers(id) ON DELETE CASCADE
);
"#,
)
.execute(pool)
.await?;
sqlx::query(
r#"
CREATE TABLE IF NOT EXISTS targeted_completion_outbox (
transfer_id TEXT PRIMARY KEY,
created_at INTEGER NOT NULL,
next_attempt_at INTEGER NOT NULL,
FOREIGN KEY(transfer_id) REFERENCES targeted_transfers(id) ON DELETE CASCADE
);
"#,
)
.execute(pool)
.await?;
let completion_columns = sqlx::query("PRAGMA table_info(targeted_completion_outbox)")
.fetch_all(pool)
.await?;
if !completion_columns
.iter()
.any(|row| row.get::<String, _>(1) == "next_attempt_at")
{
sqlx::query(
"ALTER TABLE targeted_completion_outbox ADD COLUMN next_attempt_at INTEGER NOT NULL DEFAULT 0",
)
.execute(pool)
.await?;
}
sqlx::query(
r#"
CREATE TABLE IF NOT EXISTS targeted_payload_release_outbox (
transfer_id TEXT PRIMARY KEY,
created_at INTEGER NOT NULL,
FOREIGN KEY(transfer_id) REFERENCES targeted_transfers(id) ON DELETE CASCADE
);
"#,
)
.execute(pool)
.await?;
let columns = sqlx::query("PRAGMA table_info(targeted_transfers)")
.fetch_all(pool)
.await?;
let has = |name: &str| columns.iter().any(|row| row.get::<String, _>(1) == name);
if !has("verified_bytes") {
sqlx::query(
"ALTER TABLE targeted_transfers ADD COLUMN verified_bytes INTEGER NOT NULL DEFAULT 0",
)
.execute(pool)
.await?;
}
if !has("blob_ticket") {
sqlx::query("ALTER TABLE targeted_transfers ADD COLUMN blob_ticket TEXT")
.execute(pool)
.await?;
}
if !has("authorization_secret_handle") {
sqlx::query("ALTER TABLE targeted_transfers ADD COLUMN authorization_secret_handle TEXT")
.execute(pool)
.await?;
}
if !has("role") {
sqlx::query(
"ALTER TABLE targeted_transfers ADD COLUMN role TEXT NOT NULL DEFAULT 'sender'",
)
.execute(pool)
.await?;
}
sqlx::query(
r#"
CREATE TABLE IF NOT EXISTS targeted_schema_migrations (
name TEXT PRIMARY KEY,
applied_at INTEGER NOT NULL
)
"#,
)
.execute(pool)
.await?;
let mut transaction = pool.begin().await?;
let applied = sqlx::query(
"SELECT EXISTS(SELECT 1 FROM targeted_schema_migrations WHERE name = 'authorization-delivery-outbox-v1')",
)
.fetch_one(&mut *transaction)
.await?
.get::<i64, _>(0)
!= 0;
if !applied {
// Pre-outbox Approved sender rows already hold a receiver-bound authorization.
// Re-delivery is idempotent and cannot create consent on a receiver.
sqlx::query(
r#"
INSERT OR IGNORE INTO targeted_authorization_delivery_outbox
(transfer_id, created_at, next_attempt_at)
SELECT id, updated_at, 0 FROM targeted_transfers
WHERE role = 'sender' AND state = 'approved'
AND blob_ticket IS NOT NULL AND authorization_secret_handle IS NOT NULL
"#,
)
.execute(&mut *transaction)
.await?;
sqlx::query(
"INSERT INTO targeted_schema_migrations (name, applied_at) VALUES ('authorization-delivery-outbox-v1', ?1)",
)
.bind(now_ms())
.execute(&mut *transaction)
.await?;
}
transaction.commit().await?;
Ok(())
}

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

@@ -11,101 +11,9 @@ use crate::{
util::now_ms,
};
pub(crate) async fn ensure_schema(pool: &SqlitePool) -> anyhow::Result<()> {
sqlx::query(
r#"
CREATE TABLE IF NOT EXISTS targeted_transfers (
id TEXT PRIMARY KEY,
protocol_transfer_id INTEGER NOT NULL UNIQUE,
sender_endpoint_id TEXT NOT NULL,
receiver_endpoint_id TEXT NOT NULL,
manifest_id TEXT NOT NULL,
content_hash TEXT NOT NULL,
transfer_name TEXT NOT NULL,
file_count INTEGER NOT NULL,
total_size INTEGER NOT NULL,
verified_bytes INTEGER NOT NULL DEFAULT 0,
blob_ticket TEXT,
authorization_secret_handle TEXT,
role TEXT NOT NULL DEFAULT 'sender',
state TEXT NOT NULL,
created_at INTEGER NOT NULL,
updated_at INTEGER NOT NULL
);
"#,
)
.execute(pool)
.await?;
sqlx::query(
r#"
CREATE TABLE IF NOT EXISTS targeted_completion_outbox (
transfer_id TEXT PRIMARY KEY,
created_at INTEGER NOT NULL,
next_attempt_at INTEGER NOT NULL,
FOREIGN KEY(transfer_id) REFERENCES targeted_transfers(id) ON DELETE CASCADE
);
"#,
)
.execute(pool)
.await?;
let completion_columns = sqlx::query("PRAGMA table_info(targeted_completion_outbox)")
.fetch_all(pool)
.await?;
if !completion_columns
.iter()
.any(|row| row.get::<String, _>(1) == "next_attempt_at")
{
sqlx::query(
"ALTER TABLE targeted_completion_outbox ADD COLUMN next_attempt_at INTEGER NOT NULL DEFAULT 0",
)
.execute(pool)
.await?;
}
sqlx::query(
r#"
CREATE TABLE IF NOT EXISTS targeted_payload_release_outbox (
transfer_id TEXT PRIMARY KEY,
created_at INTEGER NOT NULL,
FOREIGN KEY(transfer_id) REFERENCES targeted_transfers(id) ON DELETE CASCADE
);
"#,
)
.execute(pool)
.await?;
let columns = sqlx::query("PRAGMA table_info(targeted_transfers)")
.fetch_all(pool)
.await?;
let has = |name: &str| columns.iter().any(|row| row.get::<String, _>(1) == name);
if !has("verified_bytes") {
sqlx::query(
"ALTER TABLE targeted_transfers ADD COLUMN verified_bytes INTEGER NOT NULL DEFAULT 0",
)
.execute(pool)
.await?;
}
if !has("blob_ticket") {
sqlx::query("ALTER TABLE targeted_transfers ADD COLUMN blob_ticket TEXT")
.execute(pool)
.await?;
}
if !has("authorization_secret_handle") {
sqlx::query("ALTER TABLE targeted_transfers ADD COLUMN authorization_secret_handle TEXT")
.execute(pool)
.await?;
}
if !has("role") {
sqlx::query(
"ALTER TABLE targeted_transfers ADD COLUMN role TEXT NOT NULL DEFAULT 'sender'",
)
.execute(pool)
.await?;
}
Ok(())
}
#[derive(Clone)]
pub(crate) struct TargetedTransferStore {
pool: SqlitePool,
pub(super) pool: SqlitePool,
}
impl TargetedTransferStore {
@@ -113,6 +21,88 @@ impl TargetedTransferStore {
Self { pool }
}
pub(crate) async fn persist_accepted_offer_intent(
&self,
offer: &crate::api::PendingTargetedOffer,
) -> Result<(), VnidropError> {
let result = sqlx::query(
r#"
INSERT INTO targeted_accepted_offer_intents (
transfer_id, sender_endpoint_id, receiver_endpoint_id, manifest_id,
content_hash, transfer_name, file_count, total_size, protocol_version, accepted_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)
ON CONFLICT(transfer_id) DO UPDATE SET accepted_at = excluded.accepted_at
WHERE sender_endpoint_id = excluded.sender_endpoint_id
AND receiver_endpoint_id = excluded.receiver_endpoint_id
AND manifest_id = excluded.manifest_id
AND content_hash = excluded.content_hash
AND transfer_name = excluded.transfer_name
AND file_count = excluded.file_count
AND total_size = excluded.total_size
AND protocol_version = excluded.protocol_version
"#,
)
.bind(&offer.transfer_id)
.bind(&offer.sender_endpoint_id)
.bind(&offer.receiver_endpoint_id)
.bind(&offer.manifest_id)
.bind(&offer.content_hash)
.bind(&offer.transfer_name)
.bind(offer.file_count as i64)
.bind(offer.total_size as i64)
.bind(offer.protocol_version as i64)
.bind(now_ms())
.execute(&self.pool)
.await
.map_err(VnidropError::repository)?;
if result.rows_affected() == 0 {
return Err(VnidropError::invalid_input(anyhow::anyhow!(
"accepted targeted offer conflicts with durable consent"
)));
}
Ok(())
}
pub(crate) async fn clear_accepted_offer_intent(&self, id: &str) -> Result<(), VnidropError> {
sqlx::query("DELETE FROM targeted_accepted_offer_intents WHERE transfer_id = ?1")
.bind(id)
.execute(&self.pool)
.await
.map_err(VnidropError::repository)?;
Ok(())
}
pub(crate) async fn clear_accepted_intent_if_sender(
&self,
id: &str,
sender_endpoint_id: &str,
) -> Result<(), VnidropError> {
sqlx::query(
"DELETE FROM targeted_accepted_offer_intents WHERE transfer_id = ?1 AND sender_endpoint_id = ?2",
)
.bind(id)
.bind(sender_endpoint_id)
.execute(&self.pool)
.await
.map_err(VnidropError::repository)?;
Ok(())
}
pub(crate) async fn list_accepted_intent_senders(
&self,
) -> Result<Vec<(String, String)>, VnidropError> {
let rows = sqlx::query(
"SELECT transfer_id, sender_endpoint_id FROM targeted_accepted_offer_intents",
)
.fetch_all(&self.pool)
.await
.map_err(VnidropError::repository)?;
Ok(rows
.into_iter()
.map(|row| (row.get(0), row.get(1)))
.collect())
}
pub(crate) async fn contains_protocol_id(
&self,
protocol_transfer_id: u64,
@@ -251,6 +241,15 @@ impl TargetedTransferStore {
.map_err(VnidropError::repository)?
.get::<i64, _>(0)
!= 0;
if completed {
sqlx::query(
"DELETE FROM targeted_authorization_delivery_outbox WHERE transfer_id = ?1",
)
.bind(id)
.execute(&mut *transaction)
.await
.map_err(VnidropError::repository)?;
}
transaction
.commit()
.await
@@ -263,6 +262,11 @@ impl TargetedTransferStore {
})
};
}
sqlx::query("DELETE FROM targeted_authorization_delivery_outbox WHERE transfer_id = ?1")
.bind(id)
.execute(&mut *transaction)
.await
.map_err(VnidropError::repository)?;
sqlx::query(
"INSERT OR IGNORE INTO targeted_payload_release_outbox (transfer_id, created_at) VALUES (?1, ?2)",
)
@@ -278,189 +282,39 @@ impl TargetedTransferStore {
Ok(true)
}
pub(crate) async fn complete_receiver_and_enqueue(
&self,
id: &str,
verified_bytes: u64,
) -> Result<(), VnidropError> {
let mut transaction = self.pool.begin().await.map_err(VnidropError::repository)?;
let result = sqlx::query(
r#"
UPDATE targeted_transfers
SET state = 'completed', verified_bytes = ?2, updated_at = ?3
WHERE id = ?1 AND role = 'receiver' AND state = 'transferring'
"#,
)
.bind(id)
.bind(verified_bytes as i64)
.bind(now_ms())
.execute(&mut *transaction)
.await
.map_err(VnidropError::repository)?;
if result.rows_affected() == 0 {
return Err(VnidropError::InvalidTransition {
reason: "receiver transfer cannot be completed".to_string(),
});
}
sqlx::query(
"INSERT OR IGNORE INTO targeted_completion_outbox (transfer_id, created_at, next_attempt_at) VALUES (?1, ?2, ?2)",
)
.bind(id)
.bind(now_ms())
.execute(&mut *transaction)
.await
.map_err(VnidropError::repository)?;
transaction
.commit()
.await
.map_err(VnidropError::repository)?;
Ok(())
}
pub(crate) async fn 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> {
let rows = sqlx::query(
r#"
SELECT t.id, t.protocol_transfer_id, t.sender_endpoint_id, t.receiver_endpoint_id,
t.manifest_id, t.content_hash, t.transfer_name, t.file_count, t.total_size,
t.verified_bytes, t.blob_ticket, t.authorization_secret_handle, t.role,
t.state, t.created_at, t.updated_at
FROM targeted_transfers t
INNER JOIN targeted_completion_outbox o ON o.transfer_id = t.id
WHERE t.role = 'receiver' AND t.state = 'completed' AND o.next_attempt_at <= ?1
ORDER BY o.next_attempt_at, o.created_at
"#,
)
.bind(now_ms())
.fetch_all(&self.pool)
.await
.map_err(VnidropError::repository)?;
rows.into_iter().map(row_to_full).collect()
}
pub(crate) async fn clear_pending_completion(&self, id: &str) -> Result<(), VnidropError> {
sqlx::query("DELETE FROM targeted_completion_outbox WHERE transfer_id = ?1")
.bind(id)
.execute(&self.pool)
.await
.map_err(VnidropError::repository)?;
Ok(())
}
pub(crate) async fn defer_pending_completion(
&self,
id: &str,
next_attempt_at: i64,
) -> Result<(), VnidropError> {
sqlx::query(
"UPDATE targeted_completion_outbox SET next_attempt_at = ?2 WHERE transfer_id = ?1",
)
.bind(id)
.bind(next_attempt_at)
.execute(&self.pool)
.await
.map_err(VnidropError::repository)?;
Ok(())
}
pub(crate) async fn list_completed_sender_rows(
&self,
) -> Result<Vec<TargetedTransferRow>, VnidropError> {
let rows = sqlx::query(
r#"
SELECT t.id, t.protocol_transfer_id, t.sender_endpoint_id, t.receiver_endpoint_id,
t.manifest_id, t.content_hash, t.transfer_name, t.file_count, t.total_size,
t.verified_bytes, t.blob_ticket, t.authorization_secret_handle, t.role,
t.state, t.created_at, t.updated_at
FROM targeted_transfers t
INNER JOIN targeted_payload_release_outbox o ON o.transfer_id = t.id
WHERE t.role = 'sender' AND t.state = 'completed'
"#,
)
.fetch_all(&self.pool)
.await
.map_err(VnidropError::repository)?;
rows.into_iter().map(row_to_full).collect()
}
pub(crate) async fn clear_pending_payload_release(&self, id: &str) -> Result<(), VnidropError> {
sqlx::query("DELETE FROM targeted_payload_release_outbox WHERE transfer_id = ?1")
.bind(id)
.execute(&self.pool)
.await
.map_err(VnidropError::repository)?;
Ok(())
}
pub(crate) async fn store_authorization(
pub(crate) async fn finalize_sender_authorization_and_enqueue(
&self,
id: &str,
blob_ticket: &str,
authorization_secret_handle: &str,
) -> Result<(), VnidropError> {
sqlx::query(
let mut transaction = self.pool.begin().await.map_err(VnidropError::repository)?;
let now = now_ms();
let result = sqlx::query(
r#"
UPDATE targeted_transfers
SET blob_ticket = ?2,
authorization_secret_handle = ?3,
updated_at = ?4
WHERE id = ?1
SET state = 'approved', blob_ticket = ?2,
authorization_secret_handle = ?3, updated_at = ?4
WHERE id = ?1 AND role = 'sender' AND state = 'awaiting_approval'
"#,
)
.bind(id)
.bind(blob_ticket)
.bind(authorization_secret_handle)
.bind(now_ms())
.execute(&self.pool)
.bind(now)
.execute(&mut *transaction)
.await
.map_err(VnidropError::repository)?;
Ok(())
}
pub(crate) async fn clear_authorization(&self, id: &str) -> Result<(), VnidropError> {
let mut transaction = self.pool.begin().await.map_err(VnidropError::repository)?;
sqlx::query("DELETE FROM targeted_completion_outbox WHERE transfer_id = ?1")
.bind(id)
.execute(&mut *transaction)
.await
.map_err(VnidropError::repository)?;
sqlx::query("DELETE FROM targeted_payload_release_outbox WHERE transfer_id = ?1")
.bind(id)
.execute(&mut *transaction)
.await
.map_err(VnidropError::repository)?;
if result.rows_affected() == 0 {
return Err(VnidropError::InvalidTransition {
reason: "sender transfer cannot be approved".to_string(),
});
}
sqlx::query(
r#"
UPDATE targeted_transfers
SET blob_ticket = NULL,
authorization_secret_handle = NULL
WHERE id = ?1
"#,
"INSERT OR REPLACE INTO targeted_authorization_delivery_outbox (transfer_id, created_at, next_attempt_at) VALUES (?1, ?2, ?2)",
)
.bind(id)
.bind(now)
.execute(&mut *transaction)
.await
.map_err(VnidropError::repository)?;
@@ -471,56 +325,101 @@ impl TargetedTransferStore {
Ok(())
}
/// Commit a local terminal denial and optional secret cleanup atomically.
pub(crate) async fn transition_terminal(
pub(crate) async fn persist_receiver_authorization_and_consume_intent(
&self,
id: &str,
state: TargetedTransferState,
clear_authorization: bool,
auth: &crate::targeted_transfer::TargetedAuthorization,
authorization_secret_handle: &str,
) -> 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(
let matches = 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'))
)
SELECT EXISTS(
SELECT 1 FROM targeted_accepted_offer_intents
WHERE transfer_id = ?1 AND sender_endpoint_id = ?2 AND receiver_endpoint_id = ?3
AND manifest_id = ?4 AND content_hash = ?5 AND transfer_name = ?6
AND file_count = ?7 AND total_size = ?8 AND protocol_version = ?9
)
"#,
)
.bind(&auth.transfer_id)
.bind(&auth.sender_endpoint_id)
.bind(&auth.receiver_endpoint_id)
.bind(&auth.manifest_id)
.bind(&auth.content_hash)
.bind(&auth.transfer_name)
.bind(auth.file_count as i64)
.bind(auth.total_size as i64)
.bind(auth.protocol_version as i64)
.fetch_one(&mut *transaction)
.await
.map_err(VnidropError::repository)?
.get::<i64, _>(0)
!= 0;
if !matches {
return Err(VnidropError::permission(anyhow::anyhow!(
"authorization does not match durable receiver consent"
)));
}
let now = now_ms();
sqlx::query(
r#"
INSERT INTO targeted_transfers (
id, protocol_transfer_id, sender_endpoint_id, receiver_endpoint_id,
manifest_id, content_hash, transfer_name, file_count, total_size,
verified_bytes, blob_ticket, authorization_secret_handle, role,
state, created_at, updated_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, 0, ?10, ?11, 'receiver', 'approved', ?12, ?12)
"#,
)
.bind(&auth.transfer_id)
.bind(auth.protocol_transfer_id as i64)
.bind(&auth.sender_endpoint_id)
.bind(&auth.receiver_endpoint_id)
.bind(&auth.manifest_id)
.bind(&auth.content_hash)
.bind(&auth.transfer_name)
.bind(auth.file_count as i64)
.bind(auth.total_size as i64)
.bind(&auth.blob_ticket)
.bind(authorization_secret_handle)
.bind(now)
.execute(&mut *transaction)
.await
.map_err(VnidropError::repository)?;
sqlx::query("DELETE FROM targeted_accepted_offer_intents WHERE transfer_id = ?1")
.bind(&auth.transfer_id)
.execute(&mut *transaction)
.await
.map_err(VnidropError::repository)?;
transaction
.commit()
.await
.map_err(VnidropError::repository)?;
Ok(true)
}
pub(crate) async fn fail_resumable_and_clear_delivery(
&self,
id: &str,
) -> Result<bool, VnidropError> {
let mut transaction = self.pool.begin().await.map_err(VnidropError::repository)?;
let result = sqlx::query(
"UPDATE targeted_transfers SET state = 'failed', updated_at = ?2 WHERE id = ?1 AND state IN ('approved', 'connecting', 'transferring', 'interrupted')",
)
.bind(id)
.bind(state)
.bind(clear_authorization)
.bind(now_ms())
.execute(&mut *transaction)
.await
.map_err(VnidropError::repository)?;
if result.rows_affected() == 1 {
sqlx::query(
"DELETE FROM targeted_authorization_delivery_outbox WHERE transfer_id = ?1",
)
.bind(id)
.execute(&mut *transaction)
.await
.map_err(VnidropError::repository)?;
}
transaction
.commit()
.await
@@ -590,6 +489,7 @@ impl TargetedTransferStore {
FROM targeted_transfers
WHERE role = 'sender'
AND state IN ('approved', 'connecting', 'transferring', 'interrupted')
ORDER BY created_at, id
"#,
)
.fetch_all(&self.pool)
@@ -598,15 +498,62 @@ impl TargetedTransferStore {
rows.into_iter().map(row_to_full).collect()
}
pub(crate) async fn list_resumable_rows(
&self,
) -> Result<Vec<TargetedTransferRow>, VnidropError> {
let rows = sqlx::query(
r#"
SELECT id, protocol_transfer_id, sender_endpoint_id, receiver_endpoint_id,
manifest_id, content_hash, transfer_name, file_count, total_size,
verified_bytes, blob_ticket, authorization_secret_handle, role,
state, created_at, updated_at
FROM targeted_transfers
WHERE state IN ('approved', 'connecting', 'transferring', 'interrupted')
ORDER BY created_at, id
"#,
)
.fetch_all(&self.pool)
.await
.map_err(VnidropError::repository)?;
rows.into_iter().map(row_to_full).collect()
}
#[cfg(test)]
pub(crate) async fn corrupt_content_hash_for_test(&self, id: &str) -> Result<(), VnidropError> {
sqlx::query("UPDATE targeted_transfers SET content_hash = ?2 WHERE id = ?1")
.bind(id)
.bind(iroh_blobs::Hash::new([0xff; 32]).to_string())
.execute(&self.pool)
.await
.map_err(VnidropError::repository)?;
Ok(())
}
pub(crate) async fn cancel_by_peer(
&self,
peer_endpoint_id: &str,
) -> Result<Vec<String>, VnidropError> {
let now = now_ms();
let mut transaction = self.pool.begin().await.map_err(VnidropError::repository)?;
sqlx::query(
"DELETE FROM targeted_accepted_offer_intents WHERE sender_endpoint_id = ?1 OR receiver_endpoint_id = ?1",
)
.bind(peer_endpoint_id)
.execute(&mut *transaction)
.await
.map_err(VnidropError::repository)?;
sqlx::query(
"DELETE FROM targeted_authorization_delivery_outbox WHERE transfer_id IN (SELECT id FROM targeted_transfers WHERE sender_endpoint_id = ?1 OR receiver_endpoint_id = ?1)",
)
.bind(peer_endpoint_id)
.execute(&mut *transaction)
.await
.map_err(VnidropError::repository)?;
let rows = sqlx::query(
r#"
UPDATE targeted_transfers
SET state = 'cancelled', updated_at = ?2
SET state = 'cancelled', blob_ticket = NULL,
authorization_secret_handle = NULL, updated_at = ?2
WHERE (sender_endpoint_id = ?1 OR receiver_endpoint_id = ?1)
AND state NOT IN ('completed', 'declined', 'cancelled', 'failed', 'deleted')
RETURNING id
@@ -614,9 +561,13 @@ impl TargetedTransferStore {
)
.bind(peer_endpoint_id)
.bind(now)
.fetch_all(&self.pool)
.fetch_all(&mut *transaction)
.await
.map_err(VnidropError::repository)?;
transaction
.commit()
.await
.map_err(VnidropError::repository)?;
Ok(rows.into_iter().map(|row| row.get("id")).collect())
}
@@ -683,6 +634,46 @@ impl TargetedTransferStore {
.collect())
}
pub(crate) async fn authorizations_for_peer(
&self,
peer_endpoint_id: &str,
) -> Result<Vec<(String, String)>, VnidropError> {
let rows = sqlx::query(
r#"
SELECT id, authorization_secret_handle FROM targeted_transfers
WHERE (sender_endpoint_id = ?1 OR receiver_endpoint_id = ?1)
AND authorization_secret_handle IS NOT NULL
"#,
)
.bind(peer_endpoint_id)
.fetch_all(&self.pool)
.await
.map_err(VnidropError::repository)?;
Ok(rows
.into_iter()
.map(|row| (row.get(0), row.get(1)))
.collect())
}
pub(crate) async fn authorization_rows(
&self,
) -> Result<Vec<TargetedTransferRow>, VnidropError> {
let rows = sqlx::query(
r#"
SELECT id, protocol_transfer_id, sender_endpoint_id, receiver_endpoint_id,
manifest_id, content_hash, transfer_name, file_count, total_size,
verified_bytes, blob_ticket, authorization_secret_handle, role,
state, created_at, updated_at
FROM targeted_transfers
WHERE authorization_secret_handle IS NOT NULL
"#,
)
.fetch_all(&self.pool)
.await
.map_err(VnidropError::repository)?;
rows.into_iter().map(row_to_full).collect()
}
pub(crate) async fn mark_interrupted_in_flight(&self) -> Result<Vec<String>, VnidropError> {
let now = now_ms();
let rows = sqlx::query(
@@ -743,7 +734,9 @@ fn row_to_transfer(row: sqlx::sqlite::SqliteRow) -> Result<TargetedTransfer, Vni
})
}
fn row_to_full(row: sqlx::sqlite::SqliteRow) -> Result<TargetedTransferRow, VnidropError> {
pub(super) fn row_to_full(
row: sqlx::sqlite::SqliteRow,
) -> Result<TargetedTransferRow, VnidropError> {
Ok(TargetedTransferRow {
id: row.get("id"),
protocol_transfer_id: row.get::<i64, _>("protocol_transfer_id") as u64,

311
crates/vnidrop/src/targeted_transfer/store_outbox.rs Normal file
View File

@@ -0,0 +1,311 @@
//! Durable authorization, completion, release, and terminal transactions.
use sqlx::Row;
use super::store::{row_to_full, TargetedTransferStore};
use super::TargetedTransferRow;
use crate::{api::TargetedTransferState, error::VnidropError, util::now_ms};
impl TargetedTransferStore {
pub(crate) async fn list_pending_authorization_deliveries(
&self,
) -> Result<Vec<TargetedTransferRow>, VnidropError> {
let rows = sqlx::query(
r#"
SELECT t.id, t.protocol_transfer_id, t.sender_endpoint_id, t.receiver_endpoint_id,
t.manifest_id, t.content_hash, t.transfer_name, t.file_count, t.total_size,
t.verified_bytes, t.blob_ticket, t.authorization_secret_handle, t.role,
t.state, t.created_at, t.updated_at
FROM targeted_transfers t
INNER JOIN targeted_authorization_delivery_outbox o ON o.transfer_id = t.id
WHERE t.role = 'sender' AND t.state = 'approved' AND o.next_attempt_at <= ?1
ORDER BY o.next_attempt_at, o.created_at
"#,
)
.bind(now_ms())
.fetch_all(&self.pool)
.await
.map_err(VnidropError::repository)?;
rows.into_iter().map(row_to_full).collect()
}
pub(crate) async fn clear_pending_authorization_delivery(
&self,
id: &str,
) -> Result<(), VnidropError> {
sqlx::query("DELETE FROM targeted_authorization_delivery_outbox WHERE transfer_id = ?1")
.bind(id)
.execute(&self.pool)
.await
.map_err(VnidropError::repository)?;
Ok(())
}
pub(crate) async fn defer_pending_authorization_delivery(
&self,
id: &str,
next_attempt_at: i64,
) -> Result<(), VnidropError> {
sqlx::query(
"UPDATE targeted_authorization_delivery_outbox SET next_attempt_at = ?2 WHERE transfer_id = ?1",
)
.bind(id)
.bind(next_attempt_at)
.execute(&self.pool)
.await
.map_err(VnidropError::repository)?;
Ok(())
}
pub(crate) async fn complete_receiver_and_enqueue(
&self,
id: &str,
verified_bytes: u64,
) -> Result<(), VnidropError> {
let mut transaction = self.pool.begin().await.map_err(VnidropError::repository)?;
let result = sqlx::query(
r#"
UPDATE targeted_transfers
SET state = 'completed', verified_bytes = ?2, updated_at = ?3
WHERE id = ?1 AND role = 'receiver' AND state = 'transferring'
"#,
)
.bind(id)
.bind(verified_bytes as i64)
.bind(now_ms())
.execute(&mut *transaction)
.await
.map_err(VnidropError::repository)?;
if result.rows_affected() == 0 {
return Err(VnidropError::InvalidTransition {
reason: "receiver transfer cannot be completed".to_string(),
});
}
sqlx::query(
"INSERT OR IGNORE INTO targeted_completion_outbox (transfer_id, created_at, next_attempt_at) VALUES (?1, ?2, ?2)",
)
.bind(id)
.bind(now_ms())
.execute(&mut *transaction)
.await
.map_err(VnidropError::repository)?;
transaction
.commit()
.await
.map_err(VnidropError::repository)?;
Ok(())
}
pub(crate) async fn 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> {
let rows = sqlx::query(
r#"
SELECT t.id, t.protocol_transfer_id, t.sender_endpoint_id, t.receiver_endpoint_id,
t.manifest_id, t.content_hash, t.transfer_name, t.file_count, t.total_size,
t.verified_bytes, t.blob_ticket, t.authorization_secret_handle, t.role,
t.state, t.created_at, t.updated_at
FROM targeted_transfers t
INNER JOIN targeted_completion_outbox o ON o.transfer_id = t.id
WHERE t.role = 'receiver' AND t.state = 'completed' AND o.next_attempt_at <= ?1
ORDER BY o.next_attempt_at, o.created_at
"#,
)
.bind(now_ms())
.fetch_all(&self.pool)
.await
.map_err(VnidropError::repository)?;
rows.into_iter().map(row_to_full).collect()
}
pub(crate) async fn clear_pending_completion(&self, id: &str) -> Result<(), VnidropError> {
sqlx::query("DELETE FROM targeted_completion_outbox WHERE transfer_id = ?1")
.bind(id)
.execute(&self.pool)
.await
.map_err(VnidropError::repository)?;
Ok(())
}
pub(crate) async fn defer_pending_completion(
&self,
id: &str,
next_attempt_at: i64,
) -> Result<(), VnidropError> {
sqlx::query(
"UPDATE targeted_completion_outbox SET next_attempt_at = ?2 WHERE transfer_id = ?1",
)
.bind(id)
.bind(next_attempt_at)
.execute(&self.pool)
.await
.map_err(VnidropError::repository)?;
Ok(())
}
pub(crate) async fn list_completed_sender_rows(
&self,
) -> Result<Vec<TargetedTransferRow>, VnidropError> {
let rows = sqlx::query(
r#"
SELECT t.id, t.protocol_transfer_id, t.sender_endpoint_id, t.receiver_endpoint_id,
t.manifest_id, t.content_hash, t.transfer_name, t.file_count, t.total_size,
t.verified_bytes, t.blob_ticket, t.authorization_secret_handle, t.role,
t.state, t.created_at, t.updated_at
FROM targeted_transfers t
INNER JOIN targeted_payload_release_outbox o ON o.transfer_id = t.id
WHERE t.role = 'sender' AND t.state = 'completed'
"#,
)
.fetch_all(&self.pool)
.await
.map_err(VnidropError::repository)?;
rows.into_iter().map(row_to_full).collect()
}
pub(crate) async fn clear_pending_payload_release(&self, id: &str) -> Result<(), VnidropError> {
sqlx::query("DELETE FROM targeted_payload_release_outbox WHERE transfer_id = ?1")
.bind(id)
.execute(&self.pool)
.await
.map_err(VnidropError::repository)?;
Ok(())
}
pub(crate) async fn authorization_secret_handles(
&self,
) -> Result<std::collections::HashSet<String>, VnidropError> {
let rows = sqlx::query(
"SELECT authorization_secret_handle FROM targeted_transfers WHERE authorization_secret_handle IS NOT NULL",
)
.fetch_all(&self.pool)
.await
.map_err(VnidropError::repository)?;
Ok(rows.into_iter().map(|row| row.get(0)).collect())
}
pub(crate) async fn clear_authorization(&self, id: &str) -> Result<(), VnidropError> {
let mut transaction = self.pool.begin().await.map_err(VnidropError::repository)?;
sqlx::query("DELETE FROM targeted_authorization_delivery_outbox WHERE transfer_id = ?1")
.bind(id)
.execute(&mut *transaction)
.await
.map_err(VnidropError::repository)?;
sqlx::query("DELETE FROM targeted_completion_outbox WHERE transfer_id = ?1")
.bind(id)
.execute(&mut *transaction)
.await
.map_err(VnidropError::repository)?;
sqlx::query("DELETE FROM targeted_payload_release_outbox WHERE transfer_id = ?1")
.bind(id)
.execute(&mut *transaction)
.await
.map_err(VnidropError::repository)?;
sqlx::query(
r#"
UPDATE targeted_transfers
SET blob_ticket = NULL,
authorization_secret_handle = NULL
WHERE id = ?1
"#,
)
.bind(id)
.execute(&mut *transaction)
.await
.map_err(VnidropError::repository)?;
transaction
.commit()
.await
.map_err(VnidropError::repository)?;
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)?;
sqlx::query("DELETE FROM targeted_accepted_offer_intents WHERE transfer_id = ?1")
.bind(id)
.execute(&mut *transaction)
.await
.map_err(VnidropError::repository)?;
sqlx::query("DELETE FROM targeted_authorization_delivery_outbox WHERE transfer_id = ?1")
.bind(id)
.execute(&mut *transaction)
.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)
}
}

65
crates/vnidrop/src/tests/persistence.rs
View File

@@ -35,6 +35,9 @@ async fn open_all_returns_all_domain_stores_and_schemas() {
"protected_secret_refs",
"blocked_endpoints",
"targeted_transfers",
"targeted_accepted_offer_intents",
"targeted_authorization_delivery_outbox",
"targeted_schema_migrations",
"targeted_completion_outbox",
"targeted_payload_release_outbox",
"transfers",
@@ -53,6 +56,68 @@ async fn open_all_returns_all_domain_stores_and_schemas() {
}
}
#[tokio::test]
async fn approved_sender_rows_backfill_authorization_delivery_once() {
let temp = tempfile::tempdir().unwrap();
let stores = persistence::open_all(temp.path()).await.unwrap();
let now = crate::util::now_ms();
stores
.targeted
.insert(&crate::targeted_transfer::TargetedTransferRow {
id: "legacy-approved".to_string(),
protocol_transfer_id: 991,
sender_endpoint_id: "sender".to_string(),
receiver_endpoint_id: "receiver".to_string(),
manifest_id: "manifest".to_string(),
content_hash: "content".to_string(),
transfer_name: "legacy".to_string(),
file_count: 1,
total_size: 4,
verified_bytes: 0,
blob_ticket: Some("ticket".to_string()),
authorization_secret_handle: Some("handle".to_string()),
role: crate::targeted_transfer::TargetedTransferRole::Sender,
state: crate::TargetedTransferState::Approved,
created_at: now,
updated_at: now,
})
.await
.unwrap();
let pool = open_profile_pool(temp.path()).await;
sqlx::query(
"DELETE FROM targeted_schema_migrations WHERE name = 'authorization-delivery-outbox-v1'",
)
.execute(&pool)
.await
.unwrap();
drop(pool);
drop(stores);
let stores = persistence::open_all(temp.path()).await.unwrap();
assert_eq!(
stores
.targeted
.list_pending_authorization_deliveries()
.await
.unwrap()
.len(),
1
);
stores
.targeted
.clear_pending_authorization_delivery("legacy-approved")
.await
.unwrap();
drop(stores);
let stores = persistence::open_all(temp.path()).await.unwrap();
assert!(stores
.targeted
.list_pending_authorization_deliveries()
.await
.unwrap()
.is_empty());
}
#[tokio::test]
async fn open_all_migrates_targeted_completion_retry_schedule() {
let temp = tempfile::tempdir().unwrap();

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

@@ -794,6 +794,374 @@ fn approve_one(
transfer
}
#[test]
fn approved_authorization_delivery_retries_after_sender_restart() {
let alice = ProtectedNode::new();
let bob = ProtectedNode::new();
establish_saved(&alice, &bob, 11_000);
alice
.core()
.suppress_targeted_authorization_delivery_for_test(true);
let source_dir = tempfile::tempdir().unwrap();
let source_path = source_dir.path().join("payload.txt");
std::fs::write(&source_path, b"durable authorization").unwrap();
let bob_id = bob.core().status().endpoint_id.clone();
let alice_core = alice.core();
let create = std::thread::spawn(move || {
alice_core.create_targeted_transfer(
bob_id,
vec![targeted_source(&source_path)],
Some("payload.txt".to_string()),
)
});
let offer = wait_for_pending_offer(&bob.core());
let bob_core = bob.core();
let offer_id = offer.transfer_id.clone();
let accept = std::thread::spawn(move || bob_core.respond_to_targeted_offer(offer_id, true));
let transfer = create
.join()
.unwrap()
.expect("sender remains approved while delivery is pending");
assert_eq!(transfer.state, TargetedTransferState::Approved);
assert!(bob.core().list_targeted_transfers().unwrap().is_empty());
let alice = alice.restart();
let response = accept
.join()
.unwrap()
.expect("sender restart redelivers durable authorization");
assert!(matches!(
response,
crate::TargetedOfferResponse::Approved { transfer_id } if transfer_id == transfer.id
));
assert_eq!(
bob.core()
.get_targeted_transfer(transfer.id.clone())
.unwrap()
.unwrap()
.state,
TargetedTransferState::Approved
);
assert_eq!(
alice
.core()
.get_targeted_transfer(transfer.id)
.unwrap()
.unwrap()
.state,
TargetedTransferState::Approved
);
}
#[test]
fn accepted_intent_survives_receiver_and_sender_restart_until_delivery() {
let alice = ProtectedNode::new();
let bob = ProtectedNode::new();
establish_saved(&alice, &bob, 110_000);
alice
.core()
.suppress_targeted_authorization_delivery_for_test(true);
let source_dir = tempfile::tempdir().unwrap();
let source_path = source_dir.path().join("payload.txt");
std::fs::write(&source_path, b"restart consent").unwrap();
let bob_id = bob.core().status().endpoint_id;
let alice_core = alice.core();
let create = std::thread::spawn(move || {
alice_core.create_targeted_transfer(
bob_id,
vec![targeted_source(&source_path)],
Some("payload.txt".to_string()),
)
});
let offer = wait_for_pending_offer(&bob.core());
bob.core()
.accept_targeted_offer_without_waiting_for_test(offer.transfer_id)
.unwrap();
let transfer = create.join().unwrap().unwrap();
let bob = bob.restart();
let alice = alice.restart();
let started = Instant::now();
loop {
if bob
.core()
.get_targeted_transfer(transfer.id.clone())
.unwrap()
.is_some_and(|row| row.state == TargetedTransferState::Approved)
{
break;
}
assert!(started.elapsed() < Duration::from_secs(15));
std::thread::sleep(Duration::from_millis(50));
}
assert_eq!(
alice
.core()
.get_targeted_transfer(transfer.id)
.unwrap()
.unwrap()
.state,
TargetedTransferState::Approved
);
}
#[test]
fn restart_reconciles_targeted_authorization_orphaned_before_domain_commit() {
let node = ProtectedNode::new();
node.core()
.create_orphaned_targeted_authorization_for_test()
.unwrap();
assert_eq!(
node.core()
.targeted_authorization_handle_count_for_test()
.unwrap(),
1
);
let node = node.restart();
assert_eq!(
node.core()
.targeted_authorization_handle_count_for_test()
.unwrap(),
0
);
}
#[test]
fn terminal_receiver_secret_cleanup_retries_while_relationship_is_saved() {
let alice = ProtectedNode::new();
let bob = ProtectedNode::new();
establish_saved(&alice, &bob, 110_005);
let transfer = approve_one(&alice, &bob, b"cleanup boundary", "payload.txt");
assert_eq!(
bob.core()
.targeted_authorization_handle_count_for_test()
.unwrap(),
1
);
bob.secret_store
.fail_with(Some(ReferenceStoreFailure::Unavailable));
assert!(bob
.core()
.cancel_targeted_transfer(transfer.id.clone())
.is_err());
assert_eq!(
bob.core()
.get_targeted_transfer(transfer.id.clone())
.unwrap()
.unwrap()
.state,
TargetedTransferState::Cancelled
);
bob.secret_store.fail_with(None);
let bob = bob.restart();
assert_eq!(
bob.core()
.get_targeted_transfer(transfer.id)
.unwrap()
.unwrap()
.state,
TargetedTransferState::Cancelled
);
assert_eq!(
bob.core()
.targeted_authorization_handle_count_for_test()
.unwrap(),
0
);
}
#[test]
fn forgetting_saved_receiver_keeps_sender_denied_when_secret_delete_retries() {
let alice = ProtectedNode::new();
let bob = ProtectedNode::new();
establish_saved(&alice, &bob, 110_006);
let transfer = approve_one(&alice, &bob, b"sender cleanup boundary", "payload.txt");
assert_eq!(
alice
.core()
.targeted_authorization_handle_count_for_test()
.unwrap(),
1
);
let protected_values_before_forget = alice.secret_store.stored_value_count_for_test();
alice
.secret_store
.fail_with(Some(ReferenceStoreFailure::Unavailable));
assert!(alice
.core()
.forget_saved_device(bob.core().status().endpoint_id)
.is_err());
assert_eq!(
alice
.core()
.get_targeted_transfer(transfer.id.clone())
.unwrap()
.unwrap()
.state,
TargetedTransferState::Cancelled
);
assert!(alice
.core()
.targeted_blob_ticket_for_test(transfer.id.clone())
.is_err());
assert_eq!(
alice.secret_store.stored_value_count_for_test(),
protected_values_before_forget,
"failed secure deletion must leave retryable protected material"
);
alice.secret_store.fail_with(None);
let alice = alice.restart();
assert_eq!(
alice
.core()
.get_targeted_transfer(transfer.id.clone())
.unwrap()
.unwrap()
.state,
TargetedTransferState::Cancelled
);
assert!(alice
.core()
.targeted_blob_ticket_for_test(transfer.id)
.is_err());
assert_eq!(
alice
.core()
.targeted_authorization_handle_count_for_test()
.unwrap(),
0
);
assert!(
alice.secret_store.stored_value_count_for_test() < protected_values_before_forget,
"restart reconciliation must delete orphaned protected material"
);
}
#[test]
fn cancel_after_accepted_receiver_restart_revokes_durable_consent() {
let alice = ProtectedNode::new();
let bob = ProtectedNode::new();
establish_saved(&alice, &bob, 110_003);
alice
.core()
.suppress_targeted_authorization_delivery_for_test(true);
let source_dir = tempfile::tempdir().unwrap();
let source_path = source_dir.path().join("payload.txt");
std::fs::write(&source_path, b"revoked consent").unwrap();
let bob_id = bob.core().status().endpoint_id;
let alice_core = alice.core();
let create = std::thread::spawn(move || {
alice_core.create_targeted_transfer(
bob_id,
vec![targeted_source(&source_path)],
Some("payload.txt".to_string()),
)
});
let offer = wait_for_pending_offer(&bob.core());
bob.core()
.accept_targeted_offer_without_waiting_for_test(offer.transfer_id.clone())
.unwrap();
let transfer = create.join().unwrap().unwrap();
let bob = bob.restart();
bob.core()
.cancel_targeted_transfer(offer.transfer_id)
.unwrap();
let alice = alice.restart();
let started = Instant::now();
while alice
.core()
.targeted_authorization_delivery_attempts_for_test()
== 0
{
assert!(started.elapsed() < Duration::from_secs(10));
std::thread::sleep(Duration::from_millis(25));
}
assert!(bob.core().list_targeted_transfers().unwrap().is_empty());
assert_ne!(
alice
.core()
.get_targeted_transfer(transfer.id)
.unwrap()
.unwrap()
.state,
TargetedTransferState::Completed
);
}
#[test]
fn restart_never_restores_targeted_access_for_a_persisted_block() {
let alice = ProtectedNode::new();
let bob = ProtectedNode::new();
establish_saved(&alice, &bob, 110_001);
let transfer = approve_one(&alice, &bob, b"blocked after crash", "payload.txt");
let bob_id = bob.core().status().endpoint_id;
alice
.core()
.persist_block_without_cleanup_for_test(bob_id)
.unwrap();
let alice = alice.restart();
assert_eq!(
alice
.core()
.get_targeted_transfer(transfer.id.clone())
.unwrap()
.unwrap()
.state,
TargetedTransferState::Cancelled
);
assert!(!alice
.core()
.targeted_payload_is_registered_for_test(transfer.id.clone())
.unwrap());
let output = tempfile::tempdir().unwrap();
assert!(bob
.core()
.receive_targeted_transfer(transfer.id, output.path().to_string_lossy().into_owned(),)
.is_err());
assert!(std::fs::read_dir(output.path()).unwrap().next().is_none());
}
#[test]
fn corrupt_restored_target_does_not_strand_later_valid_target() {
let alice = ProtectedNode::new();
let bob = ProtectedNode::new();
establish_saved(&alice, &bob, 110_002);
let corrupt = approve_one(&alice, &bob, b"corrupt", "corrupt.txt");
let valid = approve_one(&alice, &bob, b"valid", "valid.txt");
alice
.core()
.corrupt_targeted_content_hash_for_test(corrupt.id.clone())
.unwrap();
let alice = alice.restart();
assert_eq!(
alice
.core()
.get_targeted_transfer(corrupt.id)
.unwrap()
.unwrap()
.state,
TargetedTransferState::Failed
);
assert_eq!(
alice
.core()
.get_targeted_transfer(valid.id.clone())
.unwrap()
.unwrap()
.state,
TargetedTransferState::Approved
);
assert!(alice
.core()
.targeted_payload_is_registered_for_test(valid.id)
.unwrap());
}
#[test]
fn protocol_ops_are_idempotent_for_stable_transfer_id() {
let alice = ProtectedNode::new();
@@ -823,6 +1191,33 @@ fn protocol_ops_are_idempotent_for_stable_transfer_id() {
assert_eq!(listed, 1, "replay must not create duplicate durable rows");
}
#[test]
fn authorization_replay_after_receiver_commit_and_restart_is_stored() {
let alice = ProtectedNode::new();
let bob = ProtectedNode::new();
establish_saved(&alice, &bob, 110_004);
let transfer = approve_one(&alice, &bob, b"lost stored response", "payload.txt");
let bob = bob.restart();
assert!(alice
.core()
.redeliver_targeted_authorization_for_test(transfer.id.clone())
.unwrap());
let receiver = bob
.core()
.get_targeted_transfer(transfer.id)
.unwrap()
.unwrap();
assert_eq!(receiver.state, TargetedTransferState::Approved);
assert_eq!(
bob.core()
.targeted_authorization_handle_count_for_test()
.unwrap(),
1,
"idempotent replay must not create another protected secret"
);
}
#[test]
fn unapproved_offers_vanish_on_cancel_and_core_restart() {
let alice = ProtectedNode::new();
@@ -1081,11 +1476,26 @@ fn cancel_revokes_access_and_stops_streaming() {
let bob = ProtectedNode::new();
establish_saved(&alice, &bob, 11_040);
let transfer = approve_one(&alice, &bob, b"cancel me", "payload.txt");
assert_eq!(
alice
.core()
.targeted_authorization_handle_count_for_test()
.unwrap(),
1
);
alice
.core()
.cancel_targeted_transfer(transfer.id.clone())
.unwrap();
assert_eq!(
alice
.core()
.targeted_authorization_handle_count_for_test()
.unwrap(),
0,
"sender cancellation must clean protected authorization custody"
);
let cancelled = alice
.core()
.get_targeted_transfer(transfer.id.clone())
@@ -1855,6 +2265,61 @@ fn targeted_receive_rejects_a_concurrent_second_pull() {
receive.join().unwrap().unwrap();
}
#[test]
fn targeted_cancel_while_waiting_for_transfer_slot_never_publishes() {
let alice = ProtectedNode::new();
let bob = ProtectedNode::new();
establish_saved(&alice, &bob, 11_077);
let transfer = approve_one(&alice, &bob, b"slot-starved payload", "queued.txt");
let release_slots = bob.core().hold_all_transfer_slots_for_test();
let output = tempfile::tempdir().unwrap();
let output_path = output.path().to_string_lossy().into_owned();
let bob_core = bob.core();
let transfer_id = transfer.id.clone();
let (finished_tx, finished_rx) = std::sync::mpsc::sync_channel(1);
let receive = std::thread::spawn(move || {
let result = bob_core.receive_targeted_transfer(transfer_id, output_path);
finished_tx.send(result).unwrap();
});
let started = Instant::now();
loop {
let state = bob
.core()
.get_targeted_transfer(transfer.id.clone())
.unwrap()
.unwrap()
.state;
if state == TargetedTransferState::Transferring {
break;
}
assert!(
started.elapsed() < Duration::from_secs(10),
"targeted receive never queued behind transfer limiter"
);
std::thread::sleep(Duration::from_millis(10));
}
bob.core()
.cancel_targeted_transfer(transfer.id.clone())
.unwrap();
let receive_result = finished_rx
.recv_timeout(Duration::from_secs(2))
.expect("queued targeted receive must observe cancel before a slot is released");
assert!(receive_result.is_err());
let _ = release_slots.send(());
receive.join().unwrap();
assert!(!output.path().join("queued.txt").exists());
assert_eq!(
bob.core()
.get_targeted_transfer(transfer.id)
.unwrap()
.unwrap()
.state,
TargetedTransferState::Cancelled
);
}
#[test]
fn targeted_cancel_aborts_each_sink_exactly_once() {
for use_v2 in [false, true] {
@@ -1981,7 +2446,7 @@ fn decline_returns_typed_declined_outcome() {
}
#[test]
fn approval_secret_failure_records_failed_receiver_snapshot_and_wakeup() {
fn approval_secret_failure_keeps_durable_consent_retryable() {
let alice = ProtectedNode::new();
let bob = ProtectedNode::new();
establish_saved(&alice, &bob, 11_081);
@@ -2000,24 +2465,37 @@ fn approval_secret_failure_records_failed_receiver_snapshot_and_wakeup() {
let offer = wait_for_pending_offer(&bob.core());
bob.secret_store
.fail_with(Some(ReferenceStoreFailure::Unavailable));
assert!(bob
bob.core()
.accept_targeted_offer_without_waiting_for_test(offer.transfer_id.clone())
.unwrap();
let transfer = create.join().unwrap().unwrap();
let started = Instant::now();
while alice
.core()
.respond_to_targeted_offer(offer.transfer_id.clone(), true)
.is_err());
bob.secret_store.fail_with(None);
let failed = bob
.targeted_authorization_delivery_attempts_for_test()
== 0
{
assert!(started.elapsed() < Duration::from_secs(10));
std::thread::sleep(Duration::from_millis(25));
}
assert!(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());
.is_none());
bob.secret_store.fail_with(None);
let started = Instant::now();
loop {
if bob
.core()
.get_targeted_transfer(offer.transfer_id.clone())
.unwrap()
.is_some_and(|row| row.state == TargetedTransferState::Approved)
{
break;
}
assert!(started.elapsed() < Duration::from_secs(15));
std::thread::sleep(Duration::from_millis(50));
}
assert_eq!(transfer.state, TargetedTransferState::Approved);
}